多线程之CyclicBarrier详解（多线程accept）

上篇我们介绍了一下 多线程之CountDownLatch详解.，这篇我们详细的介绍CyclicBarrier。

什么是CyclicBarrier

CyclicBarrier的字面意思是可循环（Cyclic）使用的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会打开，所有被屏障拦截的线程才会继续干活，线程进入屏障通过CyclicBarrier的await()方法。它也是AQS多线程同步操作的一个具体实现。

怎么使用CyclicBarrier

从上图我们可以看到，CyclicBarrier主要有两个构造方法，如下：

public CyclicBarrier(int parties)
public CyclicBarrier(int parties, Runnable barrierAction)

解析：

parties 是参与线程的个数

第二个构造方法有一个 Runnable 参数，这个参数的意思是最后一个到达线程到达屏障，开始做屏障自己的任务

核心的两个方法如下：

public int await() throws InterruptedException, BrokenBarrierException
public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException

解析：

线程调用 await() 表示自己已经到达栅栏

BrokenBarrierException 表示栅栏已经被破坏，破坏的原因可能是其中一个线程 await() 时被中断或者超时

源码解读

public int await() throws InterruptedException, BrokenBarrierException {
 try {
 return dowait(false, 0L);
 } catch (TimeoutException toe) {
 throw new Error(toe); // cannot happen
 }
}
public int await(long timeout, TimeUnit unit)
 throws InterruptedException,
 BrokenBarrierException,
 TimeoutException {
 return dowait(true, unit.toNanos(timeout));
}
private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException {
 //获取独占锁
 final ReentrantLock lock = this.lock;
 lock.lock();
 try {
 //当前代
 final Generation g = generation;
 // 如果这代损坏了，抛出BrokenBarrierException。
 if (g.broken)
 throw new BrokenBarrierException();
 // 检查中断标志位。
 if (Thread.interrupted()) {
 breakBarrier();
 throw new InterruptedException();
 }
 int index = --count;
 // 如果是0，最后一个到达barrier的线程。
 if (index == 0) {
 boolean ranAction = false;
 try {
 final Runnable command = barrierCommand;
 //执行栅栏任务
 if (command != null)
 command.run();
 ranAction = true;
 //更新一代，将count重置，将generation重置，并唤醒之前等待的线程
 // 开启下一轮。
 nextGeneration();
 return 0;
 } finally {
 // 如果执行栅栏任务时失败了，也会break掉barrier。
 if (!ranAction)
 breakBarrier();
 }
 }
 // loop until tripped, broken, interrupted, or timed out
 /*
 * 对于其它（不是最后一个）线程，会在trip条件下等待被唤醒。情况有以下几类：
 * 1. 所有线程都到达barrier，并成功执行了barrierAction。
 * 2. 有线程执行了breakBarrier方法。
 * 3. 线程本身被中断。
 * 4. 超时（如果调用的带时间限制的await）。
 */
 for (;;) {
 try {
 if (!timed)
 trip.await();
 else if (nanos > 0L)
 nanos = trip.awaitNanos(nanos);
 } catch (InterruptedException ie) {
 /*
 * g == generation && !g.broken说明此时当前这一轮还没结束，并且没有其它线程执行过breakBarrier方法。
 * 这种情况会执行breakBarrier置generation的broken标识为true并唤醒其它线程，之后继续抛出InterruptedException。
 */
 if (g == generation && ! g.broken) {
 breakBarrier();
 throw ie;
 } else {
 /*
 * 如果g != generation,此时这一轮已经结束，返回当前线程所在栅栏的下标;
 * 如果g.broken说明之前已经有其它线程执行了breakBarrier方法，后面会抛出BrokenBarrierException。
 */
 Thread.currentThread().interrupt();
 }
 }
 if (g.broken)
 throw new BrokenBarrierException();
 // 这一轮已经结束，则返回到达屏障的次序，0表示最后一个，parties-1表示第一个。
 // 如果 g == generation，说明还没有换代，那为什么会醒了？
 // 因为一个线程可以使用多个栅栏，当别的栅栏唤醒了这个线程，就会走到这里，所以需要判断是否是当前代。
 // 正是因为这个原因，才需要generation来保证正确。
 if (g != generation)
 return index;
 // 判断是否超时。
 if (timed && nanos <= 0L) {
 breakBarrier();
 throw new TimeoutException();
 }
 }
 } finally {
 lock.unlock();
 }
}

