CountdownLatch，CyclicBarrier是非常常用併發工具類，可以說是Java工程師必會技能了。不但在專案實戰中經常涉及，而且在編寫壓測程式，多執行緒demo也是必不可少，所以掌握它們的用法和實現原理非常有必要。

念念不忘，必有迴響！
點贊走一走，找到女朋友~

等待多執行緒完成的CountDownLatch

CountDownLatch允許一個或多個執行緒等待其他執行緒完成操作。也就是說通過使用CountDownLatch工具類，可以讓一組執行緒等待彼此執行完畢後在共同執行下一個操作。具體流程如下圖所示，箭頭表示任務，矩形表示柵欄，當三個任務都到達柵欄時，柵欄後wait的任務才開始執行。

CountDownLatch維護有個int型的狀態碼，每次呼叫countDown時狀態值就會減1；呼叫wait方法的執行緒會阻塞，直到狀態碼為0時才會繼續執行。

在多執行緒協同工作時，可能需要等待其他執行緒執行完畢之後，主執行緒才接著往下執行。首先我們可能會想到使用執行緒的join方法（呼叫join方法的執行緒優先執行，該執行緒執行完畢後才會執行其他執行緒），顯然這是可以完成的。

使用Thread.join()方法實現

public class RunningRaceTest {
    public static void main(String[] args) throws InterruptedException {
        Thread runner1 = new Thread(new Runner(), "1號");
        Thread runner2 = new Thread(new Runner(), "2號");
        Thread runner3 = new Thread(new Runner(), "3號");
        Thread runner4 = new Thread(new Runner(), "4號");
        Thread runner5 = new Thread(new Runner(), "5號");
        runner1.start();
        runner2.start();
        runner3.start();
        runner4.start();
        runner5.start();

        runner1.join();
        runner2.join();
        runner3.join();
        runner4.join();
        runner5.join();

        // 裁判等待5名選手準備完畢
        System.out.println("裁判：比賽開始~~");
    }
}

class Runner implements Runnable {
    @Override
    public void run() {
        try {
            int sleepMills = ThreadLocalRandom.current().nextInt(1000);
            Thread.sleep(sleepMills);
            System.out.println(Thread.currentThread().getName() + " 選手已就位, 準備共用時： " + sleepMills + "ms");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
 

Thread.join()完全可以實現這個需求，不過存在一個問題，如果呼叫join的執行緒一直存活，則當前執行緒則需要一直等待。這顯然不夠靈活，並且當前執行緒可能會出現死等的情況。

更加靈活的CountDownLatch

jdk1.5之後的併發包中提供了CountDownLatch併發工具了，也可以實現join的功能，並且功能更加強大。

// 參賽選手執行緒
class Runner implements Runnable {
    private CountDownLatch countdownLatch;
    
    public Runner(CountDownLatch countdownLatch) {
        this.countdownLatch = countdownLatch;
    }

    @Override
    public void run() {
        try {
            int sleepMills = ThreadLocalRandom.current().nextInt(1000);
            Thread.sleep(sleepMills);
            System.out.println(Thread.currentThread().getName() + " 選手已就位, 準備共用時： " + sleepMills + "ms");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 準備完畢，舉手示意
            countdownLatch.countDown();
        }
    }
}

public class RunningRaceTest {
    public static void main(String[] args) throws InterruptedException {
        // 使用執行緒池的正確姿勢
        int size = 5;
        AtomicInteger counter = new AtomicInteger();
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(size, size, 1000, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100), (r) -> new Thread(r, counter.addAndGet(1) + " 號 "), new ThreadPoolExecutor.AbortPolicy());
        
        CountDownLatch countDownLatch = new CountDownLatch(5);
        for (int i = 0; i < size; i++) {
            threadPoolExecutor.submit(new Runner(countDownLatch));
        }

        // 裁判等待5名選手準備完畢
        countDownLatch.await(); // 為了避免死等，也可以新增超時時間
        System.out.println("裁判：比賽開始~~");

        threadPoolExecutor.shutdownNow();
    }
}
 

輸出結果：

5 號  選手已就位, 準備共用時： 20ms
4 號  選手已就位, 準備共用時： 156ms
1 號  選手已就位, 準備共用時： 288ms
2 號  選手已就位, 準備共用時： 519ms
3 號  選手已就位, 準備共用時： 945ms
比賽開始~~

同步屏障CyclicBarrier

CyclicBarrier可以實現CountDownLatch一樣的功能，不同的是CountDownLatch屬於一次性物件，聲明後只能使用一次，而CyclicBarrier可以迴圈使用。

從字面意義上來看，CyclicBarrier表示迴圈的屏障，當一組執行緒全部都到達屏障時，屏障才會被移除，否則只能阻塞在屏障處。

public class RunningRace {
    public static void main(String[] args) {
        // 使用執行緒池的正確姿勢
        int size = 5;
        AtomicInteger counter = new AtomicInteger();
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(size, size, 1000, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100), (r) -> new Thread(r, counter.addAndGet(1) + " 號 "), new ThreadPoolExecutor.AbortPolicy());

        CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> System.out.println("裁判：比賽開始~~"));
        for (int i = 0; i < 10; i++) {
            threadPoolExecutor.submit(new Runner(cyclicBarrier));
        }
    }
}

