简单介绍
CyclicBarrier，回环栅栏，它会阻塞一组线程直到这些线程同时达到某个条件才继续执行。它与CountDownLatch很类似，但又不同，CountDownLatch需要调用countDown()方法触发事件，而CyclicBarrier不需要，它就像一个栅栏一样，当一组线程都到达了栅栏处才继续往下走。
public class CyclicBarrierTest {
public static void main(String[] args) {
CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
for (int i = 0; i < 3; i++) {
new Thread(()->{
System.out.println("before");
try {
cyclicBarrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println("after");
}).start();
}
}
}
重要属性
// 重入锁
private final ReentrantLock lock = new ReentrantLock();
// 条件锁，等达到一定数量了再唤醒
private final Condition trip = lock.newCondition();
// 需要等待的线程数量
private final int parties;
// 当唤醒的时候执行的命令
private final Runnable barrierCommand;
// 代
private Generation generation = new Generation();
// 当前这一代还需要等待的线程数
private int count;
通过属性可以看到，CyclicBarrier内部是通过重入锁的条件锁来实现的，其执行的逻辑大致如下：
假如初始时count = parties = 3，当第一个线程到达栅栏处，count减1，然后把它加入到Condition的队列中，第二个线程到达栅栏处也是如此，第三个线程到达栅栏处，count减为0，然后会调用Condition的signalAll()通知另外两个线程，然后把它们加入到AQS的队列中，等待当前线程运行完毕，调用lock.unlock()的时候依次从AQS的队列中唤醒一个线程继续运行，也就是说实际上三个线程先依次（排队）到达栅栏处，再依次往下运行。
上述核心字段中还有个Generation，其源码如下：
private static class Generation {
boolean broken = false;
}
Generation用于控制CyclicBarrier的循环使用。比如，上面示例中的三个线程完成后进入下一代，继续等待三个线程达到栅栏处再一起执行，而CountDownLatch则做不到这一点，CountDownLatch是一次性的，无法重置其次数。
构造函数
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
// 初始化parties
this.parties = parties;
// 初始化count等于parties
this.count = parties;
// 初始化都到达栅栏处执行的命令
this.barrierCommand = barrierAction;
}
public CyclicBarrier(int parties) {
this(parties, null);
}
构造方法需要传入一个parties变量，也就是需要等待的线程数。
重要方法
await()方法
每个需要在栅栏处等待的线程都需要显式地调用await()方法等待其它线程的到来。
public int await() throws InterruptedException, BrokenBarrierException {
try {
// 调用dowait方法，不需要超时
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException, TimeoutException {
final ReentrantLock lock = this.lock;
// 加锁
lock.lock();
try {
// 当前代
final Generation g = generation;
// 检查
if (g.broken)
throw new BrokenBarrierException();
// 中断检查
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
// count的值减1
int index = --count;
// 如果数量减到0了，走这段逻辑（最后一个线程走这里）
if (index == 0) { // tripped
boolean ranAction = false;
try {
// 如果初始化的时候传了命令，这里执行
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
// 调用下一代方法
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}
// 这个循环只有非最后一个线程可以走到
for (;;) {
try {
if (!timed)
// 调用condition的await()方法
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();
// 正常来说这里肯定不相等
// 因为上面打破栅栏的时候调用nextGeneration()方法时generation的引用已经变化了
if (g != generation)
return index;
// 超时检查
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
lock.unlock();
}
}
private void nextGeneration() {
// 调用condition的signalAll()将其队列中的等待者全部转移到AQS的队列中
trip.signalAll();
// 重置count
count = parties;
// 进入下一代
generation = new Generation();
}
private void breakBarrier() {
generation.broken = true;
count = parties;
trip.signalAll();
}
dowait()方法里的整个逻辑分成两部分：
（1）最后一个线程走上面的逻辑的时候，count减为0，打破栅栏，它调用nextGeneration()方法通知条件队列中的等待线程转移到AQS的队列中等待被唤醒，并进入下一代。
（2）非最后一个线程走下面的for循环逻辑，这些线程会阻塞在condition的await()方法处，它们会加入到条件队列中，等待被通知，当它们唤醒的时候返回。
reset
reset方法可以帮助重置 barrier 到初始化状态。代码如下：
public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
breakBarrier(); // break the current generation
nextGeneration(); // start a new generation
} finally {
lock.unlock();
}
}
通过组合 #breakBarrier() 和 #nextGeneration() 方法来实现。
总结
CyclicBarrier实现较为简单，功能上类似于CountDownLatch，只不过CyclicBarrier在每一轮控制之后可以自动重新对一组线程进行控制。