一、概念理解

CyclicBarrier允許一組線程在到達某個柵欄點(common barrier point)互相等待，直到最後一個線程到達柵欄點，柵欄才會打開，處於阻塞狀態的線程恢復繼續執行。

就比如說我們在打王者的時候，十個人必須全部加載到100%，才可以開局。否則只要有一個人沒有加載到100%,那這個遊戲就不能開始。先加載完成的玩家必須等待最後一個玩家加載成功才可以。如果你實在記不住，你可以想象成人滿發車的長途，就算你是第一個上車的人，也要等待車滿才可以發車。否則車上所有人都要等待。

與CountDownLatch的區別就是是否相互等待。舉一個例子，CountDownLatch就好比是馬拉松比賽，跑完的人不用等待其他選手是否結束，而CyclicBarrier需要等最後一個玩家加載結束。這就是區別。

我們直接代碼演示一下這個例子。

二、代碼演示

在這裡我們同樣使用的是打王者的例子。

首先我們定義main線程：

public class CyclicBarrierTest {
    public static void main(String[] args) {
        //第一步：定義玩家，這裡寫了5個
        String[] heros  = {"孫悟空","豬八戒","狄仁傑","魯班","甄姬"};
        //第二步：使用線程池來運行，也是5個。
        ExecutorService service = Executors.newFixedThreadPool(5);
        //第三步：常見圍欄，也是5個
        final CyclicBarrier barrier = new CyclicBarrier(10);
        //第四步：通過for循環傳遞給每一個玩家和圍欄
        for (int i = 0; i < 10; i++) {
            service.execute(new Player(heros[i], barrier));
        }
        service.shutdown();
    }
}

上面的代碼我們已經解釋清楚了，主要是通過線程池運行玩家，並傳遞給每一個玩家名字和圍欄。下面我們就看看這個Player玩家線程如何實現的。

public class Player implements Runnable {
    private final String hero;
    private final CyclicBarrier barrier;
    public Player(String hero, CyclicBarrier barrier) {
        this.hero = hero;
        this.barrier = barrier;
    }
    @Override
    public void run() {
        try {
            //每一個英雄加載成功的時間不一樣，所以這裡用了隨機數
            TimeUnit.SECONDS.sleep(1 + (new Random().nextInt(5)));
            System.out.println(hero + "開始加載==========等待其他玩家加載成功");
            barrier.await();
            System.out.println(hero + "：看到所有玩家加載成功，比賽開始");
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    }
}

在這個player線程中，我們使用隨機數來表示每個玩家不同的加載時間，在休眠時間結束之前，player一直處於等待的狀態，也就是調用了await方法。現在測試一遍。

現在相信通過這個案例，大家都能掌握其用法，很簡單。下面我們從源碼的角度來分析一下其實現原理。

三、源碼分析

為了分析的徹底，我們先從構造方法開始：

//構造方法1： parties主要是需要攔截的線程數
public CyclicBarrier(int parties) {
    this(parties, null);
}
//構造方法2：不僅有parties還有barrierAction
//主要是為了處理更加複雜的場景，當線程到達圍欄時候
//優先執行barrierAction
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

以上就是兩個構造方法，下面我們主要分析await方法我們進入源碼看看：

//不帶超時時間
    public int await() throws InterruptedException, 
                            BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }
    //帶有超時時間
    public int await(long timeout, TimeUnit unit)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        return dowait(true, unit.toNanos(timeout));
    }

返回的值是加載成功的玩家數量，既然內部是通過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()) {
                breakBarrier();
                throw new InterruptedException();
            }
            int index = --count;
            if (index == 0) {  // tripped
                boolean ranAction = false;
                try {
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();
                    ranAction = true;
                    nextGeneration();
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }
            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();
        }
    }

上面的代碼有點長，不過也是CyclicBarrier的核心，我在這裡說一下上面代碼的主要功能：

（1）通過ReentrantLock獲取獨佔鎖。

（2）通過try裡面的Generation判斷當前代是否損壞，通過Thread的interrupted方法判斷是否線程中斷，如果中斷通過breakBarrier方法告訴其他線程。

（3）if(index==0)判斷當前是否是最後一個線程調用了await方法，如果是則把之前等待的線程全部喚醒。就好比是最後一個運動員到達了終點，告訴其他選手比賽結束了。

（4）for(;;)循環執行等待，如果沒有超時時間，那就一直等待直到被喚醒，有超時時間，那就等時間過後自動被喚醒。就好比是在運動員在路上跑步，沒有時間限制的時候那就一直跑，一直到達終點。如果有時間限制，不管是否跑到終點，比賽都結束。

（5）通過ReentrantLock釋放獨佔鎖。

這段代碼讀起來比較麻煩，因為裡面涉及到了兩個其他的類ReentrantLock和Generation，我們只需要知道其作用即可。還有一個點，那就是線程被中斷了，如何告訴其他線程的breakBarrier方法。

/**
     * Sets current barrier generation as broken and wakes up everyone.
     * Called only while holding lock.
     */
    private void breakBarrier() {
        generation.broken = true;
        count = parties;
        trip.signalAll();
    }

我們可以看到，在breakBarrier()中除了將broken設置為true，還會調用signalAll將在CyclicBarrier處於等待狀態的線程全部喚醒。

OK，今天的文章就先到這，關於CyclicBarrier原理我們已經解釋了，其用途可以根據自己的業務場景來決定了。

詳解java中的同步工具類CyclicBarrier

一、概念理解

二、代碼演示

三、源碼分析

Leave a Reply Cancel reply