????CyclicBarrier翻譯過來是“可循環(huán)利用的屏障“，CyclicBarrier 作用是讓一組線程相互等待，當達到一個共同點時，所有之前等待的線程再繼續(xù)執(zhí)行，且 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()+"已經(jīng)到達檢查點");
});
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()+"已經(jīng)到達屏障");
cyclicBarrier.await();
System.out.println(Thread.currentThread().getName()+"已經(jīng)沖破屏障");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
});
}
threadPoolExecutor.shutdown();
}
}
執(zhí)行結果
pool-1-thread-2已經(jīng)到達屏障
pool-1-thread-4已經(jīng)到達屏障
pool-1-thread-3已經(jīng)到達屏障
pool-1-thread-6已經(jīng)到達屏障
pool-1-thread-1已經(jīng)到達屏障
pool-1-thread-5已經(jīng)到達屏障
pool-1-thread-7已經(jīng)到達屏障
pool-1-thread-8已經(jīng)到達屏障
pool-1-thread-9已經(jīng)到達屏障
pool-1-thread-10已經(jīng)到達屏障
pool-1-thread-10已經(jīng)到達檢查點
pool-1-thread-10已經(jīng)沖破屏障
pool-1-thread-2已經(jīng)沖破屏障
pool-1-thread-3已經(jīng)沖破屏障
pool-1-thread-4已經(jīng)沖破屏障
pool-1-thread-9已經(jīng)沖破屏障
pool-1-thread-8已經(jīng)沖破屏障
pool-1-thread-7已經(jīng)沖破屏障
pool-1-thread-5已經(jīng)沖破屏障
pool-1-thread-1已經(jīng)沖破屏障
pool-1-thread-6已經(jīng)沖破屏障
主要結構
public class CyclicBarrier {
// CyclicBarrier是可復用的，Generation用于標記更新?lián)Q代
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.
*/
// 可以簡單理解為還未到達屏障的成員數(shù)
private int count;
}
構造函數(shù)
// parties：屏障解除之前必須調(diào)用await方法的線程數(shù)
// barrierAction：屏障解除時執(zhí)行的動作
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
到達屏障
???? 等待直到所有的參與者在此屏障上調(diào)用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();
// 線程被中斷，調(diào)用breakBarrier打破屏障
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
int index = --count;
// 說明該線程是最后一個到達的線程
if (index == 0) { // tripped
// 標志任務是否執(zhí)行成功
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
// 任務不是空的話執(zhí)行任務
if (command != null)
command.run();
ranAction = true;
// 所有線程都到達屏障了，開啟下一代
nextGeneration();
return 0;
} finally {
// 任務執(zhí)行失敗，打破屏障
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();
// 已經(jīng)更新?lián)Q代過了，說明最后一個線程也到達了
// 那么直接返回當前線程的到達屏障時的索引就行
if (g != generation)
return index;
// 如果設置了超時時間，并且超時時間小于0，則打破屏障，并拋出TimeoutException
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
lock.unlock();
}
}
????屏障被打破：是指調(diào)用breakBarrier 方法把broken被置為true，線程被中斷、barrierCommand任務執(zhí)行失敗、超時、調(diào)用reset方法都會打破屏障
復位屏障
????將屏障重置為其初始狀態(tài)。如果任何一方當前正在等待障礙，他們將拋出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
// 初始化等待的成員數(shù)
count = parties;
// new一個新Generation
generation = new Generation();
}
打破屏障
private void breakBarrier() {
// 設置broken標志位
generation.broken = true;
count = parties;
// 喚醒在trip條件等待隊列中的所有線程
trip.signalAll();
}