dowait(boolean, long)方法的主要逻辑处理比较简单，如果该线程不是最后一个调用await方法的线程，则它会一直处于等待状态，除非发生以下情况：

• 最后一个线程到达，即index == 0
• 某个参与线程等待超时
• 某个参与线程被中断
• 调用了CyclicBarrier的reset()方法。该方法会将屏障重置为初始状态

在上面的源代码中，我们可能需要注意Generation 对象，在上述代码中我们总是可以看到抛出BrokenBarrierException异常，那么什么时候抛出异常呢？如果一个线程处于等待状态时，如果其他线程调用reset()，或者调用的barrier原本就是被损坏的，则抛出BrokenBarrierException异常。同时，任何线程在等待时被中断了，则其他所有线程都将抛出BrokenBarrierException异常，并将barrier置于损坏状态。

同时，Generation描述着CyclicBarrier的更新换代。在CyclicBarrier中，同一批线程属于同一代。当有parties个线程到达barrier之后，generation就会被更新换代。其中broken标识该当前CyclicBarrier是否已经处于中断状态。

private static class Generation {
 boolean broken = false;
}

默认barrier是没有损坏的。当barrier损坏了或者有一个线程中断了，则通过breakBarrier()来终止所有的线程：

// 保证调用时持有锁。

private void breakBarrier() {
 generation.broken = true;
 // 重置count。
 count = parties;
 // 唤醒其它在trip条件下等待的线程。
 trip.signalAll();
}

在breakBarrier()中除了将broken设置为true，还会调用signalAll将在CyclicBarrier处于等待状态的线程全部唤醒。

当所有线程都已经到达barrier处（index == 0），则会通过nextGeneration()进行更新换代操作，在这个步骤中，做了三件事：唤醒所有线程，重置count，换代generation：

// 保证调用时持有锁。
private void nextGeneration() {
 // 唤醒其它在trip条件下等待的线程。
 trip.signalAll();
 // 重置count。
 count = parties;
 // 开启下一轮。
 generation = new Generation();
}

除了上面讲到的栅栏更新换代以及损坏状态，我们在使用CyclicBarrier时还要要注意以下几点：

CyclicBarrier使用独占锁来执行await方法，并发性可能不是很高

如果在等待过程中，线程被中断了，就抛出异常。但如果中断的线程所对应的CyclicBarrier不是这代的，比如，在最后一次线程执行signalAll后，并且更新了这个“代”对象。在这个区间，这个线程被中断了，那么，JDK认为任务已经完成了，就不必在乎中断了，只需要打个标记。该部分源码已在dowait(boolean, long)方法中进行了注释。

如果线程被其他的CyclicBarrier唤醒了，那么g肯定等于generation，这个事件就不能return了，而是继续循环阻塞。反之，如果是当前CyclicBarrier唤醒的，就返回线程在CyclicBarrier的下标。完成了一次冲过栅栏的过程。该部分源码已在dowait(boolean, long)方法中进行了注释。

CountDownLatch应用demo

具体需求，多个线程并发操作，执行某个任务，等待所有线程完成任务后再继续执行下一个任务

执行结果如下：

常见应用场景

用于多线程计算数据，最后合并计算结果的场景。每个parter负责一部分计算，最后的线程barrierAction线程进行数据汇总。

CyclicBarrier 与 CountDownLatch 区别

1. CountDownLatch是线程组之间的等待，即一个(或多个)线程等待N个线程完成某件事情之后再执行；而CyclicBarrier则是线程组内的等待，即每个线程相互等待，即N个线程都被拦截之后，然后依次执行。
2. CountDownLatch是减计数方式，而CyclicBarrier是加计数方式。
3. CountDownLatch计数为0无法重置，而CyclicBarrier计数达到初始值，则可以重置。
4. CountDownLatch不可以复用，而CyclicBarrier可以复用。

总结

CyclicBarrier的await方法是使用ReentrantLock和Condition控制实现的，使用的Condition实现类是ConditionObject，它里面有一个等待队列和await方法，这个await方法会向队列中加入元素。当调用CyclicBarrier的await方法会间接调用ConditionObject的await方法，当屏障关闭后首先执行指定的barrierAction，然后依次执行等待队列中的任务，有先后顺序。

欢迎大家评论，收藏+转发，谢谢。