CyclicBarrier翻译过来是“可循环利用的屏障“，CyclicBarrier 作用是让一组线程相互等待，当达到一个共同点时，所有之前等待的线程再继续执行，且 CyclicBarrier 功能可重复使用。
image.png
代码示例
public class CyclicBarrierDemo {
public static void main(String[] args) {
TimeUnit unit = TimeUnit.SECONDS;
BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(20);
int corePoolSize = 10;
int maximumPoolSize = 10;
long keepAliveTime = 60L;
CyclicBarrier cyclicBarrier = new CyclicBarrier(10, ()-> {
System.out.println(Thread.currentThread().getName()+"已经到达检查点");
});
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
for(int i=0; i<10; i++) {
threadPoolExecutor.submit(() -> {
try {
System.out.println(Thread.currentThread().getName()+"已经到达屏障");
cyclicBarrier.await();
System.out.println(Thread.currentThread().getName()+"已经冲破屏障");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
});
}
threadPoolExecutor.shutdown();
}
}
执行结果
pool-1-thread-2已经到达屏障
pool-1-thread-4已经到达屏障
pool-1-thread-3已经到达屏障
pool-1-thread-6已经到达屏障
pool-1-thread-1已经到达屏障
pool-1-thread-5已经到达屏障
pool-1-thread-7已经到达屏障
pool-1-thread-8已经到达屏障
pool-1-thread-9已经到达屏障
pool-1-thread-10已经到达屏障
pool-1-thread-10已经到达检查点
pool-1-thread-10已经冲破屏障
pool-1-thread-2已经冲破屏障
pool-1-thread-3已经冲破屏障
pool-1-thread-4已经冲破屏障
pool-1-thread-9已经冲破屏障
pool-1-thread-8已经冲破屏障
pool-1-thread-7已经冲破屏障
pool-1-thread-5已经冲破屏障
pool-1-thread-1已经冲破屏障
pool-1-thread-6已经冲破屏障
主要结构
public class CyclicBarrier {
// CyclicBarrier是可复用的，Generation用于标记更新换代
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;
}
构造函数
// parties：屏障解除之前必须调用await方法的线程数
// barrierAction：屏障解除时执行的动作
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
到达屏障
等待直到所有的参与者在此屏障上调用await
返回值：当前线程的到达索引，0表示最后一个到达
public int await() throws InterruptedException, BrokenBarrierException {
try {
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();
// 线程被中断，调用breakBarrier打破屏障
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();
// 设置了超时时间，当前线程等到nanos时间
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
// 线程阻塞期间被中断，朝代没有换，屏障没有被打破，则breakBarrier打破屏障
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();
}
}
// 线程被唤醒，如果屏障被打断则抛出BrokenBarrierException
if (g.broken)
throw new BrokenBarrierException();
// 已经更新换代过了，说明最后一个线程也到达了
// 那么直接返回当前线程的到达屏障时的索引就行
if (g != generation)
return index;
// 如果设置了超时时间，并且超时时间小于0，则打破屏障，并抛出TimeoutException
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
lock.unlock();
}
}
屏障被打破：是指调用breakBarrier 方法把broken被置为true，线程被中断、barrierCommand任务执行失败、超时、调用reset方法都会打破屏障
复位屏障
将屏障重置为其初始状态。如果任何一方当前正在等待障碍，他们将抛出BrokenBarrierException
public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 在当前代打破屏障
breakBarrier(); // break the current generation
// 开启新代
nextGeneration(); // start a new generation
} finally {
lock.unlock();
}
}
开启新代
private void nextGeneration() {
// 唤醒在trip条件等待队列中的所有线程
trip.signalAll();
// set up next generation
// 初始化等待的成员数
count = parties;
// new一个新Generation
generation = new Generation();
}
打破屏障
private void breakBarrier() {
// 设置broken标志位
generation.broken = true;
count = parties;
// 唤醒在trip条件等待队列中的所有线程
trip.signalAll();
}
网友评论