CyclicBarrier是一个同步器，允许一组线程相互之间等待，直到到达某个公共屏障点 (common barrier point)，再继续执行
CyclicBarrier 与 CountDownLatch异同
都可以阻塞一组线程等待被唤醒
CyclicBarrier是最后一个线程到达后会自动唤醒，而CountDownLatch需要显式调用countDown方法
CyclicBarrier基于可重入锁和条件队列实现，CountDownLatch基于AQS实现
CyclicBarrier可以重复使用，而CountDownLatch只能使用一次
前面看过CountdownLatch的源码我们知道，CountdownLatch 利用的共享锁特性；主线程调用await()会阻塞当前线程；子线程通过调用countdown()方法来减少state的值，直到state为0时，等待队列中的线程被释放。结合 CyclicBarrier的特性，很自然的能知道其实现的原理，当state 为0是，重新将参数重新设置到state中即可
源码解析
成员变量
// 可重入锁ReentrantLock，条件队列基于ReentrantLock创建
private final ReentrantLock lock = new ReentrantLock();
// 一个条件队列
private final Condition trip = lock.newCondition();
// 代表需要等待多少个线程，我们在使用时一般使用count，只有初始化时才会使用parties，将parties赋值
// 给count，这也达到循环的目的
private final int parties;
// 代表最后一个线程到达后需要执行的任务
private final Runnable barrierCommand;
// 一个Generation内部类，当前代对象
private Generation generation = new Generation();
// 代表当前代还需要等待多少个线程
private int count;
内部类
// 这个是一个简简单单的内部类里面只有一个boolean变量，代表当前“代”是否被打破
private static class Generation {
// 只有一个变量broken,表示当前代是否被打破，如果代被打破，则再来到这一代的线程，就会直接抛出BrokenException异常
// 在这一代的线程会被唤醒（加入AQS队列），然后抛出BrokenException异常
boolean broken = false;
}
构造方法
// parties表示需要等待的线程数量，barrierAction代表最后一个线程到达后需要执行的任务
public CyclicBarrier(int parties, Runnable barrierAction) {
// 小于等于0没有意义，直接抛出异常
if (parties <= 0) throw new IllegalArgumentException();
// 代表需要等待的线程数量
this.parties = parties;
// 代表当前代需要等待的线程数量
this.count = parties;
// 最后一个线程到达后需要执行的任务
this.barrierCommand = barrierAction;
}
// parties表示需要等待的线程数量，这个构造方法最后一个线程到达后不需要执行额外的任务
public CyclicBarrier(int parties) {
// 调用另一个构造方法
this(parties, null);
}
成员方法
await方法
public int await() throws InterruptedException, BrokenBarrierException {
try {
// 调用dowait方法
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
dowait方法
// nanos：单位为纳秒，表示等待时长
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
// 获取ReentrantLock锁
final ReentrantLock lock = this.lock;
// 加锁
lock.lock();
try {
// 当前代对象
final Generation g = generation;
// 如果当前代被打破，则调用await方法之后进入dowait方法，抛出BrokenBarrierException异常
if (g.broken)
throw new BrokenBarrierException();
// 中断检查
if (Thread.interrupted()) {
// 设置broken为true，然后唤醒（加入AQS队列）trip中的对象
breakBarrier();
throw new InterruptedException();
}
// 获取count-1的值
int index = --count;
// 如果count-1 == 0，要看看Runnable是否为空（最后一个线程到达后是否有任务要执行）
if (index == 0) { // tripped
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
// 最后一个线程走这里，调用nextGeneration方法，开启下一代
nextGeneration();
return 0;
} finally {
// 如果command.run()抛出异常，会执行breakBarrier方法
if (!ranAction)
breakBarrier();
}
}
// 只有不是最后一个的线程才可以执行到这里
// 自旋，一直到条件满足：当前代被打破，线程被中断，等待超时
// loop until tripped, broken, interrupted, or timed out
for (;;) {
try {
// 说明不指定超时时间，调用await方法，之后await又会走到这里，自旋
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();
}
}
breakBarrier方法
// 打破屏障
private void breakBarrier() {
// 设置当前代被打破
generation.broken = true;
// 重置count的值
count = parties;
// 将当前代所有线程假如AQS队列
trip.signalAll();
}
nextGeneration方法
// 当最后一个线程到达，dowait方法内部会调用这个方法开启下一代
private void nextGeneration() {
// signal completion of last generation
// 将当前代所有对象加入AQS队列
trip.signalAll();
// set up next generation
// 重置count的值
count = parties;
// 新new一个Generation对象
generation = new Generation();
}
总结
- CyclicBarrier在调用await后，在最后一个线程到达前会一直阻塞在await处，只有当最后一个线程到达后，才会将当前代所有线程加入AQS队列
- CyclicBarrier基于ReentrantLock和Condition实现，并没有使用AQS