二十八、併發程式設計之併發工具類CyclicBarrier詳解
阿新 • • 發佈：2019-01-08
一、概述
CyclicBarrier是一個同步工具類，它允許一組執行緒互相等待，直到到達某個公共屏障點。與CountDownLatch不同的是該barrier在釋放等待執行緒後可以重用，所以稱它為迴圈（Cyclic）的屏障（Barrier）。
CyclicBarrier支援一個可選的Runnable命令，在一組執行緒中的最後一個執行緒到達之後（但在釋放所有執行緒之前），該命令只在每個屏障點執行一次。若在繼續所有參與執行緒之前更新共享狀態，此屏障操作很有用。
二、提供的方法：
//parties表示屏障攔截的執行緒數量，當屏障撤銷時，先執行barrierAction，然後在釋放所有執行緒 對於失敗的同步嘗試，CyclicBarrier 使用了一種要麼全部要麼全不 (all-or-none) 的破壞模式：如果因為中斷、失敗或者超時等原因，導致執行緒過早地離開了屏障點，那麼在該屏障點等待的其他所有執行緒也將通過 BrokenBarrierException（如果它們幾乎同時被中斷，則用 InterruptedException）以反常的方式離開。
三、實現原理
基於ReentrantLock和Condition機制實現。除了getParties()方法，CyclicBarrier的其他方法都需要獲取鎖。
- 域
private final ReentrantLock lock = new ReentrantLock(); //可重入鎖
private final Condition trip = lock.newCondition();
private final int parties; //攔截的執行緒數量
private final Runnable barrierCommand; //當屏障撤銷時，需要執行的屏障操作
//當前的Generation。每當屏障失效或者開閘之後都會自動替換掉。從而實現重置的功能。
private Generation generation = new Generation();
//還能阻塞的執行緒數（即parties-當前阻塞的執行緒數），當新建generation或generation被破壞時，count會被重置。因為對Count的操作都是在獲取鎖之後，所以不需要其他同步措施。
private int count;
//靜態內聯類
private static class Generation {
boolean broken = false; //當前的屏障是否破壞
}
- await()
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 {
//儲存此時的generation
final Generation g = generation;
//判斷屏障是否被破壞
if (g.broken)
throw new BrokenBarrierException();
//判斷執行緒是否被中斷，如果被中斷，呼叫breakBarrier()進行屏障破壞處理，並丟擲InterruptedException
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
int index = --count; //剩餘count遞減，並賦值給執行緒索引，作為方法的返回值
//如果執行緒索引將為0，說明當前執行緒是最後一個到達的執行緒。執行可能存在的屏障操作 barrierCommand，設定下一個Generation。相當於每次開閘之後都進行了一次reset。
if (index == 0) { // tripped
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
if (command != null)
command.run(); //同步執行barrierCommand
ranAction = true;
nextGeneration(); //執行成功設定下一個nextGeneration
return 0;
} finally {
if (!ranAction) //如果barrierCommand執行失敗，進行屏障破壞處理
breakBarrier();
}
}
//如果當前執行緒不是最後一個到達的執行緒
for (;;) {
try {
if (!timed)
trip.await(); //呼叫Condition的await()方法阻塞
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos); //呼叫Condition的awaitNanos()方法阻塞
} catch (InterruptedException ie) {
//如果當前執行緒被中斷，則判斷是否有其他執行緒已經使屏障破壞。若沒有則進行屏障破壞處理，並丟擲異常；否則再次中斷當前執行緒
if (g == generation && ! g.broken) {
breakBarrier();
throw ie;
} else {
Thread.currentThread().interrupt();
//這種捕獲了InterruptException之後呼叫Thread.currentThread().interrupt()是一種通用的方式。其實就是為了儲存中斷狀態，從而讓其他更高層次的程式碼注意到這個中斷。
}
}
//如果屏障被破壞，當前執行緒拋BrokenBarrierException
if (g.broken)
throw new BrokenBarrierException();
//如果已經換代，直接返回index（last thread已經執行的nextGeneration，但當前執行緒還沒有執行到該語句）
if (g != generation)
return index;
//超時，進行屏障破壞處理，並拋TimeoutException
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
lock.unlock(); //釋放鎖
}
}
//將當前屏障置為破壞狀態、重置count、並喚醒所有被阻塞的執行緒。
//必須先獲取鎖，才能呼叫此方法
private void breakBarrier() {
generation.broken = true;
count = parties;
trip.signalAll();
}
//喚醒trip上等待的所有執行緒，設定下一個Generation
private void nextGeneration() {
trip.signalAll();
count = parties;
generation = new Generation();
}
- reset()
//重置屏障，先進行屏障破壞處理，再設定下一代generation
public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
breakBarrier(); // break the current generation
nextGeneration(); // start a new generation
} finally {
lock.unlock();
}
}
四、例子
public class Demo {
Random random = new Random();
public void meeting(CyclicBarrier barrier) {
try {
Thread.sleep(random.nextInt(40000));//讓執行緒隨機休息一段時間
} catch (InterruptedException e1) {
e1.printStackTrace();
}
System.out.println(Thread.currentThread().getName()+"到達了會議室，等待開會");
// if(Thread.currentThread().getName().equals("Thread-1")) {
// throw new RuntimeException();
// }
try {
barrier.await();//等待
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName()+"發言");
}
public static void main(String[] args) {
Demo d = new Demo();
CyclicBarrier barrier = new CyclicBarrier(10,new Runnable() {
@Override
public void run() {
System.out.println("好！，我們開始開會。。。");
}
});
for(int i=0;i<10;i++) {
new Thread(new Runnable(){
@Override
public void run() {
d.meeting(barrier);
}
}).start();
}
}
}
五、CyclicBarrier與CountDownLatch比較
- CountDownLatch是執行緒組之間的等待，即一個(或多個)執行緒等待N個執行緒完成某件事情之後再執行；而CyclicBarrier則是執行緒組內的等待，即每個執行緒相互等待，即N個執行緒都被攔截之後，然後依次執行。
- CountDownLatch是減計數方式，而CyclicBarrier是加計數方式。
- CountDownLatch計數為0無法重置，而CyclicBarrier計數達到初始值，則可以重置。
- CountDownLatch不可以複用，而CyclicBarrier可以複用。
- CountDownLatch基於AQS；CyclicBarrier基於鎖和Condition。本質上都是依賴於volatile和CAS實現的。