版权声明: https://blog.csdn.net/ph3636/article/details/84970598
1. CyclicBarrier类似可重复使用的CountDownLatch,可以允许多个线程执行完后才可以执行某一操作,并且可以循环使用。第一个变量是允许线程数,第二个是所有线程执行完后最后需要执行的任务,线程执行时只会递减count值,循环使用时复原状态需要用parties重新设置count值。
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
2.线程执行await操作,当所有的线程执行完该操作后,整个流程才算是一次完整执行。
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,主要使用ReentrantLock锁,首先判断broken状态是否被强制打断。继而判断是否线程被中断,中断后则通知所有的线程,并且设置打断状态broken为true,它的初始值为false。然后给总数进行减一操作,当所有线程执行到该操作时,count值会为0,开始执行最后的任务,通知所有阻塞的线程,并且开始复原所有状态准备下一次循环使用。如果count不为0,也就是并不是所有线程都执行了dowait操作时,会进入等待操作trip.await(),当所有线程执行完时会通知等待的线程,唤醒后判断broken状态,最后g != generation为true,返回线程的执行顺序。如果是超时等待的话,会判断状态是否返回超时异常。
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();
}
int index = --count;
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 {
// 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();
}
}
3. 设置下一次循环的代码如下。
private void nextGeneration() {
// signal completion of last generation
trip.signalAll();
// set up next generation
count = parties;
generation = new Generation();
}