线程的三个同步器
- 2020 年 4 月 12 日
- 笔记
不知不觉就遇到了线程同步器问题，查了资料写下了总结
1. CountDownLatch
日常中会有开启多个线程去并发执行任务，而主线程要等所有子线程执行完之后才能运行的需求。之前我们是使用Thread.join方法来实现的，过程如下：
public static void main(String[] args) throws InterruptedException { Thread t1 = new Thread( () -> { try { Thread.sleep(1000); System.out.println("t1 over"); } catch (InterruptedException e) { e.printStackTrace(); } }); Thread t2 = new Thread( () -> { try { Thread.sleep(2000); System.out.println("t2 over"); } catch (InterruptedException e) { e.printStackTrace(); } }); t1.start(); t2.start(); t1.join(); t2.join(); System.out.println("mian over"); } t1 over t2 over mian over join()方法不够灵活，现在JDK提供了CountDownLatch这个类来实现所需功能
private static CountDownLatch countDownLatch = new CountDownLatch(2); public static void main(String[] args) throws InterruptedException { ExecutorService t = Executors.newCachedThreadPool(); Runnable r1 = () -> { try { System.out.println("r1 sleep"); Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } finally { countDownLatch.countDown(); } }; Runnable r2 = () -> { try { System.out.println("r2 sleep"); Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } finally { countDownLatch.countDown(); } }; t.submit(r1); t.submit(r2); System.out.println("main wait"); countDownLatch.await(); System.out.println("main over"); } main wait r1 sleep r2 sleep main over CountDownLatch流程：
- 新建CountDownLatch实例，传入计数器次数
- 主线程调用CountDownLatch.await()方法后会被阻塞
- 子线程中在某处调用CountDownLatch.countDown()方法可使内部计数器减1
- 当计数器变成0时，主线程的await()方法才会返回
CountDownLatch优点：
- 调用Thread.join()调用线程会被阻塞至子线程运行完毕，而CountDownLatch.countDown()可在线程运行中执行
- 使用线程池时是提交任务的，而没有接触到线程无法使用线程方法，那么countDown()可加在Runnable中执行
CountDownLatch原理：
内部维护了一个计数器，当计数器为0就放行，源码就不放了，熟悉AQS的同学想想就知道怎么回事
- 继承了AQS，其实就是用AQS的state来表示计数器
- await()方法内部有acquireSharedInterruptibly()，后者调用了重写tryaquireShared()其实就是判断计数器是否为0，不为0则阻塞进AQS队列
- countDown()方法内部有releaseShared()，后者调用了重写tryReleaseShared()计数器减一，若为0，则唤醒阻塞线程
2. CyclicBarrier
满足多个线程都到达同一个位置后才全部开始运行的需求。CountDownLatch是一次性使用的，计数器为0后再次调用会直接返回，此时升级版的CyclicBarrier来了，其一可以满足计数器重置功能，且二还可以让一组线程达到一个状态后再全部同时执行
场景要求：假设一个任务分为3个阶段，每个线程要串行地从低阶段执行到高阶段
private static CyclicBarrier cyclicBarrier = new CyclicBarrier(2, () -> System.out.println("一个阶段完成")); public static void main(String[] args) throws InterruptedException { ExecutorService service = Executors.newCachedThreadPool(); Runnable r1 = () -> { try { System.out.println(Thread.currentThread() + "Step1"); cyclicBarrier.await(); System.out.println(Thread.currentThread() + "Step2"); cyclicBarrier.await(); System.out.println(Thread.currentThread() + "Step3"); } catch (Exception e) { e.printStackTrace(); } }; Runnable r2 = () -> { try { System.out.println(Thread.currentThread() + "Step1"); cyclicBarrier.await(); System.out.println(Thread.currentThread() + "Step2"); cyclicBarrier.await(); System.out.println(Thread.currentThread() + "Step3"); } catch (Exception e) { e.printStackTrace(); } }; service.submit(r1); service.submit(r2); service.shutdown(); } Thread[pool-1-thread-1,5,main]Step1 Thread[pool-1-thread-2,5,main]Step1 一个阶段完成 Thread[pool-1-thread-1,5,main]Step2 Thread[pool-1-thread-2,5,main]Step2 一个阶段完成 Thread[pool-1-thread-1,5,main]Step3 Thread[pool-1-thread-2,5,main]Step3 CyclicBarrier的流程
- 和上面差不多就不一一解释了
- CyclicBarrier的构造方法中，第一个参数为计数器次数，第二个为阶段结束后要执行的方法
CyclicBarrier的原理
- 基于独占锁，底层是AQS实现，独占锁可以原子性改变计数器，以及条件队列阻塞线程来实现线程同步
- 内部有parties和count变量，实现重置功能
- await()方法内调用dowait()方法
- 获取锁更新次数减一
- 没有为0，阻塞当前线程加入条件队列
- 为0执行屏蔽点任务，然后唤醒条件队列的全部线程
3. Semaphore
不同与前两者，Semaphore信号量内部计数器是递增的，在需要同步的地方调用acquire指定需要同步的个数即可
private static Semaphore semaphore = new Semaphore(0); public static void main(String[] args) throws InterruptedException { ExecutorService service = Executors.newCachedThreadPool(); Runnable r1 = () -> { try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread() + "over"); semaphore.release();; }; Runnable r2 = () -> { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread() + "over"); semaphore.release(); }; service.submit(r1); service.submit(r2); semaphore.acquire(2); System.out.println("All child thread over"); service.shutdown(); } Semaphore的流程
- Semaphore的构造函数传参复制当前计数器的值
- 每个线程内部调用release()即计数器加1
- 主线程调用acquire()方法传参为2 ，会被阻塞至计数器到达2
Semaphore的原理
- 底层还是使用AQS，提供了公平与非公平，也是用state表示次数
- acquire()方法获取一个信号量，并且state减一
- 若为0，直接返回
- 不为0当前线程会被加入AQS阻塞队列
- release()方法，把当前Semaphore的信号量加1，然后会选择一个信号量满足的线程进行激活
- 内部还实现了公平与非公平策略
