CyclicBarrier简介
CyclicBarrier是一个同步辅助类，允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。因为该 barrier 在释放等待线程后可以重用，所以称它为循环 的 barrier。
CyclicBarrier函数列表
// 创建一个新的 CyclicBarrier，它将在给定数量的线程处于等待状态时启动，但它不会在启动 barrier 时执行预定义的操作。 public CyclicBarrier(int parties) {this(parties, null);}// 创建一个新的 CyclicBarrier，它将在给定数量的线程处于等待状态时启动，并在启动 barrier 时执行给定的屏障操作，该操作由最后一个进入 barrier 的线程执行public CyclicBarrier(int parties, Runnable barrierAction) {if (parties <= 0) throw new IllegalArgumentException();this.parties = parties;this.count = parties;this.barrierCommand = barrierAction;}
// 在所有线程都已经在此 barrier 上调用 await 方法之前，将一直等待。public int await() throws InterruptedException, BrokenBarrierException {try {return dowait(false, 0L);} catch (TimeoutException toe) {throw new Error(toe); // cannot happen}}
// 在所有参与者都已经在此 barrier 上调用 await 方法之前，将一直等待，或者超出时间public int await(long timeout, TimeUnit unit)throws InterruptedException,BrokenBarrierException,TimeoutException {return dowait(true, unit.toNanos(timeout));}// 返回 barrier 中等待的数量public int getNumberWaiting() {final ReentrantLock lock = this.lock;lock.lock();try {return parties - count;} finally {lock.unlock();}}// 返回 barrier 是否损坏public boolean isBroken() {final ReentrantLock lock = this.lock;lock.lock();try {return generation.broken;} finally {lock.unlock();}}
// 重置barrierpublic void reset() {final ReentrantLock lock = this.lock;lock.lock();try {breakBarrier(); // break the current generationnextGeneration(); // start a new generation} finally {lock.unlock();}}
// 返回要求启动此 barrier 的线程数量public int getParties() {return parties;}
CyclicBarrier数据结构
CyclicBarrier是包含了"ReentrantLock对象lock"和"Condition对象trip"，它是通过独占锁实现的
/** The lock for guarding barrier entry */private final ReentrantLock lock = new ReentrantLock();/** Condition to wait on until tripped */private final Condition trip = lock.newCondition();
CyclicBarrier源码分析
1. 构造函数
CyclicBarrier的构造函数共2个：CyclicBarrier 和 CyclicBarrier(int parties, Runnable barrierAction)。第1个构造函数是调用第2个构造函数来实现的
public CyclicBarrier(int parties) {this(parties, null);}
public CyclicBarrier(int parties, Runnable barrierAction) {if (parties <= 0) throw new IllegalArgumentException();// parties表示必须同时到达barrier的线程个数this.parties = parties;// count表示处在等待状态的线程个数this.count = parties;// barrierCommand表示parties个线程到达barrier时，会执行的动作this.barrierCommand = barrierAction;}
2. 等待函数
await()是通过dowait()实现的。
public int await() throws InterruptedException, BrokenBarrierException {try {return dowait(false, 0L);} catch (TimeoutException toe) {throw new Error(toe); // cannot happen}}
dowait() 源码：
private int dowait(boolean timed, long nanos)throws InterruptedException, BrokenBarrierException,TimeoutException {final ReentrantLock lock = this.lock;// 获取独占锁locklock.lock();try {//保存当期generationfinal Generation g = generation;// 如果当前generation false 则抛出异常if (g.broken)throw new BrokenBarrierException();// 如果当前线程被中断，则通过breakBarrier()终止CyclicBarrier，唤醒 CyclicBarrier中所有等待线程if (Thread.interrupted()) {breakBarrier();throw new InterruptedException();}// 将count 计数器减一int index = --count;// 如果index = 0 则parties个线程到barrierif (index == 0) { // trippedboolean ranAction = false;try {final Runnable command = barrierCommand;// 如果barrierCommand不为null，则执行该动作if (command != null)command.run();ranAction = true;// 唤醒所有等待线程，并更新generationnextGeneration();return 0;} finally {if (!ranAction)breakBarrier();}}// loop until tripped, broken, interrupted, or timed out// 当前线程一直阻塞，知道到达barrier个数，或者打断，或者 超时for (;;) {try {// 如果不是超时等待，则嗲用await等待，否则调用awaitNanos等待if (!timed)trip.await();else if (nanos > 0L)nanos = trip.awaitNanos(nanos);} catch (InterruptedException ie) {if (g == generation && ! g.broken) {breakBarrier();throw ie;} else {// We're about to finish waiting even if we had not// been interrupted, so this interrupt is deemed to// "belong" to subsequent execution.Thread.currentThread().interrupt();}}if (g.broken)throw new BrokenBarrierException();if (g != generation)return index;if (timed && nanos <= 0L) {breakBarrier();throw new TimeoutException();}}} finally {//释放独占锁lock.unlock();}}
说明：dowait()的作用就是让当前线程阻塞，直到“有parties个线程到达barrier” 或 “当前线程被中断” 或 “超时”这3者之一发生，当前线程才继续执行。
(01) generation是CyclicBarrier的一个成员遍历，它的定义如下：
private Generation generation = new Generation();private static class Generation {boolean broken = false;
}
在CyclicBarrier中，同一批的线程属于同一代，即同一个Generation；CyclicBarrier中通过generation对象，记录属于哪一代。
当有parties个线程到达barrier，generation就会被更新换代。
(02) 如果当前线程被中断，即Thread.interrupted()为true；则通过breakBarrier()终止CyclicBarrier。breakBarrier()的源码如下：
private void breakBarrier() {generation.broken = true;count = parties;trip.signalAll();
}
breakBarrier()会设置当前中断标记broken为true，意味着“将该Generation中断”；同时，设置count=parties，即重新初始化count；最后，通过signalAll()唤醒CyclicBarrier上所有的等待线程。
(03) 将“count计数器”-1，即--count；然后判断是不是“有parties个线程到达barrier”，即index是不是为0。
当index=0时，如果barrierCommand不为null，则执行该barrierCommand，barrierCommand就是我们创建CyclicBarrier时，传入的Runnable对象。然后，调用nextGeneration()进行换代工作，nextGeneration()的源码如下：
private void nextGeneration() {trip.signalAll();count = parties;generation = new Generation();
}
首先，它会调用signalAll()唤醒CyclicBarrier上所有的等待线程；接着，重新初始化count；最后，更新generation的值。
(04) 在for(;;)循环中。timed是用来表示当前是不是“超时等待”线程。如果不是，则通过trip.await()进行等待；否则，调用awaitNanos()进行超时等待。
CyclicBarrier的使用
public class Test {static class TaskThread extends Thread {CyclicBarrier barrier;public TaskThread(CyclicBarrier barrier) {this.barrier = barrier;}@Overridepublic void run() {try {Thread.sleep(1000);System.out.println(getName() + " 到达栅栏 A");barrier.await();System.out.println(getName() + " 冲破栅栏 A");} catch (Exception e) {e.printStackTrace();}}}public static void main(String[] args) {int threadNum = 5;CyclicBarrier barrier = new CyclicBarrier(threadNum, new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName() + " 完成最后任务");}});for(int i = 0; i < threadNum; i++) {new TaskThread(barrier).start();}}
}