[[388240]]

引言

前面一篇文章我們《Java高并發編程基礎三大利器之CountDownLatch》它有一個缺點，就是它的計數器只能夠使用一次，也就是說當計數器(state)減到為 0的時候，如果 再有線程調用去 await() 方法，該線程會直接通過，不會再起到等待其他線程執行結果起到同步的作用。為了解決這個問題CyclicBarrier就應運而生了。

什么是CyclicBarrier

CyclicBarrier是什么?把它拆開來翻譯就是循環(Cycle)和屏障(Barrier)

它的主要作用其實和CountDownLanch差不多，都是讓一組線程到達一個屏障時被阻塞，直到最后一個線程到達屏障時，屏障會被打開，所有被屏障阻塞的線程才會繼續執行，不過它是可以循環執行的，這是它與CountDownLanch最大的不同。CountDownLanch是只有當最后一個線程把計數器置為0的時候，其他阻塞的線程才會繼續執行。學習CyclicBarrier之前建議先去看看這幾篇文章：

《Java高并發編程基礎之AQS》

《Java高并發編程基礎三大利器之Semaphore》

《Java高并發編程基礎三大利器之CountDownLatch》

如何使用

我們首先先來看下關于使用CyclicBarrier的一個demo：比如游戲中有個關卡的時候，每次進入下一關的時候都需要進行加載一些地圖、特效背景音樂什么的只有全部加載完了才能夠進行游戲:

/**demo 來源https://blog.csdn.net/lstcui/article/details/107389371 
 * 公眾號【java金融】 
 */ 
public class CyclicBarrierExample { 
    static class PreTaskThread implements Runnable { 
        private String task; 
        private CyclicBarrier cyclicBarrier; 
 
        public PreTaskThread(String task, CyclicBarrier cyclicBarrier) { 
            this.task = task; 
            this.cyclicBarrier = cyclicBarrier; 
        } 
 
        @Override 
        public void run() { 
            for (int i = 0; i < 4; i++) { 
                Random random = new Random(); 
                try { 
                    Thread.sleep(random.nextInt(1000)); 
                    System.out.println(String.format("關卡 %d 的任務 %s 完成", i, task)); 
                    cyclicBarrier.await(); 
                } catch (InterruptedException | BrokenBarrierException e) { 
                    e.printStackTrace(); 
                } 
            } 
        } 
 
        public static void main(String[] args) { 
            CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> { 
                System.out.println("本關卡所有的前置任務完成，開始游戲... ..."); 
            }); 
            new Thread(new PreTaskThread("加載地圖數據", cyclicBarrier)).start(); 
            new Thread(new PreTaskThread("加載人物模型", cyclicBarrier)).start(); 
            new Thread(new PreTaskThread("加載背景音樂", cyclicBarrier)).start(); 
        } 
    } 
} 

輸出結果如下：

我們可以看到每次游戲開始都會等當前關卡把游戲的人物模型，地圖數據、背景音樂加載完成后才會開始進行游戲。并且還是可以循環控制的。

源碼分析

結構組成

/** 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(); 

• lock:用于保護屏障入口的鎖
• trip :達到屏障并且不能放行的線程在trip條件變量上等待
• parties :柵欄開啟需要的到達線程總數
• barrierCommand：最后一個線程到達屏障后執行的回調任務
• generation：這是一個內部類，通過它實現CyclicBarrier重復利用，每當await達到最大次數的時候，就會重新new 一個，表示進入了下一個輪回。里面只有一個boolean型屬性，用來表示當前輪回是否有線程中斷。

主要方法

public int await() throws InterruptedException, BrokenBarrierException { 
    try { 
        return dowait(false, 0L); 
    } catch (TimeoutException toe) { 
        throw new Error(toe); // cannot happen 
    } 
} 
 
 * Main barrier code, covering the various policies. 
 */ 
