CyclicBarrier:(基于Condition)
让一组线程到达一个屏障(也可以叫同步点)时被阻塞,直到最后一个线程到达屏障时,屏障才会开门,所有被屏障拦截的线程才会继续运行。
原理
在CyclicBarrier类的内部有一个计数器,每个线程在到达屏障点的时候都会调用await方法将自己阻塞,此时计数器会减1,当计数器减为0的时候所有因调用await方法而被阻塞的线程将被唤醒。
成员变量
private static class Generation {
boolean broken = false;
}
/** The lock for guarding barrier entry 同步操作锁*/
private final ReentrantLock lock = new ReentrantLock();
/** Condition to wait on until tripped 线程拦截器*/
private final Condition trip = lock.newCondition();
/** The number of parties 每次拦截的线程数*/
private final int parties;
/* The command to run when tripped 换代前执行的任务*/
private final Runnable barrierCommand;
/** The current generation 栅栏的当前代*/
private Generation generation = new Generation();
/**
* Number of parties still waiting. Counts down from parties to 0
* on each generation. It is reset to parties on each new
* generation or when broken. 计数器
*/
private int count;
构造函数
public CyclicBarrier(int parties) {
this(parties, null);
}
/** barrierAction由最后一个到达栅栏的线程执行*/
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
await()、await(long timeout, TimeUnit unit)
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;
// 检查当前屏障是否被打破
if (g.broken)
throw new BrokenBarrierException();
/** 如果当前线程被中断会做以下三件事
* 1、打破当前屏障
* 2、唤醒拦截的所有线程
* 3、抛出中断异常
*/
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
// 计数器值-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();
}
}
// loop until tripped, broken, interrupted, or timed out
for (;;) {
try {
if (!timed)
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
// 如果当前线程在等待期间被中断则打破屏障唤醒其他线程
if (g == generation && ! g.broken) {
breakBarrier();
throw ie;
} else {
// 如果在捕获中断异常前已经完成了屏障上的等待,则直接调用线程中断
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();
}
}
// 转化到下一代
private void nextGeneration() {
// signal completion of last generation
trip.signalAll();
// set up next generation
count = parties;
generation = new Generation();
}
// 打破屏障
private void breakBarrier() {
generation.broken = true;
count = parties;
trip.signalAll();
}
// 重置
public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
breakBarrier(); // break the current generation
nextGeneration(); // start a new generation
} finally {
lock.unlock();
}
}