--- title: Java并发编程通信工具类CountDownLatch等一网打尽 category: - Java核心 - 并发编程 tag: - Java --- JDK中提供了一些工具类以供开发者使用。这样的话我们在遇到一些常见的应用场景时就可以使用这些工具类，而不用自己再重复造轮子了。 它们都在java.util.concurrent包下。先总体概括一下都有哪些工具类，它们有什么作用，然后再分别介绍它们的主要使用方法和原理。 | 类 | 作用 | | -------------- | ------------------------------------------ | | Semaphore | 限制线程的数量 | | Exchanger | 两个线程交换数据 | | CountDownLatch | 线程等待直到计数器减为0时开始工作 | | CyclicBarrier | 作用跟CountDownLatch类似，但是可以重复使用 | | Phaser | 增强的CyclicBarrier | 下面分别介绍这几个类。 ## Semaphore ### Semaphore介绍 Semaphore翻译过来是信号的意思。顾名思义，这个工具类提供的功能就是多个线程彼此“打信号”。而这个“信号”是一个`int`类型的数据，也可以看成是一种“资源”。 可以在构造函数中传入初始资源总数，以及是否使用“公平”的同步器。默认情况下，是非公平的。 ```java // 默认情况下使用非公平 public Semaphore(int permits) { sync = new NonfairSync(permits); } public Semaphore(int permits, boolean fair) { sync = fair ? new FairSync(permits) : new NonfairSync(permits); } ``` 最主要的方法是acquire方法和release方法。acquire()方法会申请一个permit，而release方法会释放一个permit。当然，你也可以申请多个acquire(int permits)或者释放多个release(int permits)。 每次acquire，permits就会减少一个或者多个。如果减少到了0，再有其他线程来acquire，那就要阻塞这个线程直到有其它线程release permit为止。 ### Semaphore案例 Semaphore往往用于资源有限的场景中，去限制线程的数量。举个例子，我想限制同时只能有3个线程在工作： ```java public class SemaphoreDemo { static class MyThread implements Runnable { private int value; private Semaphore semaphore; public MyThread(int value, Semaphore semaphore) { this.value = value; this.semaphore = semaphore; } @Override public void run() { try { semaphore.acquire(); // 获取permit System.out.println(String.format("当前线程是%d, 还剩%d个资源，还有%d个线程在等待", value, semaphore.availablePermits(), semaphore.getQueueLength())); // 睡眠随机时间，打乱释放顺序 Random random =new Random(); Thread.sleep(random.nextInt(1000)); System.out.println(String.format("线程%d释放了资源", value)); } catch (InterruptedException e) { e.printStackTrace(); } finally{ semaphore.release(); // 释放permit } } } public static void main(String[] args) { Semaphore semaphore = new Semaphore(3); for (int i = 0; i < 10; i++) { new Thread(new MyThread(i, semaphore)).start(); } } } ``` 输出： > 当前线程是1, 还剩2个资源，还有0个线程在等待 > 当前线程是0, 还剩1个资源，还有0个线程在等待 > 当前线程是6, 还剩0个资源，还有0个线程在等待 > 线程6释放了资源 > 当前线程是2, 还剩0个资源，还有6个线程在等待 > 线程2释放了资源 > 当前线程是4, 还剩0个资源，还有5个线程在等待 > 线程0释放了资源 > 当前线程是7, 还剩0个资源，还有4个线程在等待 > 线程1释放了资源 > 当前线程是8, 还剩0个资源，还有3个线程在等待 > 线程7释放了资源 > 当前线程是5, 还剩0个资源，还有2个线程在等待 > 线程4释放了资源 > 当前线程是3, 还剩0个资源，还有1个线程在等待 > 线程8释放了资源 > 当前线程是9, 还剩0个资源，还有0个线程在等待 > 线程9释放了资源 > 线程5释放了资源 > 线程3释放了资源 可以看到，在这次运行中，最开始是1, 0, 6这三个线程获得了资源，而其它线程进入了等待队列。然后当某个线程释放资源后，就会有等待队列中的线程获得资源。 当然，Semaphore默认的acquire方法是会让线程进入等待队列，且会抛出中断异常。但它还有一些方法可以忽略中断或不进入阻塞队列： ```java // 忽略中断 public void acquireUninterruptibly() public void acquireUninterruptibly(int permits) // 不进入等待队列，底层使用CAS public boolean tryAcquire public boolean tryAcquire(int permits) public boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException public boolean tryAcquire(long timeout, TimeUnit unit) ``` ### Semaphore原理 Semaphore内部有一个继承了AQS的同步器Sync，重写了`tryAcquireShared`方法。在这个方法里，会去尝试获取资源。 如果获取失败（想要的资源数量小于目前已有的资源数量），就会返回一个负数（代表尝试获取资源失败）。然后当前线程就会进入AQS的等待队列。 ## Exchanger Exchanger类用于两个线程交换数据。它支持泛型，也就是说你可以在两个线程之间传送任何数据。先来一个案例看看如何使用，比如两个线程之间想要传送字符串： ```java public class ExchangerDemo { public static void main(String[] args) throws InterruptedException { Exchanger exchanger = new Exchanger<>(); new Thread(() -> { try { System.out.println("这是线程A，得到了另一个线程的数据：" + exchanger.exchange("这是来自线程A的数据")); } catch (InterruptedException e) { e.printStackTrace(); } }).start(); System.out.println("这个时候线程A是阻塞的，在等待线程B的数据"); Thread.sleep(1000); new Thread(() -> { try { System.out.println("这是线程B，得到了另一个线程的数据：" + exchanger.exchange("这是来自线程B的数据")); } catch (InterruptedException e) { e.printStackTrace(); } }).start(); } } ``` 输出： > 这个时候线程A是阻塞的，在等待线程B的数据 > 这是线程B，得到了另一个线程的数据：这是来自线程A的数据 > 这是线程A，得到了另一个线程的数据：这是来自线程B的数据 可以看到，当一个线程调用exchange方法后，它是处于阻塞状态的，只有当另一个线程也调用了exchange方法，它才会继续向下执行。看源码可以发现它是使用park/unpark来实现等待状态的切换的，但是在使用park/unpark方法之前，使用了CAS检查，估计是为了提高性能。 Exchanger一般用于两个线程之间更方便地在内存中交换数据，因为其支持泛型，所以我们可以传输任何的数据，比如IO流或者IO缓存。根据JDK里面的注释的说法，可以总结为一下特性： - 此类提供对外的操作是同步的； - 用于成对出现的线程之间交换数据； - 可以视作双向的同步队列； - 可应用于基因算法、流水线设计等场景。 Exchanger类还有一个有超时参数的方法，如果在指定时间内没有另一个线程调用exchange，就会抛出一个超时异常。 ```java public V exchange(V x, long timeout, TimeUnit unit) ``` 那么问题来了，Exchanger只能是两个线程交换数据吗？那三个调用同一个实例的exchange方法会发生什么呢？答案是只有前两个线程会交换数据，第三个线程会进入阻塞状态。 需要注意的是，exchange是可以重复使用的。也就是说。两个线程可以使用Exchanger在内存中不断地再交换数据。 ## CountDownLatch ### CountDownLatch介绍 先来解读一下CountDownLatch这个类名字的意义。CountDown代表计数递减，Latch是“门闩”的意思。也有人把它称为“屏障”。而CountDownLatch这个类的作用也很贴合这个名字的意义，假设某个线程在执行任务之前，需要等待其它线程完成一些前置任务，必须等所有的前置任务都完成，才能开始执行本线程的任务。 CountDownLatch的方法也很简单，如下： ```java // 构造方法： public CountDownLatch(int count) public void await() // 等待 public boolean await(long timeout, TimeUnit unit) // 超时等待 public void countDown() // count - 1 public long getCount() // 获取当前还有多少count ``` ### CountDownLatch案例 我们知道，玩游戏的时候，在游戏真正开始之前，一般会等待一些前置任务完成，比如“加载地图数据”，“加载人物模型”，“加载背景音乐”等等。只有当所有的东西都加载完成后，玩家才能真正进入游戏。下面我们就来模拟一下这个demo。 ```java public class CountDownLatchDemo { // 定义前置任务线程 static class PreTaskThread implements Runnable { private String task; private CountDownLatch countDownLatch; public PreTaskThread(String task, CountDownLatch countDownLatch) { this.task = task; this.countDownLatch = countDownLatch; } @Override public void run() { try { Random random = new Random(); Thread.sleep(random.nextInt(1000)); System.out.println(task + " - 任务完成"); countDownLatch.countDown(); } catch (InterruptedException e) { e.printStackTrace(); } } } public static void main(String[] args) { // 假设有三个模块需要加载 CountDownLatch countDownLatch = new CountDownLatch(3); // 主任务 new Thread(() -> { try { System.out.println("等待数据加载..."); System.out.println(String.format("还有%d个前置任务", countDownLatch.getCount())); countDownLatch.await(); System.out.println("数据加载完成，正式开始游戏！"); } catch (InterruptedException e) { e.printStackTrace(); } }).start(); // 前置任务 new Thread(new PreTaskThread("加载地图数据", countDownLatch)).start(); new Thread(new PreTaskThread("加载人物模型", countDownLatch)).start(); new Thread(new PreTaskThread("加载背景音乐", countDownLatch)).start(); } } ``` 输出： > 等待数据加载... > 还有3个前置任务 > 加载人物模型 - 任务完成 > 加载背景音乐 - 任务完成 > 加载地图数据 - 任务完成 > 数据加载完成，正式开始游戏！ ### CountDownLatch原理 其实CountDownLatch类的原理挺简单的，内部同样是一个继承了AQS的实现类Sync，且实现起来还很简单，可能是JDK里面AQS的子类中最简单的实现了，有兴趣的读者可以去看看这个内部类的源码。 需要注意的是构造器中的**计数值（count）实际上就是闭锁需要等待的线程数量**。这个值只能被设置一次，而且CountDownLatch**没有提供任何机制去重新设置这个计数值**。 ## CyclicBarrier ### CyclicBarrier介绍 CyclicBarrirer从名字上来理解是“循环的屏障”的意思。前面提到了CountDownLatch一旦计数值`count`被降为0后，就不能再重新设置了，它只能起一次“屏障”的作用。而CyclicBarrier拥有CountDownLatch的所有功能，还可以使用`reset()`方法重置屏障。 ### CyclicBarrier Barrier被破坏 如果参与者（线程）在等待的过程中，Barrier被破坏，就会抛出BrokenBarrierException。可以用`isBroken()`方法检测Barrier是否被破坏。 1. 如果有线程已经处于等待状态，调用reset方法会导致已经在等待的线程出现BrokenBarrierException异常。并且由于出现了BrokenBarrierException，将会导致始终无法等待。 2. 如果在等待的过程中，线程被中断，会抛出InterruptedException异常，并且这个异常会传播到其他所有的线程。 3. 如果在执行屏障操作过程中发生异常，则该异常将传播到当前线程中，其他线程会抛出BrokenBarrierException，屏障被损坏。 4. 如果超出指定的等待时间，当前线程会抛出 TimeoutException 异常，其他线程会抛出BrokenBarrierException异常。 ### CyclicBarrier案例 我们同样用玩游戏的例子。如果玩一个游戏有多个“关卡”，那使用CountDownLatch显然不太合适，那需要为每个关卡都创建一个实例。那我们可以使用CyclicBarrier来实现每个关卡的数据加载等待功能。 ```java public class CyclicBarrierDemo { static class PreTaskThread implements Runnable { private String task; private CyclicBarrier cyclicBarrier; public PreTaskThread(String task, CyclicBarrier cyclicBarrier) { this.task = task; this.cyclicBarrier = cyclicBarrier; } @Override public void run() { // 假设总共三个关卡 for (int i = 1; i < 4; i++) { try { Random random = new Random(); Thread.sleep(random.nextInt(1000)); System.out.println(String.format("关卡%d的任务%s完成", i, task)); cyclicBarrier.await(); } catch (InterruptedException | BrokenBarrierException e) { e.printStackTrace(); } } } } public static void main(String[] args) { CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> { System.out.println("本关卡所有前置任务完成，开始游戏..."); }); new Thread(new PreTaskThread("加载地图数据", cyclicBarrier)).start(); new Thread(new PreTaskThread("加载人物模型", cyclicBarrier)).start(); new Thread(new PreTaskThread("加载背景音乐", cyclicBarrier)).start(); } } ``` 输出： > 关卡1的任务加载地图数据完成 > 关卡1的任务加载背景音乐完成 > 关卡1的任务加载人物模型完成 > 本关卡所有前置任务完成，开始游戏... > 关卡2的任务加载地图数据完成 > 关卡2的任务加载背景音乐完成 > 关卡2的任务加载人物模型完成 > 本关卡所有前置任务完成，开始游戏... > 关卡3的任务加载人物模型完成 > 关卡3的任务加载地图数据完成 > 关卡3的任务加载背景音乐完成 > 本关卡所有前置任务完成，开始游戏... 注意这里跟CountDownLatch的代码有一些不同。CyclicBarrier没有分为`await()`和`countDown()`，而是只有单独的一个`await()`方法。 一旦调用await()方法的线程数量等于构造方法中传入的任务总量（这里是3），就代表达到屏障了。CyclicBarrier允许我们在达到屏障的时候可以执行一个任务，可以在构造方法传入一个Runnable类型的对象。上述案例就是在达到屏障时，输出“本关卡所有前置任务完成，开始游戏...”。 ```java // 构造方法 public CyclicBarrier(int parties) { this(parties, null); } public CyclicBarrier(int parties, Runnable barrierAction) { // 具体实现 } ``` ### CyclicBarrier原理 CyclicBarrier虽说功能与CountDownLatch类似，但是实现原理却完全不同，CyclicBarrier内部使用的是Lock + Condition实现的等待/通知模式。详情可以查看这个方法的源码： ```java private int dowait(boolean timed, long nanos) ``` ## Phaser ### Phaser介绍 Phaser这个单词是“移相器，相位器”的意思（好吧，笔者并不懂这是什么玩意，下方资料来自百度百科）。这个类是从JDK 1.7 中出现的。 > 移相器（Phaser）能够对波的相位进行调整的一种装置。任何传输介质对在其中传导的波动都会引入相移，这是早期模拟移相器的原理；现代电子技术发展后利用A/D、D/A转换实现了数字移相，顾名思义，它是一种不连续的移相技术，但特点是移相精度高。 > 移相器在雷达、导弹姿态控制、加速器、通信、仪器仪表甚至于音乐等领域都有着广泛的应用 Phaser类有点复杂，这里只介绍一些基本的用法和知识点。详情可以查看JDK文档，文档里有这个类非常详尽的介绍。 前面我们介绍了CyclicBarrier，可以发现它在构造方法里传入“任务总量”`parties`之后，就不能修改这个值了，并且每次调用`await()`方法也只能消耗一个`parties`计数。但Phaser可以动态地调整任务总量！ 名词解释： - party：对应一个线程，数量可以通过register或者构造参数传入; - arrive：对应一个party的状态，初始时是unarrived，当调用`arriveAndAwaitAdvance()`或者 `arriveAndDeregister()`进入arrive状态，可以通过`getUnarrivedParties()`获取当前未到达的数量; - register：注册一个party，每一阶段必须所有注册的party都到达才能进入下一阶段; - deRegister：减少一个party。 - phase：阶段，当所有注册的party都arrive之后，将会调用Phaser的`onAdvance()`方法来判断是否要进入下一阶段。 Phaser终止的两种途径，Phaser维护的线程执行完毕或者`onAdvance()`返回`true` 此外Phaser还能维护一个树状的层级关系，构造的时候new Phaser(parentPhaser)，对于Task执行时间短的场景（竞争激烈），也就是说有大量的**party**, 那可以把每个Phaser的任务量设置较小，多个Phaser共同继承一个父Phaser。 > Phasers with large numbers of parties that would otherwise experience heavy synchronization contention costs may instead be set up so that groups of sub-phasers share a common parent. This may greatly increase throughput even though it incurs greater per-operation overhead. > > 翻译：如果有大量的party，那许多线程可能同步的竞争成本比较高。所以可以拆分成多个子Phaser共享一个共同的父Phaser。这可能会大大增加吞吐量，即使它会带来更多的每次操作开销。 ### Phaser案例 还是游戏的案例。假设我们游戏有三个关卡，但只有第一个关卡有新手教程，需要加载新手教程模块。但后面的第二个关卡和第三个关卡都不需要。我们可以用Phaser来做这个需求。 代码： ```java public class PhaserDemo { static class PreTaskThread implements Runnable { private String task; private Phaser phaser; public PreTaskThread(String task, Phaser phaser) { this.task = task; this.phaser = phaser; } @Override public void run() { for (int i = 1; i < 4; i++) { try { // 第二次关卡起不加载NPC，跳过 if (i >= 2 && "加载新手教程".equals(task)) { continue; } Random random = new Random(); Thread.sleep(random.nextInt(1000)); System.out.println(String.format("关卡%d，需要加载%d个模块，当前模块【%s】", i, phaser.getRegisteredParties(), task)); // 从第二个关卡起，不加载NPC if (i == 1 && "加载新手教程".equals(task)) { System.out.println("下次关卡移除加载【新手教程】模块"); phaser.arriveAndDeregister(); // 移除一个模块 } else { phaser.arriveAndAwaitAdvance(); } } catch (InterruptedException e) { e.printStackTrace(); } } } } public static void main(String[] args) { Phaser phaser = new Phaser(4) { @Override protected boolean onAdvance(int phase, int registeredParties) { System.out.println(String.format("第%d次关卡准备完成", phase + 1)); return phase == 3 || registeredParties == 0; } }; new Thread(new PreTaskThread("加载地图数据", phaser)).start(); new Thread(new PreTaskThread("加载人物模型", phaser)).start(); new Thread(new PreTaskThread("加载背景音乐", phaser)).start(); new Thread(new PreTaskThread("加载新手教程", phaser)).start(); } } ``` 输出： > 关卡1，需要加载4个模块，当前模块【加载背景音乐】 > 关卡1，需要加载4个模块，当前模块【加载新手教程】 > 下次关卡移除加载【新手教程】模块 > 关卡1，需要加载3个模块，当前模块【加载地图数据】 > 关卡1，需要加载3个模块，当前模块【加载人物模型】 > 第1次关卡准备完成 > 关卡2，需要加载3个模块，当前模块【加载地图数据】 > 关卡2，需要加载3个模块，当前模块【加载背景音乐】 > 关卡2，需要加载3个模块，当前模块【加载人物模型】 > 第2次关卡准备完成 > 关卡3，需要加载3个模块，当前模块【加载人物模型】 > 关卡3，需要加载3个模块，当前模块【加载地图数据】 > 关卡3，需要加载3个模块，当前模块【加载背景音乐】 > 第3次关卡准备完成 这里要注意关卡1的输出，在“加载新手教程”线程中调用了`arriveAndDeregister()`减少一个party之后，后面的线程使用`getRegisteredParties()`得到的是已经被修改后的parties了。但是当前这个阶段(phase)，仍然是需要4个parties都arrive才触发屏障的。从下一个阶段开始，才需要3个parties都arrive就触发屏障。 另外Phaser类用来控制某个阶段的线程数量很有用，但它并在意这个阶段具体有哪些线程arrive，只要达到它当前阶段的parties值，就触发屏障。所以我这里的案例虽然制定了特定的线程（加载新手教程）来更直观地表述Phaser的功能，但是其实Phaser是没有分辨具体是哪个线程的功能的，它在意的只是数量，这一点需要读者注意。 ### Phaser原理 Phaser类的原理相比起来要复杂得多。它内部使用了两个基于Fork-Join框架的原子类辅助： ```java private final AtomicReference evenQ; private final AtomicReference oddQ; static final class QNode implements ForkJoinPool.ManagedBlocker { // 实现代码 } ``` 有兴趣的读者可以去看看JDK源代码，这里不做过多叙述。 总的来说，CountDownLatch，CyclicBarrier，Phaser是一个比一个强大，但也一个比一个复杂。根据自己的业务需求合理选择即可。 --- >编辑：沉默王二，内容大部分来源以下三个开源仓库： >- [深入浅出 Java 多线程](http://concurrent.redspider.group/) >- [并发编程知识总结](https://github.com/CL0610/Java-concurrency) >- [Java八股文](https://github.com/CoderLeixiaoshuai/java-eight-part)