private int dowait(boolean timed, long nanos) 
    throws InterruptedException, BrokenBarrierException, 
           TimeoutException { 
    final ReentrantLock lock = this.lock; 
    lock.lock(); 
     try { 
           //獲取barrier當前的 “代”也就是當前循環 
         final Generation g = generation; 
        if (g.broken) 
            throw new BrokenBarrierException(); 
 
        if (Thread.interrupted()) { 
            breakBarrier(); 
            throw new InterruptedException(); 
        } 
        // 每來一個線程調用await方法都會進行減1 
        int index = --count; 
        if (index == 0) {  // tripped 
            boolean ranAction = false; 
            try { 
                final Runnable command = barrierCommand; 
                // new CyclicBarrier 傳入 的barrierCommand, command.run()這個方法是同步的，如果耗時比較多的話,是否執行的時候需要考慮下是否異步來執行。 
                if (command != null) 
                    command.run(); 
                ranAction = true; 
                // 這個方法1. 喚醒所有阻塞的線程，2. 重置下count（count 每來一個線程都會進行減1）和generation，以便于下次循環。 
                nextGeneration(); 
                return 0; 
            } finally { 
                if (!ranAction) 
                    breakBarrier(); 
            } 
        } 
 
        // loop until tripped, broken, interrupted, or timed out 
        for (;;) { 
            try { 
                 // 進入if條件，說明是不帶超時的await 
                if (!timed) 
                     // 當前線程會釋放掉lock，然后進入到trip條件隊列的尾部，然后掛起自己，等待被喚醒。 
                    trip.await(); 
                else if (nanos > 0L) 
                     //說明當前線程調用await方法時 是指定了 超時時間的！ 
                    nanos = trip.awaitNanos(nanos); 
            } catch (InterruptedException ie) { 
                 //Node節點在 條件隊列內 時 收到中斷信號時 會拋出中斷異常！ 
                //g == generation 成立，說明當前代并沒有變化。 
                //! g.broken 當前代如果沒有被打破，那么當前線程就去打破，并且拋出異常.. 
                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. 
                //執行到else有幾種情況？ 
                //1.代發生了變化，這個時候就不需要拋出中斷異常了，因為 代已經更新了，這里喚醒后就走正常邏輯了..只不過設置下 中斷標記。 
                //2.代沒有發生變化，但是代被打破了，此時也不用返回中斷異常，執行到下面的時候會拋出  brokenBarrier異常。也記錄下中斷標記位。 
                    Thread.currentThread().interrupt(); 
                } 
            } 
           //喚醒后，執行到這里，有幾種情況？ 
          //1.正常情況，當前barrier開啟了新的一代（trip.signalAll()） 
          //2.當前Generation被打破，此時也會喚醒所有在trip上掛起的線程 
          //3.當前線程trip中等待超時，然后主動轉移到 阻塞隊列 然后獲取到鎖 喚醒。 
            if (g.broken) 
                throw new BrokenBarrierException(); 
           //喚醒后，執行到這里，有幾種情況？ 
        //1.正常情況，當前barrier開啟了新的一代（trip.signalAll()） 
        //2.當前線程trip中等待超時，然后主動轉移到 阻塞隊列 然后獲取到鎖 喚醒。 
            if (g != generation) 
                return index; 
           //喚醒后，執行到這里，有幾種情況？ 
        //.當前線程trip中等待超時，然后主動轉移到 阻塞隊列 然后獲取到鎖 喚醒。 
            if (timed && nanos <= 0L) { 
                breakBarrier(); 
                throw new TimeoutException(); 
            } 
        } 
    } finally { 
         lock.unlock(); 
    } 
} 

小結

到了這里我們是不是可以知道為啥CyclicBarrier可以進行循環計數?

• CyclicBarrier采用一個內部類Generation來維護當前循環，每一個await方法都會存儲當前的generation，獲取到相同generation對象的屬于同一組，每當count的次數耗盡就會重新new一個Generation并且重新設置count的值為parties，表示進入下一次新的循環。

從這個await方法我們是不是可以知道只要有一個線程被中斷了，當代的 generation的broken 就會被設置為true，所以會導致其他的線程也會被拋出BrokenBarrierException。相當于一個失敗其他也必須失敗，感覺有“強一致性“的味道。

總結

• CountDownLanch是為計數器是設置一個值，當多次執行countdown后，計數器減為0的時候所有線程被喚醒，然后CountDownLanch失效，只能夠使用一次。
• CyclicBarrier是當count為0時同樣喚醒全部線程，同時會重新設置count為parties，重新new一個generation來實現重復利用。

本文轉載自微信公眾號「java金融」，可以通過以下二維碼關注。轉載本文請聯系java金融公眾號。