class Runner implements Runnable {
    private CyclicBarrier cyclicBarrier;

    public Runner(CyclicBarrier countdownLatch) {
        this.cyclicBarrier = countdownLatch;
    }

    @Override
    public void run() {
        try {
            int sleepMills = ThreadLocalRandom.current().nextInt(1000);
            Thread.sleep(sleepMills);
            System.out.println(Thread.currentThread().getName() + " 選手已就位, 準備共用時： " + sleepMills + "ms" + cyclicBarrier.getNumberWaiting());
            cyclicBarrier.await();
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    }
}

由於CyclicBarrier可以迴圈使用，所以CyclicBarrier的構造方法中可以傳入一個Runnable引數，在每一輪執行完畢之後就會立刻執行這個Runnable任務。

CountDownLatch設計與實現

CountDownLath是基於AQS框架的一種簡單實現，有兩個核心的方法，即await()和countDown()，通過構造方法傳入一個狀態值，呼叫await()方法時執行緒會阻塞，直到狀態碼被修改成0時才會返回，每次呼叫countDown()時會將狀態值減1。

wait方法：執行wait方法後，會嘗試獲取同步狀態，如果為狀態為0則方法繼續執行，否擇當前執行緒會被加入到同步佇列中，詳情可見筆者關於AQS的兩篇文章。

public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
    throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // 如果狀態碼不為0，嘗試獲取同步狀態，如果失敗則被加入到同步佇列中
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}
// 當狀態碼為0時返回1，否擇返回-1，這個方法中引數沒有任何用處
protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}

countDown方法：每次執行countDown方法時，會將狀態碼的值減1.

public void countDown() {
    sync.releaseShared(1);
}

CyclicBarrier的設計與實現

CyclicBarrier與CountDownLatch實現思想相同，也是基於AQS框架實現。不同的是CyclicBarrier內部維護一個狀態值， 藉助基於AQS實現的鎖ReentrantLock來實現狀態值的同步更新，以及AQS除了同步狀態之外的另一個核心概念條件佇列來完成執行緒的阻塞。

parties： 和CountdownLatch中的狀態值一樣，用來記錄每次要相互等待的執行緒數量，只有parties個執行緒同時到達屏障時，才會喚醒阻塞的執行緒。

count臨時計數器: 由於CyclicBarrier是可以迴圈使用的，count可以理解為是一個臨時變數，每一輪執行完畢或者被打斷都會重置count為parties值。

Generation內部類： 只有一個屬性 broken表示當前這一輪執行是否被中斷，如果被中斷後其他執行緒再執行await方法會丟擲異常（目的是停止本輪執行緒未執行執行緒的繼續執行）。

await方法： 當執行await方法時，會同步得對內部的count執行--count操作， 如果count = 0，則執行barrierCommand任務（通過構造方法傳來的Runnable引數）。

reset方法：中斷本輪執行，重置count值，喚醒等待的執行緒然後開始下一輪，此時本輪正在執行的執行緒呼叫await方法會丟擲異常。

// await方法實際執行的程式碼
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();
        }
        // 本輪執行的計數器 數值-1
        int index = --count;
        if (index == 0) {  // 計數器值=1， 本輪執行緒全部到達屏障，執行barrierCommand任務
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;
                if (command != null)
                    command.run();
                ranAction = true;
                nextGeneration();// 喚醒所有等待在條件佇列上的任務
                return 0;
            } finally {
                if (!ranAction)
                    breakBarrier();
            }
        }

        // 如果狀態不等於0，迴圈等待直到計數器值為0，本輪執行被打破，執行緒被中斷，或者等待超時
        for (;;) {
            try {
                if (!timed)
                    // 狀態碼不為0，將當前執行緒加入到條件佇列中，進入阻塞狀態
                    trip.await();
                else if (nanos > 0L)
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                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.
                    Thread.currentThread().interrupt();
                }
            }

            if (g.broken)
                throw new BrokenBarrierException();

            if (g != generation)
                return index;

            if (timed && nanos <= 0L) {
                breakBarrier();// 喚醒所有條件佇列中的執行緒，重置count的值
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock();
    }
}

重置柵欄的狀態

public void reset() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        breakBarrier();   // break the current generation
        nextGeneration(); // start a new generation
    } finally {
        lock.unlock();
    }
}
/**
 * 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();
}

當一輪執行完畢之後，既count=0後，CyclicBarrier的臨時狀態會重置為parties

/**
 * 進入下一輪
 * 喚醒所有等待執行緒，充值count
 */
private void nextGeneration() {
    // signal completion of last generation
    trip.signalAll();
    // set up next generation
    count = parties;
    generation = new Generation();
}

總結

1. CountDownLatch建立後只能使用一次，而CyclicBarrier可以迴圈使用，並且CyclicBarrier功能更完善。
2. CountDownLatch內部的狀態是基於AQS中的狀態資訊，而CyclicBarrier中的狀態值是單獨維護的，使用ReentrantLock加鎖保證併發修改狀態值的資料一致性。
3. 它們的使用場景：允許一個或多個執行緒等待其他執行緒完成操作， 即當指定數量執行緒執行完某個操作再繼續執行下一個操作。

相關推薦