CyclicBarrier:人齊了,老司機就可以發車了!

上一篇咱講了 CountDownLatch 可以解決多個執行緒同步的問題,相比於 join 來說它的應用範圍更廣,不僅可以應用在執行緒上,還可以應用在執行緒池上。然而 CountDownLatch 卻是一次性的計數器,以王者農藥來說,咱們不可能一場團戰就決定比賽的輸贏,所以在某些場景下,咱們是需要重複使用某個等待功能的,這就是我們今天要介紹的另一個主角——CyclicBarrier。

CyclicBarrier

CyclicBarrier 翻譯為中文是循環(Cyclic)柵欄(Barrier)的意思,它的大概含義是實現一個可循環利用的屏障。
image.png
CyclicBarrier 作用是讓一組執行緒相互等待,當達到一個共同點時,所有之前等待的執行緒再繼續執行,且 CyclicBarrier 功能可重複使用。

20181218144511688.gif

舉個栗子

比如磊哥要坐班車回老家,因為中途不允許上、下乘客,所以營運的公司為了收益最大化,就會等人滿之後再發車。像這種等人坐滿就發一班車的場景,就是 CyclicBarrier 所擅長的,因為它可以重複使用(不像 CountDownLatch 那樣只能用一次)。
image.png

CyclicBarrier VS CountDownLatch

CountDownLatch:一個或者多個執行緒,等待另外 N 個執行緒完成某個事情之後才能執行。

CountDownLatch 就像玩王者農藥開局的載入一樣,所有人要等待其他人都載入 100% 之後才能開始遊戲。

image.png

CyclicBrrier:N 個執行緒相互等待,直到有足夠數量的執行緒都到達屏障點之後,之前等待的執行緒就可以繼續執行了。

CyclicBrrier 就像老司機開車一樣,如果車上還有空餘的座位,那麼所有人都得等著,直到座位被坐滿之後,老司機才會發車。
老司机发车.gif

CyclicBarrier使用

import java.util.Date;
import java.util.Random;
import java.util.concurrent.*;

public class CyclicBarrierExample {
    public static void main(String[] args) {
        // 創建 CyclicBarrier
        final CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new Runnable() {
            @Override
            public void run() {
                System.out.println("人滿了,準備發車:" + new Date());
            }
        });
        
        // 執行緒調用的任務
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                // 生成隨機數 1-3
                int randomNumber = new Random().nextInt(3) + 1;
                // 進入任務
                System.out.println(String.format("我是:%s 再走:%d 秒就到車站了,現在時間:%s",
                        Thread.currentThread().getName(), randomNumber, new Date()));
                try {
                    // 模擬執行
                    TimeUnit.SECONDS.sleep(randomNumber);
                    // 調用 CyclicBarrier
                    cyclicBarrier.await();
                    // 任務執行
                    System.out.println(String.format("執行緒:%s 上車,時間:%s",
                            Thread.currentThread().getName(), new Date()));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        };

        // 創建執行緒池
        ExecutorService threadPool = Executors.newFixedThreadPool(10);
        // 執行任務 1
        threadPool.submit(runnable);
        // 執行任務 2
        threadPool.submit(runnable);
        // 執行任務 3
        threadPool.submit(runnable);
        // 執行任務 4
        threadPool.submit(runnable);

        // 等待所有任務執行完終止執行緒池
        threadPool.shutdown();
    }
}

以上程式碼執行結果如下:
image.png
從上述結果可以看出:當 CyclicBarrier 的計數器設置為 2 時,執行緒 2 和 執行緒 3 都到屏障點之後,老司機才會發第一波車,再 2s 之後,執行緒 1 和執行緒 4 也同時進入了屏障點,這時候老司機又可以再發一波車了。

實現原理

我們先來看下 CyclicBarrier 的類圖:
image.png
由上圖可知 CyclicBarrier 是基於獨佔鎖 ReentrantLock 實現的,其底層也是基於 AQS 的。

