并發編程之CyclicBarrier原理與使用
前言
控制并發流程的工具類，作用就是幫助我們程序員更容易的讓線程之間合作，讓線程之間相互配合來滿足業務邏輯。比如讓線程A等待線程B執行完畢后再執行等合作策略。
控制并發流程的工具類主要有：
簡介
從字面意思看，這個類的中文意思是“循環柵欄”。大概的意思就是一個可循環利用的屏障。它的作用就是會讓所有線程都等待完成后才會繼續下一步行動。
舉個例子，就像生活中我們會約朋友到某個餐廳一起吃飯，有些朋友可能會早到，有些朋友可能會晚到，但這個餐廳規定必須等到所有人到期之后才會讓我們進去。這里的朋友們就各個線程，餐廳就是CyclicBarrier。
在JUC包中為我們提供了一個同步工具類能夠很好的模擬這類場景，它就是CyclicBarrier類。利用CyclicBarrier類可以實現一組線程相互等待，當所有線程都到達某個屏障點后再進行后續的操作。下圖演示了這一過程。
應用場景
可用于多線程計數數據，最后合并計數結果的場景。
使用CyclicBarrier實現等待的線程都被稱為參與方。參與方只需要執行cyclicBarrier.await() 就可以實現等待。由于CyclicBarrier內部維護了一個顯示鎖，這可以知道參與方中誰最后一個執行cyclicBarrier.await() 。當最后一個線程執行完，會使得使用相應CyclicBarrier實例的其他參與方被喚醒，而最后一個線程自身不會被暫停。其流程圖如下：
- public static void main(String[] args) {
- CyclicBarrier cyclicBarrier = new CyclicBarrier(7,() ->{
- System.out.println("****召喚神龍");
- });
- for(int i = 1;i <= 7; i++){
- int finalI = i;
- new Thread(() -> {
- System.out.println(Thread.currentThread().getName() + "\t 收集到第"+ finalI +"顆龍珠");
- try {
- cyclicBarrier.await();
- } catch (InterruptedException e) {
- e.printStackTrace();
- } catch (BrokenBarrierException e) {
- e.printStackTrace();
- }
- },String.valueOf(i)).start();
- }
- }
源碼分析
CyclicBarrier 類圖
CyclicBarrier是包含了 “ReentrantLock對象lock” 和 “Condition對象trip”，它是通過獨占鎖實現的。
其內部主要變量和方法如下：
成員變量
//同步操作鎖
private final ReentrantLock lock = new ReentrantLock();
- //同步操作鎖
- private final ReentrantLock lock = new ReentrantLock();
- //線程攔截器
- private final Condition trip = lock.newCondition();
- //每次攔截的線程數
- private final int parties;
- //換代前執行的任務
- private final Runnable barrierCommand;
- //表示柵欄的當前代
- private Generation generation = new Generation();
- //計數器
- private int count;
- //靜態內部類Generation
- private static class Generation {
- boolean broken = false;
- }
可以看到 CyclicBarrier 內部是通過條件隊列 trip 來對線程進行阻塞的，并且其內部維護了兩個 int 型的變量 parties 和 count：
- parties 表示每次攔截的線程數，該值在構造時進行賦值;
- count 是內部計數器，它的初始值和 parties 相同，以后隨著每次 await 方法的調用而減 1，直到減為 0 就將所有線程喚醒。
CycliBarrier 有一個靜態內部類 Generation，該類的對象代表柵欄的當前代，就像玩游戲時代表的本局有些，利用它可以實現循環等待。barrierCommand 表示換代前執行的任務，當 count 減為 0 時表示本局游戲結束，需要轉到下一句。在轉到下一句游戲之前會將所有阻塞的線程喚醒，在喚醒所有線程之前你可以通過指定 barrierCommand 來執行自己的任務。
構造函數
主要提供了兩個構造方法
- public CyclicBarrier(int parties) {
- this(parties, null);
- }
- public CyclicBarrier(int parties, Runnable barrierAction) {
- if (parties <= 0) throw new IllegalArgumentException();
- // parties表示“必須同時到達barrier的線程個數”。
- this.parties = parties;
- // count表示“處在等待狀態的線程個數”。
- this.count = parties;
- // barrierCommand表示“parties個線程到達barrier時，會執行的動作”。
- this.barrierCommand = barrierAction;
- }
解析：
- parties 是參與線程的個數
- 第二個構造方法有一個Runnable參數，這個參數的意思是最后一個到達線程要執行的動作。
重要方法
CyclicBarrier類最主要的功能就是使先到達屏障點的線程阻塞并等待后面的線程，其中它提供了兩種等待的方法，分別是定時等待和非定時等待。
await()方法
- //非定時等待
- public int await() throws InterruptedException, BrokenBarrierException {
- try {
- return dowait(false, 0L);
- } catch (TimeoutException toe) {
- throw new Error(toe);
- }
- }
- //定時等待
- public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException {
- return dowait(true, unit.toNanos(timeout));
- }
解析：
- 線程調用await()表示總結已經到達柵欄
- BrokenBarrierException表示柵欄已經被破壞，破壞的原因可能是其中一個線程await()時被中斷或者超時。
dowait()方法
可以看到不管是定時等待還是非定時等待，它們都調用了dowait方法，只不過是傳入的參數不同而已。下面我們就來看看dowait方法都做了些什么。
- //核心等待方法
- 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()) {
- //如果當前線程被中斷會做以下三件事
- //1.打翻當前柵欄
- //2.喚醒攔截的所有線程
- //3.拋出中斷異常
- breakBarrier();
- throw new InterruptedException();
- }
- //每次都將計數器的值減1
- int index = --count;
- //計數器的值減為0則需喚醒所有線程并轉換到下一代
- if (index == 0) {
- boolean ranAction = false;
- try {
- //喚醒所有線程前先執行指定的任務
- final Runnable command = barrierCommand;
- if (command != null) {
- command.run();
- }
- ranAction = true;
- //喚醒所有線程并轉到下一代
- nextGeneration();
- return 0;
- } finally {
- //確保在任務未成功執行時能將所有線程喚醒
- if (!ranAction) {
- breakBarrier();
- }
- }
- }
- //如果計數器不為0則執行此循環
- 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();
- }
- }
上面執行的代碼相對比較容易看懂，我們再來看一下執行流程：
- 獲得顯示鎖，判斷當前線程狀態是否被中斷，如果是，則執行 breakBarrier 方法，喚醒之前阻塞的所有線程，并將計數器重置;
- 計數器 count 減 1，如果 count == 0，表示最后一個線程達到柵欄，接著執行之前指定的 Runnable 接口，同時執行 nextGeneration 方法進入下一代;
- 否則，進入自旋，判斷當前線程是進入定時等待還是非定時等待，如果在等待過程中被中斷，執行 breakBarrier 方法，喚醒之前阻塞的所有線程;
- 判斷是否是因為執行 breakBarrier 方法而被喚醒，如果是，則拋出異常;
- 判斷是否是正常的換代操作而被喚醒，如果是，則返回計數器的值;
- 判斷是否是超時而被喚醒，如果是，則喚醒之前阻塞的所有線程，并拋出異常;
- 釋放鎖。
breakBarrier()方法
- private void breakBarrier() {
- generation.broken = true;//柵欄被打破
- count = parties;//重置count
- trip.signalAll();//喚醒之前阻塞的線程
- }
nextGeneration()方法
- private void nextGeneration() {
- //喚醒所以的線程
- trip.signalAll();
- //重置計數器
- count = parties;
- //重新開始
- generation = new Generation();
- }
reset()方法
接下來看看柵欄重置的方法
- // 重置barrier到初始狀態，所有還在等待中的線程最終會拋出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();
- }
- }
其它方法
CyclicBarrier 其它還提供了例如getParties，isBroken，getNumberWaiting等方法，都比較簡單，其中除了getParties由于parties被final修飾不可變，其余方法都會先去獲得互斥鎖。
- /**
- * 獲取當前這一輪是否已經broken。
- */
- public boolean isBroken() {
- final ReentrantLock lock = this.lock;
- lock.lock();
- try {
- return generation.broken;
- } finally {
- lock.unlock();
- }
- }
- /**
- * 獲得當前在barrier中等待的線程數。
- */
- public int getNumberWaiting() {
- final ReentrantLock lock = this.lock;
- lock.lock();
- try {
- return parties - count;
- } finally {
- lock.unlock();
- }
- }
總結
CountDownLatch和CyclicBarrier區別
- CountDownLatch和CyclicBarrier都能夠實現線程之間的等待，只不過它們側重點不同：
- CountDownLatch一般用于一個或多個線程，等待其他線程執行完任務后，再才執行;
- CyclicBarrier一般用于一組線程互相等待至某個狀態，然后這一組線程再同時執行;
- CountDownLatch 是一次性的，CyclicBarrier 是可循環利用的;
- CountDownLathch是一個計數器，線程完成一個記錄一個，計數器遞減，只能用一次。如下圖：
CyclicBarrier的計數器更像一個閥門，需要所有線程都到達，然后繼續執行，計數器遞減，提供reset功能，可以多次使用。如下圖：
PS：以上代碼提交在 Github ：
https://github.com/Niuh-Study/niuh-juc-final.git
PS：這里有一個技術交流群(QQ群:1158819530)，方便大家一起交流，持續學習，共同進步，有需要的可以加一下。
