前言

CountDownLatch 可以实现多个线程协调，在所有制定的线程到达屏障（栅栏）后，主线程才执行任务。
但是，CountDownLatch 有个缺陷，它只能使用 1 次。而在有些场合，似乎有点浪费，因为需要不停创建 CountDownLatch 实例。JDK 在 CountDownLatch 的文档中向我们介绍了 CyclicBarrier，方便我们重复完成某个任务。

源码分析

内部成员

  1. /** 重入锁，栅栏的保护者 */
  2. private final ReentrantLock lock = new ReentrantLock();
  3. /** 进入Condition Queue，等待同一个 Generation 的其它线程 */
  4. private final Condition trip = lock.newCondition();
  5. /** 参与者（线程）数量，构造时指定。用于控制什么时候释放打开栅栏，让所有线程通过 */
  6. private final int parties;
  7. /** 所有线程执行完毕后，另外执行的任务，构造时指定 */
  8. private final Runnable barrierCommand;
  9. /** 当代的标识 */
  10. private Generation generation = new Generation();
  11. /**
  12. 正在等待着的参与者数量，每一代都会从参与者数量倒数至 0。
  13. 在栅栏被破坏或开启新一代时会重置参与者数量
  14. */
  15. private int count;

在 CyclicBarrier 中，有个“代”的概念。因为 CyclicBarrier 是可以复用的，那么每次所有的线程通过了栅栏，就表示一代过去了，就如同跨年。

构造方法

CyclicBarrier 支持在所有线程通过栅栏的时候，另外执行一个任务。

  1. public CyclicBarrier(int parties) {
  2.     this(parties, null);
  3. }
  4. public CyclicBarrier(int parties, Runnable barrierAction) {
  5.     if (parties <= 0) throw new IllegalArgumentException();
  6.     this.parties = parties;
  7.     this.count = parties;
  8.     this.barrierCommand = barrierAction;
  9. }

await 方法

CyclicBarrier 最重要的方法是 await，当执行了这个方法，就是树立了一个栅栏，将所有的线程挡住了。只有所有的线程都到了这个栅栏的时候，栅栏才会打开。
举个例子，报团出去旅游时，导游经常会把我们拉到金银首饰等消费品店，通知我们自由活动，等 1 个小时后集合。经过 1 个小时，导游必须等到所有的游客都到齐后才能通知司机开车，然后前往下个消费品店。
await 方法调用的 dowait 方法。

  1. private int dowait(boolean timed, long nanos)
  2.     throws InterruptedException, BrokenBarrierException,
  3.            TimeoutException {
  4.     final ReentrantLock lock = this.lock;
  5.     // 锁住
  6.     lock.lock();
  7.     try {
  8.         // 当前代
  9.         final Generation g = generation;
  10.         // 如果这代损坏了，抛出异常
  11.         if (g.broken)
  12.             throw new BrokenBarrierException();
  13.         // 如果线程中断了，抛出异常
  14.         if (Thread.interrupted()) {
  15.             // 将损坏状态设置为 true
  16.             // 并通知其他阻塞在此栅栏上的线程
  17.             breakBarrier();
  18.             throw new InterruptedException();
  19.         }
  20.         // 获取下标    
  21.         int index = --count;
  22.         // 如果是 0 ,说明到头了
  23.         if (index == 0) {  // tripped
  24.             boolean ranAction = false;
  25.             try {
  26.                 final Runnable command = barrierCommand;
  27.                 // 执行栅栏任务
  28.                 if (command != null)
  29.                     command.run();
  30.                 ranAction = true;
  31.                 // 更新一代,将 count 重置,将 generation 重置.
  32.                 // 唤醒之前等待的线程
  33.                 nextGeneration();
  34.                 // 结束
  35.                 return 0;
  36.             } finally {
  37.                 // 如果执行栅栏任务的时候失败了,就将栅栏失效
  38.                 if (!ranAction)
  39.                     breakBarrier();
  40.             }
  41.         }
  42.         for (;;) {
  43.             try {
  44.                 // 如果没有时间限制,则直接等待,直到被唤醒
  45.                 if (!timed)
  46.                     trip.await();
  47.                 // 如果有时间限制,则等待指定时间
  48.                 else if (nanos > 0L)
  49.                     nanos = trip.awaitNanos(nanos);
  50.             } catch (InterruptedException ie) {
  51.                 // g == generation >> 当前代
  52.                 // ! g.broken >>> 没有损坏
  53.                 if (g == generation && ! g.broken) {
  54.                     // 让栅栏失效
  55.                     breakBarrier();
  56.                     throw ie;
  57.                 } else {
  58.                     // 上面条件不满足,说明这个线程不是这代的.
  59.                     // 就不会影响当前这代栅栏执行逻辑.所以,就打个标记就好了
  60.                     Thread.currentThread().interrupt();
  61.                 }
  62.             }
  63.             // 当有任何一个线程中断了,会调用 breakBarrier 方法.
  64.             // 就会唤醒其他的线程,其他线程醒来后,也要抛出异常
  65.             if (g.broken)
  66.                 throw new BrokenBarrierException();
  67.             // g != generation >>> 正常换代了
  68.             // 一切正常,返回当前线程所在栅栏的下标
  69.             if (g != generation)
  70.                 return index;
  71.             // 如果有时间限制,且时间小于等于0,销毁栅栏,并抛出异常
  72.             if (timed && nanos <= 0L) {
  73.                 breakBarrier();
  74.                 throw new TimeoutException();
  75.             }
  76.         }
  77.     } finally {
  78.         lock.unlock();
  79.     }
  80. }