在 CyclicBarrier 類的內部有一個計數器 count,當 count 不為 0 時,每個執行緒在到達屏障點會先調用 await 方法將自己阻塞,此時計數器會減 1,直到計數器減為 0 的時候,所有因調用 await 方法而被阻塞的執行緒就會被喚醒繼續執行。當 count 計數器變成 0 之後,就會進入下一輪阻塞,此時 parties(parties 是在 new CyclicBarrier(parties) 時設置的值)會將它的值賦值給 count 從而實現復用。

常用方法

CyclicBarrier(parties):初始化相互等待的執行緒數量的構造方法。

CyclicBarrier(parties,Runnable barrierAction):初始化相互等待的執行緒數量以及屏障執行緒的構造方法,當 CyclicBarrier 的計數器變為 0 時,會執行 barrierAction 構造方法。

getParties():獲取 CyclicBarrier 打開屏障的執行緒數量,也稱為方數。

getNumberWaiting():獲取正在CyclicBarrier上等待的執行緒數量。

await():在 CyclicBarrier 上進行阻塞等待,直到發生以下情形之一:
在 CyclicBarrier 上等待的執行緒數量達到 parties,則所有執行緒被釋放,繼續執行;

  • 當前執行緒被中斷,則拋出 InterruptedException 異常,並停止等待,繼續執行;
  • 其他等待的執行緒被中斷,則當前執行緒拋出 BrokenBarrierException 異常,並停止等待,繼續執行;
  • 其他等待的執行緒超時,則當前執行緒拋出 BrokenBarrierException 異常,並停止等待,繼續執行;
  • 其他執行緒調用 CyclicBarrier.reset() 方法,則當前執行緒拋出 BrokenBarrierException 異常,並停止等待,繼續執行。

await(timeout,TimeUnit):在CyclicBarrier上進行限時的阻塞等待,直到發生以下情形之一:

  • 在 CyclicBarrier 上等待的執行緒數量達到 parties,則所有執行緒被釋放,繼續執行;
  • 當前執行緒被中斷,則拋出 InterruptedException 異常,並停止等待,繼續執行;
  • 當前執行緒等待超時,則拋出 TimeoutException 異常,並停止等待,繼續執行;
  • 其他等待的執行緒被中斷,則當前執行緒拋出 BrokenBarrierException 異常,並停止等待,繼續執行;
  • 其他等待的執行緒超時,則當前執行緒拋出 BrokenBarrierException 異常,並停止等待,繼續執行;
  • 其他執行緒調用 CyclicBarrier.reset() 方法,則當前執行緒拋出 BrokenBarrierException 異常,並停止等待,繼續執行。

isBroken():獲取是否破損標誌位 broken 的值,此值有以下幾種情況:

  • CyclicBarrier 初始化時,broken=false,表示屏障未破損;
  • 如果正在等待的執行緒被中斷,則 broken=true,表示屏障破損;
  • 如果正在等待的執行緒超時,則 broken=true,表示屏障破損;
  • 如果有執行緒調用 CyclicBarrier.reset() 方法,則 broken=false,表示屏障回到未破損狀態。

reset():使得CyclicBarrier回歸初始狀態,直觀來看它做了兩件事:

  • 如果有正在等待的執行緒,則會拋出 BrokenBarrierException 異常,且這些執行緒停止等待,繼續執行。
  • 將是否破損標誌位 broken 置為 false。

總結

CyclicBrrier 是通過獨佔鎖 ReentrantLock 實現計數器的原子性更新的,CyclicBrrier 最常用的是 await() 方法,使用此方法會將計數器 -1,並判斷當前的計數器是否為 0,如果不為 0 就會阻塞等待,並計時器為 0 之後,才能繼續執行剩餘任務。CyclicBrrier 相比於 CountDownLatch 來說,它的優勢在於可以重複使用。

參考 & 鳴謝

blog.csdn.net/qq_39241239/article/details/87030142
blog.csdn.net/zzg1229059735/article/details/61191679
www.cnblogs.com/yaochunhui/p/13494689.html

關注公眾號:「Java中文社群」查看更多精彩內容。

Tags: