Java多线程与高并发专题—— CyclicBarrier 和 CountDownLatch 有什么异同？

引入

上一篇我们了解CountDownLatch的原理和常见用法，在CountDownLatch的源码注释中，有提到：

另一种典型用法是将一个问题分解为 N 个部分，用一个Runnable描述每个部分，该Runnable执行相应部分的任务并对闭锁进行倒计时操作，然后将所有的Runnable排入一个Executor中。当所有子部分都完成时，协调线程将能够通过await。（当线程必须以这种方式反复进行倒计时操作时，应使用CyclicBarrier代替。）

本文我们就来看一下这个CyclicBarrier和CountDownLatch 的异同。

还是老样子，先初步了解一下CyclicBarrier，先看看CyclicBarrier的源码注释：

A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point. CyclicBarriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other. The barrier is called cyclic because it can be re-used after the waiting threads are released.
A CyclicBarrier supports an optional Runnable command that is run once per barrier point, after the last thread in the party arrives, but before any threads are released. This barrier action is useful for updating shared-state before any of the parties continue.
Sample usage: Here is an example of using a barrier in a parallel decomposition design:

class Solver {   final int N;   final float[][] data;   final CyclicBarrier barrier;    class Worker implements Runnable {    int myRow;     Worker(int row) { myRow = row; }     public void run() {       while (!done()) {       processRow(myRow);       try {          barrier. await();       } catch (InterruptedException ex) {      return;       } catch (BrokenBarrierException ex) {    return;        }   }  }  } public Solver(float[][] matrix) {  data = matrix;  N = matrix. length;   Runnable barrierAction = new Runnable() { public void run() { mergeRows(...); }};   barrier = new CyclicBarrier(N, barrierAction);    List<Thread> threads = new ArrayList<Thread>(N);    for (int i = 0; i < N; i++) {       Thread thread = new Thread(new Worker(i));    threads. add(thread);      thread. start();     }      // wait until done     for (Thread thread : threads)      thread. join();   } 
}

Here, each worker thread processes a row of the matrix then waits at the barrier until all rows have been processed. When all rows are processed the supplied Runnable barrier action is executed and merges the rows. If the merger determines that a solution has been found then done() will return true and each worker will terminate.
If the barrier action does not rely on the parties being suspended when it is executed, then any of the threads in the party could execute that action when it is released. To facilitate this, each invocation of await returns the arrival index of that thread at the barrier. You can then choose which thread should execute the barrier action, for example:
 

if (barrier. await() == 0) {// log the completion of this iteration
}

The CyclicBarrier uses an all-or-none breakage model for failed synchronization attempts: If a thread leaves a barrier point prematurely because of interruption, failure, or timeout, all other threads waiting at that barrier point will also leave abnormally via BrokenBarrierException (or InterruptedException if they too were interrupted at about the same time).
Memory consistency effects: Actions in a thread prior to calling await() happen-before actions that are part of the barrier action, which in turn happen-before actions following a successful return from the corresponding await() in other threads.

翻译：

同步辅助工具，它允许一组线程相互等待，直到全部到达一个公共的屏障点。循环屏障（CyclicBarrier）在涉及固定数量线程组且这些线程偶尔需要相互等待的程序中很有用。之所以称为 “循环”，是因为在等待的线程被释放后，它可以再次使用。
循环屏障支持一个可选的 Runnable 命令，在组内最后一个线程到达后，但在任何线程被释放之前，每次到达屏障点时都会运行此命令。在任何线程组继续执行之前，此屏障操作对于更新共享状态很有用。
示例用法：以下是在并行分解设计中使用屏障的一个示例：

/*** Solver 类用于解决矩阵相关的并发计算问题。* 该类使用 CyclicBarrier 来同步多个工作线程的执行。*/
class Solver {// 矩阵的行数final int N;// 存储矩阵数据的二维数组final float[][] data;// 用于线程同步的 CyclicBarrierfinal CyclicBarrier barrier;/*** Worker 类是一个内部类，实现了 Runnable 接口，用于处理矩阵的每一行。*/class Worker implements Runnable {// 当前工作线程负责处理的行号int myRow;/*** 构造函数，初始化当前工作线程负责处理的行号。* @param row 当前工作线程负责处理的行号*/Worker(int row) { myRow = row; }/*** 实现 Runnable 接口的 run 方法，该方法包含线程的主要逻辑。* 线程会持续处理矩阵行，直到完成所有任务。*/public void run() {// 循环处理矩阵行，直到 done 方法返回 truewhile (!done()) {// 处理当前行的数据processRow(myRow);try {// 等待所有线程到达屏障点barrier. await();} catch (InterruptedException ex) {// 如果线程被中断，则返回return;} catch (BrokenBarrierException ex) {// 如果屏障被破坏，则返回return;}}}}/*** Solver 类的构造函数，初始化矩阵数据和线程池。* @param matrix 输入的矩阵数据*/public Solver(float[][] matrix) {// 初始化矩阵数据data = matrix;// 获取矩阵的行数N = matrix. length;// 定义屏障动作，当所有线程到达屏障点时执行Runnable barrierAction = new Runnable() {/*** 实现 Runnable 接口的 run 方法，该方法包含屏障动作的逻辑。* 这里调用 mergeRows 方法来合并行数据。*/public void run() { mergeRows(...); }};// 初始化 CyclicBarrierbarrier = new CyclicBarrier(N, barrierAction);// 创建一个包含 N 个线程的列表List<Thread> threads = new ArrayList<Thread>(N);// 为矩阵的每一行创建一个工作线程for (int i = 0; i < N; i++) {// 创建一个新的线程，执行 Worker 任务Thread thread = new Thread(new Worker(i));// 将线程添加到线程列表中threads. add(thread);// 启动线程thread. start();}// 等待所有线程完成任务for (Thread thread : threads)// 等待线程执行完毕thread. join();}
}

在此示例中，每个工作线程处理矩阵的一行，然后在屏障处等待，直到所有行都处理完毕。当所有行都处理完成后，所提供的 Runnable 屏障动作将被执行，以合并这些行。如果合并操作确定已找到解决方案，那么done()方法将返回true，每个工作线程将终止。
如果屏障动作在执行时不依赖于线程组中的线程处于挂起状态，那么在线程组中的任何线程被释放时，都可以执行该动作。为了便于实现这一点，每次调用await方法都会返回该线程在屏障处的到达索引。然后，你可以选择应由哪个线程执行屏障动作，例如：

/*** 检查是否所有线程都已到达屏障点。* 当所有线程都到达屏障点时，CyclicBarrier的await()方法将返回0。* 如果返回值为0，则表示当前迭代已完成，此时记录该迭代的完成信息。*/
if (barrier. await() == 0) {// log the completion of this iteration
}

循环屏障（CyclicBarrier）对于同步尝试失败采用 “全有或全无” 的中断模型：如果一个线程由于中断、失败或超时而过早地离开屏障点，那么在该屏障点等待的所有其他线程也会通过BrokenBarrierException异常不正常地离开（如果它们几乎在同一时间也被中断，则会抛出InterruptedException异常）。
内存一致性影响：一个线程中调用await()方法之前的操作，先行发生于作为屏障动作一部分的操作，而这些操作又先行发生于其他线程中从相应的await()方法成功返回之后的操作。

具体源码如下：

/*** 一个同步辅助类，允许一组线程相互等待，直到所有线程都到达一个共同的屏障点。* 该屏障是可循环使用的，因为在等待的线程被释放后，它可以被再次使用。*/
public class CyclicBarrier {/*** 每次使用屏障都由一个Generation实例表示。* 当屏障被触发或重置时，Generation会发生变化。* 由于锁分配给等待线程的方式是不确定的，可能会有多个Generation与使用屏障的线程相关联，* 但在任何时候只有一个是活跃的（即{@code count}所适用的那个），其余的要么已损坏，要么已被触发。* 如果发生了损坏但没有后续的重置，可能就没有活跃的Generation。*/private static class Generation {// 标记屏障是否已损坏boolean broken = false;}/** 用于保护屏障入口的锁 */private final ReentrantLock lock = new ReentrantLock();/** 线程等待直到屏障被触发的条件 */private final Condition trip = lock.newCondition();/** 参与屏障的线程数量 */private final int parties;/* 屏障被触发时要执行的命令 */private final Runnable barrierCommand;/** 当前的Generation实例 */private Generation generation = new Generation();/*** 仍在等待的线程数量。在每个Generation中，从parties递减到0。* 在每个新的Generation或屏障损坏时，会重置为parties。*/private int count;/*** 更新屏障触发后的状态，并唤醒所有等待的线程。* 此方法必须在持有锁的情况下调用。*/private void nextGeneration() {// 通知上一个Generation完成trip.signalAll();// 设置下一个Generationcount = parties;generation = new Generation();}/*** 将当前屏障的Generation标记为损坏，并唤醒所有等待的线程。* 此方法必须在持有锁的情况下调用。*/private void breakBarrier() {// 标记屏障已损坏generation.broken = true;// 重置等待线程数量count = parties;// 唤醒所有等待的线程trip.signalAll();}/*** 屏障的主要逻辑，涵盖了各种策略。** @param timed  是否设置超时* @param nanos  超时时间（纳秒）* @return       当前线程的到达索引* @throws InterruptedException  如果线程在等待过程中被中断* @throws BrokenBarrierException 如果屏障已损坏* @throws TimeoutException       如果设置了超时且超时发生*/private int dowait(boolean timed, long nanos)throws InterruptedException, BrokenBarrierException,TimeoutException {// 获取锁final ReentrantLock lock = this.lock;lock.lock();try {// 获取当前的Generationfinal Generation g = generation;// 如果屏障已损坏，抛出异常if (g.broken)throw new BrokenBarrierException();// 如果当前线程被中断，打破屏障并抛出异常if (Thread.interrupted()) {breakBarrier();throw new InterruptedException();}// 减少等待的线程数量int index = --count;if (index == 0) {  // 所有线程都已到达，触发屏障boolean ranAction = false;try {// 获取屏障触发时要执行的命令final Runnable command = barrierCommand;if (command != null)command.run();ranAction = true;// 更新到下一个GenerationnextGeneration();return 0;} finally {// 如果命令执行失败，打破屏障if (!ranAction)breakBarrier();}}// 循环等待，直到屏障被触发、损坏、线程被中断或超时for (;;) {try {if (!timed)// 无超时等待trip.await();else if (nanos > 0L)// 超时等待nanos = trip.awaitNanos(nanos);} catch (InterruptedException ie) {if (g == generation && ! g.broken) {// 如果当前Generation未损坏，打破屏障并抛出异常breakBarrier();throw ie;} else {// 重新设置中断状态Thread.currentThread().interrupt();}}// 如果屏障已损坏，抛出异常if (g.broken)throw new BrokenBarrierException();// 如果Generation已更新，返回当前线程的到达索引if (g != generation)return index;// 如果设置了超时且超时发生，打破屏障并抛出异常if (timed && nanos <= 0L) {breakBarrier();throw new TimeoutException();}}} finally {// 释放锁lock.unlock();}}/*** 创建一个新的{@code CyclicBarrier}，当指定数量的线程等待时触发，并在屏障触发时执行指定的命令。** @param parties       必须调用{@link #await}方法才能触发屏障的线程数量* @param barrierAction 屏障触发时要执行的命令，如果没有则为{@code null}* @throws IllegalArgumentException 如果{@code parties}小于1*/public CyclicBarrier(int parties, Runnable barrierAction) {if (parties <= 0) throw new IllegalArgumentException();this.parties = parties;this.count = parties;this.barrierCommand = barrierAction;}/*** 创建一个新的{@code CyclicBarrier}，当指定数量的线程等待时触发，不执行预定义的命令。** @param parties 必须调用{@link #await}方法才能触发屏障的线程数量* @throws IllegalArgumentException 如果{@code parties}小于1*/public CyclicBarrier(int parties) {this(parties, null);}/*** 返回触发此屏障所需的线程数量。** @return 触发此屏障所需的线程数量*/public int getParties() {return parties;}/*** 等待直到所有{@linkplain #getParties 参与方}都在这个屏障上调用了{@code await}方法。** @return 当前线程的到达索引，其中索引{@code getParties() - 1}表示第一个到达，0表示最后一个到达* @throws InterruptedException 如果当前线程在等待过程中被中断* @throws BrokenBarrierException 如果在当前线程等待时，另一个线程被中断或超时，或者屏障被重置，或者屏障在调用{@code await}时已损坏，或者屏障动作（如果存在）由于异常而失败*/public int await() throws InterruptedException, BrokenBarrierException {try {return dowait(false, 0L);} catch (TimeoutException toe) {// 这种情况不会发生throw new Error(toe);}}/*** 等待直到所有{@linkplain #getParties 参与方}都在这个屏障上调用了{@code await}方法，或者指定的等待时间过去。** @param timeout 等待屏障的时间* @param unit    超时参数的时间单位* @return 当前线程的到达索引，其中索引{@code getParties() - 1}表示第一个到达，0表示最后一个到达* @throws InterruptedException 如果当前线程在等待过程中被中断* @throws TimeoutException 如果指定的超时时间过去。在这种情况下，屏障将被打破。* @throws BrokenBarrierException 如果在当前线程等待时，另一个线程被中断或超时，或者屏障被重置，或者屏障在调用{@code await}时已损坏，或者屏障动作（如果存在）由于异常而失败*/public int await(long timeout, TimeUnit unit)throws InterruptedException,BrokenBarrierException,TimeoutException {return dowait(true, unit.toNanos(timeout));}/*** 查询此屏障是否处于损坏状态。** @return 如果自构造或上次重置以来，有一个或多个参与方由于中断或超时退出了此屏障，或者屏障动作由于异常而失败，则返回{@code true}；否则返回{@code false}。*/public boolean isBroken() {final ReentrantLock lock = this.lock;lock.lock();try {return generation.broken;} finally {lock.unlock();}}/*** 将屏障重置为初始状态。如果有任何参与方当前正在屏障处等待，它们将以{@link BrokenBarrierException}返回。* 注意，在由于其他原因发生损坏后进行重置可能会很复杂；线程需要以其他方式重新同步，并选择一个来执行重置。* 可能更可取的做法是创建一个新的屏障以供后续使用。*/public void reset() {final ReentrantLock lock = this.lock;lock.lock();try {// 打破当前的GenerationbreakBarrier();// 开始一个新的GenerationnextGeneration();} finally {lock.unlock();}}/*** 返回当前正在屏障处等待的参与方数量。此方法主要用于调试和断言。** @return 当前在{@link #await}方法中被阻塞的参与方数量*/public int getNumberWaiting() {final ReentrantLock lock = this.lock;lock.lock();try {return parties - count;} finally {lock.unlock();}}
}

CyclicBarrier的作用

可以看到CyclicBarrier 和 CountDownLatch 确实有一定的相似性，它们都能阻塞一个或者一组线程，直到某种预定的条件达到之后，这些之前在等待的线程才会统一出发，继续向下执行。正因为它们有这个相似点，你可能会认为它们的作用是完全一样的，其实并不是。

CyclicBarrier 可以构造出一个集结点，当某一个线程执行 await() 的时候，它就会到这个集结点开始等待，等待这个栅栏被撤销。直到预定数量的线程都到了这个集结点之后，这个栅栏就会被撤销，之前等待的线程就在此刻统一出发，继续去执行剩下的任务。

举一个生活中的例子。假设我们班级春游去公园里玩，并且会租借三人自行车，每个人都可以骑，但由于这辆自行车是三人的，所以要凑齐三个人才能骑一辆，而且从公园大门走到自行车驿站需要一段时间。那么我们模拟这个场景，写出如下代码：

/*** 该类演示了 CyclicBarrier 的使用，CyclicBarrier 是一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点。*/
public class CyclicBarrierDemo {/*** 主方法，程序的入口点。* 创建一个 CyclicBarrier 实例，该实例在 3 个线程到达屏障点时触发。* 创建并启动 6 个线程，每个线程执行一个 Task 任务。* * @param args 命令行参数*/public static void main(String[] args) {// 创建一个 CyclicBarrier 实例，当 3 个线程调用 await() 方法时，屏障将被打破CyclicBarrier cyclicBarrier = new CyclicBarrier(3);// 创建并启动 6 个线程，每个线程执行一个 Task 任务for (int i = 0; i < 6; i++) {// 创建一个新线程并启动，每个线程执行一个 Task 任务new Thread(new Task(i + 1, cyclicBarrier)).start();}}/*** 静态内部类，实现了 Runnable 接口，表示一个可以在新线程中执行的任务。* 每个任务代表一个同学，从大门出发前往自行车驿站，到达后等待其他同学到达。*/static class Task implements Runnable {// 同学的编号private int id;// 用于同步的 CyclicBarrier 实例private CyclicBarrier cyclicBarrier;/*** 构造函数，初始化任务的同学编号和 CyclicBarrier 实例。* * @param id            同学的编号* @param cyclicBarrier 用于同步的 CyclicBarrier 实例*/public Task(int id, CyclicBarrier cyclicBarrier) {// 初始化同学的编号this.id = id;// 初始化 CyclicBarrier 实例this.cyclicBarrier = cyclicBarrier;}/*** 线程执行的任务逻辑。* 模拟同学从大门出发前往自行车驿站，到达后等待其他同学到达，然后一起骑车。*/@Overridepublic void run() {// 输出同学从大门出发的信息System.out.println("同学" + id + "现在从大门出发，前往自行车驿站");try {// 模拟同学前往自行车驿站的时间，随机睡眠 0 到 10 秒Thread.sleep((long) (Math.random() * 10000));// 输出同学到达自行车驿站并开始等待的信息System.out.println("同学" + id + "到了自行车驿站，开始等待其他人到达");// 调用 CyclicBarrier 的 await() 方法，等待其他同学到达cyclicBarrier.await();// 输出同学开始骑车的信息System.out.println("同学" + id + "开始骑车");} catch (InterruptedException e) {// 当线程在睡眠或等待过程中被中断时，打印堆栈跟踪信息e.printStackTrace();} catch (BrokenBarrierException e) {// 当 CyclicBarrier 被破坏时，打印堆栈跟踪信息e.printStackTrace();}}}
}

在这段代码中可以看到，首先建了一个参数为 3 的 CyclicBarrier，参数为 3 的意思是需要等待 3 个线程到达这个集结点才统一放行；然后我们又在 for 循环中去开启了 6 个线程，每个线程中执行的Runnable 对象就在下方的 Task 类中，直接看到它的 run 方法，它首先会打印出"同学某某现在从大门出发，前往自行车驿站"，然后是一个随机时间的睡眠，这就代表着从大门开始步行走到自行车驿站的时间，由于每个同学的步行速度不一样，所以时间用随机值来模拟。

当同学们都到了驿站之后，比如某一个同学到了驿站，首先会打印出“同学某某到了自行车驿站，开始等待其他人到达”的消息，然后去调用 CyclicBarrier 的 await() 方法。一旦它调用了这个方法，它就会陷入等待，直到三个人凑齐，才会继续往下执行，一旦开始继续往下执行，就意味着 3 个同学开始一起骑车了，所以打印出“某某开始骑车”这个语句。

接下来我们运行一下这个程序，结果如下所示：

可以看到 6 个同学纷纷从大门出发走到自行车驿站，因为每个人的速度不一样，所以会有 3 个同学先到自行车驿站，不过在这 3 个先到的同学里面，前面 2 个到的都必须等待第 3 个人到齐之后，才可以开始骑车。后面的同学也一样，由于第一辆车已经被骑走了，第二辆车依然也要等待 3 个人凑齐才能统一发车。

要想实现这件事情，如果你不利用 CyclicBarrier 去做的话，逻辑可能会非常复杂，因为你也不清楚哪个同学先到、哪个后到。而用了 CyclicBarrier 之后，可以非常简洁优雅的实现这个逻辑，这就是它的一个非常典型的应用场景。

执行动作 barrierAction

public CyclicBarrier(int parties, Runnable barrierAction)：当 parties 线程到达集结点时，继续往下执行前，会执行这一次这个动作。

接下来我们再介绍一下它的一个额外功能，就是执行动作 barrierAction 功能。CyclicBarrier 还有一个构造函数是传入两个参数的，第一个参数依然是 parties，代表需要几个线程到齐；第二个参数是一个Runnable 对象，它就是我们下面所要介绍的 barrierAction。

当预设数量的线程到达了集结点之后，在出发的时候，便会执行这里所传入的 Runnable 对象，那么假设我们把刚才那个代码的构造函数改成如下这个样子：

具体代码如下： 

public class CyclicBarrierDemo {/*** 主方法,用于演示CyclicBarrier的使用* @param args 命令行参数(未使用)*/public static void main(String[] args) {// 创建一个CyclicBarrier实例,当3个线程到达屏障点时,触发屏障动作CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> {System.out.println("凑齐3人了,出发!");});// 循环创建6个线程,每个线程代表一个同学for (int i = 0; i < 6; i++) {// 创建一个新的线程,并启动它new Thread(new CyclicBarrierDemo.Task(i + 1, cyclicBarrier)).start();}}/*** 内部静态类Task,实现Runnable接口,代表每个同学的任务*/static class Task implements Runnable {// 同学的编号,使用final修饰,确保其值在初始化后不可变private final int id;// 用于同步的CyclicBarrier实例private CyclicBarrier cyclicBarrier;/*** Task类的构造方法* @param id 同学的编号* @param cyclicBarrier 用于同步的CyclicBarrier实例*/public Task(int id, CyclicBarrier cyclicBarrier) {this.id = id;this.cyclicBarrier = cyclicBarrier;}@Overridepublic void run() {// 输出同学从大门出发的信息System.out.println("同学" + id + "现在从大门出发,前往自行车驿站");try {// 模拟同学到达自行车驿站所需的时间,随机睡眠一段时间Thread.sleep((long) (Math.random() * 10000));// 输出同学到达自行车驿站并开始等待的信息System.out.println("同学" + id + "到了自行车驿站,开始等待其他人到达");// 线程到达屏障点,等待其他线程cyclicBarrier.await();// 输出同学开始骑车的信息System.out.println("同学" + id + "开始骑车");} catch (InterruptedException e) {// 捕获线程中断异常,并输出日志System.err.println("同学" + id + "的线程被中断: " + e.getMessage());} catch (BrokenBarrierException e) {// 捕获屏障损坏异常,并输出日志System.err.println("同学" + id + "遇到屏障损坏异常: " + e.getMessage());}}}
}

可以看出，我们传入了第二个参数，它是一个 Runnable 对象，在这里传入了这个 Runnable 之后，这个任务就会在到齐的时候去打印"凑齐3人了，出发！"。上面的代码如果改成这个样子，则执行结果如下所示：

可以看出，三个人凑齐了一组之后，就会打印出“凑齐 3 人了，出发！”这样的语句，该语句恰恰是我们在这边传入 Runnable 所执行的结果。

值得注意的是，这个语句每个周期只打印一次，不是说你有几个线程在等待就打印几次，而是说这个任务只在“开闸”的时候执行一次。

CyclicBarrier 和 CountDownLatch 的异同

下面我们来总结一下 CyclicBarrier 和 CountDownLatch 有什么异同。

相同点：都能阻塞一个或一组线程，直到某个预设的条件达成发生，再统一出发。

但是它们也有很多不同点，具体如下：

• 作用对象不同：CyclicBarrier 要等固定数量的线程都到达了栅栏位置才能继续执行，而CountDownLatch 只需等待数字倒数到 0，也就是说 CountDownLatch 作用于事件，但CyclicBarrier 作用于线程；CountDownLatch 是在调用了 countDown 方法之后把数字倒数减 1，而 CyclicBarrier 是在某线程开始等待后把计数减 1。
• 可重用性不同：CountDownLatch 在倒数到 0  并且触发门闩打开后，就不能再次使用了，除非新建一个新的实例；而 CyclicBarrier 可以重复使用，在刚才的代码中也可以看出，每 3 个同学到了之后都能出发，并不需要重新新建实例。CyclicBarrier 还可以随时调用 reset 方法进行重置，如果重置时有线程已经调用了 await 方法并开始等待，那么这些线程则会抛出BrokenBarrierException 异常。
• 执行动作不同：CyclicBarrier 有执行动作 barrierAction，而 CountDownLatch 没这个功能。