Jdk1.6 JUC源码解析(11)-CyclicBarrier
作者：大飞
功能简介：
- CyclicBarrier是一种可重复使用的栅栏机制，可以让一组线程在某个点上相互等待，这个点就可以类比为栅栏。并且这个栅栏是可重复使用的，这点可以和前面分析过的CountDownLatch做对比，CountDownLatch只能用一次。
- CyclicBarrier还支持在所有线程到达栅栏之后，在所有线程从等待状态转到可运行状态之前，执行一个命令(或者说是动作)。
- 当然，在某些情况下，栅栏可以被打破。比如某个线程无法在规定的时间内到达栅栏。
源码分析：
- 先看下CyclicBarrier内部的结构：
public class CyclicBarrier {
/**
* 每次对栅栏的使用可以表示为一个generation。栅栏每次开放或者重置，
* generation都会发生改变。使用栅栏的线程可以关联多个generations，
* 由于等待线程可能会以多种方式请求锁，但是在特定的时间只有一个是
* 可用的，其他的要么被打破，要么开放。
* 如果一个栅栏已经被打破。且没有后续的重置动作，那么可以不存在可
* 用的generation。
*/
private static class Generation {
boolean broken = false;
}
/** 用于保护栅栏的锁 */
private final ReentrantLock lock = new ReentrantLock();
/** 栅栏开放的条件 */
private final Condition trip = lock.newCondition();
/** 表示当前使用栅栏的使用方(线程)数量 */
private final int parties;
/* 当栅栏开放时，要使用的命令(动作) */
private final Runnable barrierCommand;
/** The current generation */
private Generation generation = new Generation();
/**
* 处于等待状态的使用方(线程)的数量，在每一个generation上从
* parties递减为0。当新建generation(栅栏开放)或者栅栏被打破
* 时，重置为parties。
*/
private int count;从上述的结构细节中可见，CyclicBarrier内部使用ReentrantLock来实现，并包含一个trip条件，来作为栅栏模拟栅栏的行为(所有使用方都在这个条件上等待)。
- 具体使用栅栏时，各个线程会在要互相等待的地方调用一个”等待”方法，然后在这个方法处等待。当所有线程都到达次方法时，栅栏打开，所有线程从等待出继续执行。接下来就从这个”等待”方法入手开始分析：
/**
* 线程调用此方法后等待，直到所有parties都调用当前barrier的这个方法。
*
* 如果当前线程是不最后一个到达此方法的线程，那么会阻塞，直到下面的
* 事情发生：
*
* 最后的线程也到达此方法。
* 其他线程中断了当前线程。
* 其他线程中断了在栅栏处等待的某个线程。
* 某个线程调用了栅栏的reset方法。
*/
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));
}
看下上述等待方法中调用的dowait方法：
/**
* 栅栏主体代码，涵盖了所有情况。
*/
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
//获取当前的generation实例。
final Generation g = generation;
if (g.broken) //如果当前generation状态为broken，说明栅栏被打破，抛出BrokenBarrierException异常。
throw new BrokenBarrierException();
if (Thread.interrupted()) {
//如果当前线程被中断，打破栅栏，然后抛出中断异常。
breakBarrier();
throw new InterruptedException();
}
//计算当前到达线程的下标。
int index = --count;
//下标为0表示当前线程为最后一个使用栅栏的线程。
if (index == 0) { // 栅栏开放。
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
if (command != null)//如果有栅栏命令，执行栅栏命令。
command.run();//看来栅栏的命令是由最后一个到达栅栏的线程执行。
ranAction = true;
//产生新的generation。
nextGeneration();
return 0;
} finally {
if (!ranAction) //如果栅栏命令未执行，打破栅栏。
breakBarrier();
}
}
// 等待中的主循环，直到栅栏开放、栅栏被打破、线程被打断或者超时时退出。
for (;;) {
try {
if (!timed)
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
if (g == generation && ! g.broken) {
//如果出于当前generation 且generation状态为未打破，那么打破栅栏。
breakBarrier();
throw ie;
} else {
// 如果没被中断的话，我们即将完成等待。
// 所以这个中断被算作下一次执行的中断。
Thread.currentThread().interrupt();
}
}
if (g.broken)
throw new BrokenBarrierException();
if (g != generation)//如果generation改变了，说明之前的栅栏已经开放，返回index。
return index;
if (timed && nanos <= 0L) {
breakBarrier();//如果超时，打破栅栏，并返回超时异常。
throw new TimeoutException();
}
}
} finally {
lock.unlock();
}
}
从dowait方法中可以看到，当所有使用者都到达时，栅栏开放，会调用nextGeneration方法；如果有其他情况(超时、中断等)发生，会调用breakBarrier方法。先看下nextGeneration：
/**
* 更新栅栏状态，唤醒所有在栅栏处等待的线程。
* 这个方法只有在持有锁的情况下被调用。
*/
private void nextGeneration() {
//唤醒所有在栅栏处等待的线程。
trip.signalAll();
//重置count。
count = parties;
//产生新的generation。
generation = new Generation();
}
再看下breakBarrier方法：
/**
* 设置当前栅栏generation状态为打破状态，并唤醒栅栏处的等待线程。
* 这个方法只有在持有锁的情况下被调用。
*/
private void breakBarrier() {
//设置"打破"状态。
generation.broken = true;
//重置count。
count = parties;
//唤醒所有在栅栏处等待的线程。
trip.signalAll();
}
dowait内部为栅栏的主要逻辑，经过上面的分析，应该很清晰了。最后看下其他的方法：
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
public CyclicBarrier(int parties) {
this(parties, null);
}
public int getParties() {
return parties;
}
public boolean isBroken() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return generation.broken;
} finally {
lock.unlock();
}
}
public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
//重置逻辑，先打破当前的栅栏，然后建立一个新的。
breakBarrier();
nextGeneration();
} finally {
lock.unlock();
}
}
/**
* 获取在栅栏处等待的使用方(线程)数量。
*/
public int getNumberWaiting() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return parties - count;
} finally {
lock.unlock();
}
}
小总结一下：
1.当建立一个使用方数量为n的栅栏时，栅栏内部有一个为n的计数。当使用方调用await方法时，如果其他n-1个使用方没有全部到达await方法(内部计数减1后，不等于0)，那么使用方(线程)阻塞等待。
2.当第n个使用方调用await时，栅栏开放(内部计数减1后等于0)，会唤醒所有在await方法上等待着的使用方(线程)，大家一起通过栅栏，然后重置栅栏(内部计数又变成n)，栅栏变成新建后的状态，可以再次使用。
CyclicBarrier的代码解析完毕！ 参见：
Jdk1.6 JUC源码解析(7)-locks-ReentrantLock