nextGeneration 方法

  1. /** 生成下一个 generation */
  2. private void nextGeneration(){
  3.     // signal completion of last generation
  4.     // 唤醒所有等待的线程来获取 AQS 的state的值
  5.     trip.signalAll();
  6.     // set up next generation
  7.     // 重新赋值计算器
  8.     count = parties;
  9.     // 重新初始化 generation
  10.     generation = new Generation();
  11. }

代码虽然长，但是主体还是挺清晰，这里简单总结一下方法的流程：
1 .每个 CyclicBarrier 都有一个 Lock，想执行 await 方法，就必须获得这把锁。所以，CyclicBarrier 在并发情况下的性能是不高的。

  1. 一些线程中断的判断，注意，CyclicBarrier 中，只要有一个线程中断了，其余的线程也会抛出中断异常。并且，这个 CyclicBarrier 就不能再次使用了。
  2. 每次线程调用一次 await 方法，表示这个线程到了栅栏这里了，那么就将计数器减一。如果计数器到 0 了，表示这是这一代最后一个线程到达栅栏，就尝试执行我们构造方法中输入的任务。最后，将代更新，计数器重置，并唤醒所有之前等待在栅栏上的线程。
  3. 如果不是最后一个线程到达栅栏了，就使用 Condition 的 await 方法阻塞线程。如果等待过程中，线程中断了，就抛出异常。这里，注意一下，如果中断的线程的使用 CyclicBarrier 不是这代的，比如，在最后一次线程执行 signalAll 后，并且更新了这个“代”对象。在这个区间，这个线程被中断了，那么，JDK 认为任务已经完成了，就不必在乎中断了，只需要打个标记。所以，catch 里的 else 判断用于极少情况下出现的判断——任务完成，“代” 更新了，突然出现了中断。这个时候，CyclicBarrier 是不在乎的。因为任务已经完成了。
  4. 当有一个线程中断了，也会唤醒其他线程，那么就需要判断 broken 状态。
  5. g != generation 所有线程到达 barrier 直接返回

    总结

    从 await 方法看，CyclicBarrier 还是比较简单的，JDK 的思路就是：设置一个计数器，线程每调用一次计数器，就减 1，并使用 Condition 阻塞线程。当计数器是 0 的时候，就唤醒所有线程，并尝试执行构造函数中的任务。由于 CyclicBarrier 是可重复执行的，所以，就需要重置计数器。
    CyclicBarrier 源码分析 - 图1