摘要：在創(chuàng)建對(duì)象時(shí)，需要轉(zhuǎn)入一個(gè)值，用于初始化的成員變量，該成員變量表示屏障攔截的線程數(shù)。當(dāng)?shù)竭_(dá)屏障的線程數(shù)小于時(shí)，這些線程都會(huì)被阻塞住。當(dāng)所有線程到達(dá)屏障后，將會(huì)被更新，表示進(jìn)入新一輪的運(yùn)行輪次中。

在分析完AbstractQueuedSynchronizer（以下簡(jiǎn)稱 AQS）和ReentrantLock的原理后，本文將分析 java.util.concurrent 包下的兩個(gè)線程同步組件CountDownLatch和CyclicBarrier。這兩個(gè)同步組件比較常用，也經(jīng)常被放在一起對(duì)比。通過分析這兩個(gè)同步組件，可使我們對(duì) Java 線程間協(xié)同有更深入的了解。同時(shí)通過分析其原理，也可使我們做到知其然，并知其所以然。

這里首先來介紹一下 CountDownLatch 的用途，CountDownLatch 允許一個(gè)或一組線程等待其他線程完成后再恢復(fù)運(yùn)行。線程可通過調(diào)用await方法進(jìn)入等待狀態(tài)，在其他線程調(diào)用countDown方法將計(jì)數(shù)器減為0后，處于等待狀態(tài)的線程即可恢復(fù)運(yùn)行。CyclicBarrier （可循環(huán)使用的屏障）則與此不同，CyclicBarrier 允許一組線程到達(dá)屏障后阻塞住，直到最后一個(gè)線程進(jìn)入到達(dá)屏障，所有線程才恢復(fù)運(yùn)行。它們之間主要的區(qū)別在于喚醒等待線程的時(shí)機(jī)。CountDownLatch 是在計(jì)數(shù)器減為0后，喚醒等待線程。CyclicBarrier 是在計(jì)數(shù)器（等待線程數(shù)）增長(zhǎng)到指定數(shù)量后，再喚醒等待線程。除此之外，兩種之間還有一些其他的差異，這個(gè)將會(huì)在后面進(jìn)行說明。

在下一章中，我將會(huì)介紹一下兩者的實(shí)現(xiàn)原理，繼續(xù)往下看吧。

CountDownLatch 的同步功能是基于 AQS 實(shí)現(xiàn)的，CountDownLatch 使用 AQS 中的 state 成員變量作為計(jì)數(shù)器。在 state 不為0的情況下，凡是調(diào)用 await 方法的線程將會(huì)被阻塞，并被放入 AQS 所維護(hù)的同步隊(duì)列中進(jìn)行等待。大致示意圖如下：

每個(gè)阻塞的線程都會(huì)被封裝成節(jié)點(diǎn)對(duì)象，節(jié)點(diǎn)之間通過 prev 和 next 指針形成同步隊(duì)列。初始情況下，隊(duì)列的頭結(jié)點(diǎn)是一個(gè)虛擬節(jié)點(diǎn)。該節(jié)點(diǎn)僅是一個(gè)占位符，沒什么特別的意義。每當(dāng)有一個(gè)線程調(diào)用 countDown 方法，就將計(jì)數(shù)器 state--。當(dāng) state 被減至0時(shí)，隊(duì)列中的節(jié)點(diǎn)就會(huì)按照 FIFO 順序被喚醒，被阻塞的線程即可恢復(fù)運(yùn)行。

CountDownLatch 本身的原理并不難理解，不過如果大家想深入理解 CountDownLatch 的實(shí)現(xiàn)細(xì)節(jié)，那么需要先去學(xué)習(xí)一下 AQS 的相關(guān)原理。CountDownLatch 是基于 AQS 實(shí)現(xiàn)的，所以理解 AQS 是學(xué)習(xí) CountDownLatch 的前置條件。我在之前寫過一篇關(guān)于 AQS 的文章 Java 重入鎖 ReentrantLock 原理分析，有興趣的朋友可以去讀一讀。

與 CountDownLatch 的實(shí)現(xiàn)方式不同，CyclicBarrier 并沒有直接通過 AQS 實(shí)現(xiàn)同步功能，而是在重入鎖 ReentrantLock 的基礎(chǔ)上實(shí)現(xiàn)的。在 CyclicBarrier 中，線程訪問 await 方法需先獲取鎖才能訪問。在最后一個(gè)線程訪問 await 方法前，其他線程進(jìn)入 await 方法中后，會(huì)調(diào)用 Condition 的 await 方法進(jìn)入等待狀態(tài)。在最后一個(gè)線程進(jìn)入 CyclicBarrier await 方法后，該線程將會(huì)調(diào)用 Condition 的 signalAll 方法喚醒所有處于等待狀態(tài)中的線程。同時(shí)，最后一個(gè)進(jìn)入 await 的線程還會(huì)重置 CyclicBarrier 的狀態(tài)，使其可以重復(fù)使用。

在創(chuàng)建 CyclicBarrier 對(duì)象時(shí)，需要轉(zhuǎn)入一個(gè)值，用于初始化 CyclicBarrier 的成員變量 parties，該成員變量表示屏障攔截的線程數(shù)。當(dāng)?shù)竭_(dá)屏障的線程數(shù)小于 parties 時(shí)，這些線程都會(huì)被阻塞住。當(dāng)最后一個(gè)線程到達(dá)屏障后，此前被阻塞的線程才會(huì)被喚醒。

通過前面簡(jiǎn)單的分析，相信大家對(duì) CountDownLatch 和 CyclicBarrier 的原理有一定的了解了。那么接下來趁熱打鐵，我們一起探索一下這兩個(gè)同步組件的具體實(shí)現(xiàn)吧。

CountDownLatch 的原理不是很復(fù)雜，所以在具體的實(shí)現(xiàn)上，也不是很復(fù)雜。當(dāng)然，前面說過 CountDownLatch 是基于 AQS 實(shí)現(xiàn)的，AQS 的實(shí)現(xiàn)則要復(fù)雜的多。不過這里僅要求大家掌握 AQS 的基本原理，知道它內(nèi)部維護(hù)了一個(gè)同步隊(duì)列，同步隊(duì)列中的線程會(huì)按照 FIFO 依次獲取同步狀態(tài)就行了。好了，下面我們一起去看一下 CountDownLatch 的源碼吧。

CountDownLatch 的代碼量不大，加上注釋也不過300多行，所以它的代碼結(jié)構(gòu)也會(huì)比較簡(jiǎn)單。如下：

如上圖，CountDownLatch 源碼包含一個(gè)構(gòu)造方法和一個(gè)私有成員變量，以及數(shù)個(gè)普通方法和一個(gè)重要的靜態(tài)內(nèi)部類 Sync。CountDownLatch 的主要邏輯都是封裝在 Sync 和其父類 AQS 里的。所以分析 CountDownLatch 的源碼，本質(zhì)上是分析 Sync 和 AQS 的原理。相關(guān)的分析，將會(huì)在下一節(jié)中展開，本節(jié)先說到這。

本節(jié)來分析一下 CountDownLatch 的構(gòu)造方法和其 Sync 類型的成員變量實(shí)現(xiàn)，如下：

public class CountDownLatch {

    private final Sync sync;

    /** CountDownLatch 的構(gòu)造方法，該方法要求傳入大于0的整型數(shù)值作為計(jì)數(shù)器 */
    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        // 初始化 Sync
        this.sync = new Sync(count);
    }
    
    /** CountDownLatch 的同步控制器，繼承自 AQS */
    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
            // 設(shè)置 AQS state
            setState(count);
        }

        int getCount() {
            return getState();
        }

        /** 嘗試在共享狀態(tài)下獲取同步狀態(tài)，該方法在 AQS 中是抽象方法，這里進(jìn)行了覆寫 */
        protected int tryAcquireShared(int acquires) {
            /*
             * 如果 state = 0，則返回1，表明可獲取同步狀態(tài)，
             * 此時(shí)線程調(diào)用 await 方法時(shí)就不會(huì)被阻塞。
             */ 
            return (getState() == 0) ? 1 : -1;
        }

        /** 嘗試在共享狀態(tài)下釋放同步狀態(tài)，該方法在 AQS 中也是抽象方法 */
        protected boolean tryReleaseShared(int releases) {
            /*
             * 下面的邏輯是將 state--，state 減至0時(shí)，調(diào)用 await 等待的線程會(huì)被喚醒。
             * 這里使用循環(huán) + CAS，表明會(huì)存在競(jìng)爭(zhēng)的情況，也就是多個(gè)線程可能會(huì)同時(shí)調(diào)用 
             * countDown 方法。在 state 不為0的情況下，線程調(diào)用 countDown 是必須要完
             * 成 state-- 這個(gè)操作。所以這里使用了循環(huán) + CAS，確保 countDown 方法可正
             * 常運(yùn)行。
             */
            for (;;) {
                // 獲取 state
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                
                // 使用 CAS 設(shè)置新的 state 值
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }
}

需要說明的是，Sync 中的 tryAcquireShared 和 tryReleaseShared 方法并不是直接給 await 和 countDown 方法調(diào)用了的，這兩個(gè)方法以“try”開頭的方法最終會(huì)在 AQS 中被調(diào)用。

CountDownLatch中有兩個(gè)版本的 await 方法，一個(gè)響應(yīng)中斷，另一個(gè)在此基礎(chǔ)上增加了超時(shí)功能。本節(jié)將分析無超時(shí)功能的 await，如下：

/** 
 * 該方法會(huì)使線程進(jìn)入等待狀態(tài)，直到計(jì)數(shù)器減至0，或者線程被中斷。當(dāng)計(jì)數(shù)器為0時(shí)，調(diào)用
 * 此方法將會(huì)立即返回，不會(huì)被阻塞住。
 */
public void await() throws InterruptedException {
    // 調(diào)用 AQS 中的 acquireSharedInterruptibly 方法
    sync.acquireSharedInterruptibly(1);
}

/** 帶有超時(shí)功能的 await */
public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

+--- AbstractQueuedSynchronizer
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
    // 若線程被中斷，則直接拋出中斷異常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 調(diào)用 Sync 中覆寫的 tryAcquireShared 方法，嘗試獲取同步狀態(tài)
    if (tryAcquireShared(arg) < 0)
        /*
         * 若 tryAcquireShared 小于0，則表示獲取同步狀態(tài)失敗，
         * 此時(shí)將線程放入 AQS 的同步隊(duì)列中進(jìn)行等待。
         */ 
        doAcquireSharedInterruptibly(arg);
}

從上面的代碼中可以看出，CountDownLatch await 方法實(shí)際上調(diào)用的是 AQS 的 acquireSharedInterruptibly 方法。該方法會(huì)在內(nèi)部調(diào)用 Sync 所覆寫的 tryAcquireShared 方法。在 state != 0時(shí)，tryAcquireShared 返回值 -1。此時(shí)線程將進(jìn)入 doAcquireSharedInterruptibly 方法中，在此方法中，線程會(huì)被放入同步隊(duì)列中進(jìn)行等待。若 state = 0，此時(shí) tryAcquireShared 返回1，acquireSharedInterruptibly 會(huì)直接返回。此時(shí)調(diào)用 await 的線程也不會(huì)被阻塞住。

與 await 方法一樣，countDown 實(shí)際上也是對(duì) AQS 方法的一層封裝。具體的實(shí)現(xiàn)如下：

/** 該方法的作用是將計(jì)數(shù)器進(jìn)行自減操作，當(dāng)計(jì)數(shù)器為0時(shí)，喚醒正在同步隊(duì)列中等待的線程 */
public void countDown() {
    // 調(diào)用 AQS 中的 releaseShared 方法
    sync.releaseShared(1);
}

+--- AbstractQueuedSynchronizer
public final boolean releaseShared(int arg) {
    // 調(diào)用 Sync 中的 tryReleaseShared 嘗試釋放同步狀態(tài)
    if (tryReleaseShared(arg)) {
        /*
         * tryReleaseShared 返回 true 時(shí)，表明 state = 0，即計(jì)數(shù)器為0。此時(shí)調(diào)用
         * doReleaseShared 方法喚醒正在同步隊(duì)列中等待的線程
         */ 
        doReleaseShared();
        return true;
    }
    return false;
}

以上就是 countDown 的源碼分析，不是很難懂，這里就不啰嗦了。

如前面所說，CyclicBarrier 是基于重入鎖 ReentrantLock 實(shí)現(xiàn)相關(guān)邏輯的。所以要弄懂 CyclicBarrier 的源碼，僅需有 ReentrantLock 相關(guān)的背景知識(shí)即可。關(guān)于重入鎖 ReentrantLock 方面的知識(shí)，有興趣的朋友可以參考我之前寫的文章 Java 重入鎖 ReentrantLock 原理分析。下面看一下 CyclicBarrier 的代碼結(jié)構(gòu)吧，如下：

從上圖可以看出，CyclicBarrier 包含了一個(gè)靜態(tài)內(nèi)部類Generation、數(shù)個(gè)方法和一些成員變量。結(jié)構(gòu)上比 CountDownLatch 略為復(fù)雜一些，但總體仍比較簡(jiǎn)單。好了，接下來進(jìn)入源碼分析部分吧。

CyclicBarrier 包含兩個(gè)有參構(gòu)造方法，分別如下：

/** 創(chuàng)建一個(gè)允許 parties 個(gè)線程通行的屏障 */
public CyclicBarrier(int parties) {
    this(parties, null);
}

/** 
 * 創(chuàng)建一個(gè)允許 parties 個(gè)線程通行的屏障，若 barrierAction 回調(diào)對(duì)象不為 null，
 * 則在最后一個(gè)線程到達(dá)屏障后，執(zhí)行相應(yīng)的回調(diào)邏輯
 */
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

上面的第二個(gè)構(gòu)造方法初始化了一些成員變量，下面我們就來說明一下這些成員變量的作用。

除了上面幾個(gè)成員變量，還有一個(gè)成員變量需要說明一下，如下：

/** 
 * CyclicBarrier 是可循環(huán)使用的屏障，這里使用 Generation 記錄當(dāng)前輪次 CyclicBarrier 
 * 的運(yùn)行狀態(tài)。當(dāng)所有線程到達(dá)屏障后，generation 將會(huì)被更新，表示 CyclicBarrier 進(jìn)入新一
 * 輪的運(yùn)行輪次中。
*/
private Generation generation = new Generation();

private static class Generation {
    // 用于記錄屏障有沒有被破壞
    boolean broken = false;
}

上一節(jié)所提到的幾個(gè)成員變量，在 await 方法中將會(huì)悉數(shù)登場(chǎng)。下面就來分析一下 await 方法的試下，如下：

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        // await 的邏輯封裝在 dowait 中
        return dowait(false, 0L);
    } catch (TimeoutException toe) {
        throw new Error(toe); // cannot happen
    }
}
    
private int dowait(boolean timed, long nanos)
    throws InterruptedException, BrokenBarrierException, TimeoutException {
    
    final ReentrantLock lock = this.lock;
    // 加鎖
    lock.lock();
    try {
        final Generation g = generation;

        // 如果 g.broken = true，表明屏障被破壞了，這里直接拋出異常
        if (g.broken)
            throw new BrokenBarrierException();

        // 如果線程中斷，則調(diào)用 breakBarrier 破壞屏障
        if (Thread.interrupted()) {
            breakBarrier();
            throw new InterruptedException();
        }

        /*
         * index 表示線程到達(dá)屏障的順序，index = parties - 1 表明當(dāng)前線程是第一個(gè)
         * 到達(dá)屏障的。index = 0，表明當(dāng)前線程是最有一個(gè)到達(dá)屏障的。
         */ 
        int index = --count;
        // 當(dāng) index = 0 時(shí)，喚醒所有處于等待狀態(tài)的線程
        if (index == 0) {  // tripped
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;
                // 如果回調(diào)對(duì)象不為 null，則執(zhí)行回調(diào)
                if (command != null)
                    command.run();
                ranAction = true;
                // 重置屏障狀態(tài)，使其進(jìn)入新一輪的運(yùn)行過程中
                nextGeneration();
                return 0;
            } finally {
                // 若執(zhí)行回調(diào)的過程中發(fā)生異常，此時(shí)調(diào)用 breakBarrier 破壞屏障
                if (!ranAction)
                    breakBarrier();
            }
        }

        // 線程運(yùn)行到此處的線程都會(huì)被屏障擋住，并進(jìn)入等待狀態(tài)。
        for (;;) {
            try {
                if (!timed)
                    trip.await();
                else if (nanos > 0L)
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                /*
                 * 若下面的條件成立，則表明本輪運(yùn)行還未結(jié)束。此時(shí)調(diào)用 breakBarrier 
                 * 破壞屏障，喚醒其他線程，并拋出異常
                 */ 
                if (g == generation && ! g.broken) {
                    breakBarrier();
                    throw ie;
                } else {
                    /*
                     * 若上面的條件不成立，則有兩種可能：
                     * 1. g != generation
                     *     此種情況下，表明循環(huán)屏障的第 g 輪次的運(yùn)行已經(jīng)結(jié)束，屏障已經(jīng)
                     *     進(jìn)入了新的一輪運(yùn)行輪次中。當(dāng)前線程在稍后返回 到達(dá)屏障 的順序即可
                     *     
                     * 2. g = generation 但 g.broken = true
                     *     此種情況下，表明已經(jīng)有線程執(zhí)行過 breakBarrier 方法了，當(dāng)前
                     *     線程則會(huì)在稍后拋出 BrokenBarrierException
                     */
                    Thread.currentThread().interrupt();
                }
            }
            
            // 屏障被破壞，則拋出 BrokenBarrierException 異常
            if (g.broken)
                throw new BrokenBarrierException();

            // 屏障進(jìn)入新的運(yùn)行輪次，此時(shí)返回線程在上一輪次到達(dá)屏障的順序
            if (g != generation)
                return index;

            // 超時(shí)判斷
            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock();
    }
}
    
/** 開啟新的一輪運(yùn)行過程 */
private void nextGeneration() {
    // 喚醒所有處于等待狀態(tài)中的線程
    trip.signalAll();
    // 重置 count
    count = parties;
    // 重新創(chuàng)建 Generation，表明進(jìn)入循環(huán)屏障進(jìn)入新的一輪運(yùn)行輪次中
    generation = new Generation();
}

/** 破壞屏障 */
private void breakBarrier() {
    // 設(shè)置屏障是否被破壞標(biāo)志
    generation.broken = true;
    // 重置 count
    count = parties;
    // 喚醒所有處于等待狀態(tài)中的線程
    trip.signalAll();
}

reset 方法用于強(qiáng)制重置屏障，使屏障進(jìn)入新一輪的運(yùn)行過程中。代碼如下：

public void reset() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 破壞屏障
        breakBarrier();   // break the current generation
        // 開啟新一輪的運(yùn)行過程
        nextGeneration(); // start a new generation
    } finally {
        lock.unlock();
    }
}

reset 方法并不復(fù)雜，沒什么好講的。CyclicBarrier 中還有其他一些方法，均不復(fù)雜，這里就不一一分析了。

看完上面的分析，相信大家對(duì)著兩個(gè)同步組件有了更深入的認(rèn)識(shí)。那么下面趁熱打鐵，簡(jiǎn)單對(duì)比一下兩者之間的區(qū)別。這里用一個(gè)表格列舉一下：

除了上面列舉的差異點(diǎn)，還有一些其他方面的差異，這里就不一一列舉了。

分析完 CountDownLatch 和 CyclicBarrier，不知道大家有什么感覺。我個(gè)人的感覺是這兩個(gè)類的源碼并不復(fù)雜，比較好理解。當(dāng)然，前提是建立在對(duì) AQS 以及 ReentrantLock 有較深的理解之上。所以在學(xué)習(xí)這兩個(gè)類的源碼時(shí)，還是建議大家先看看前置知識(shí)。

好了，本文到這里就結(jié)束了。謝謝閱讀，再見。

本文在知識(shí)共享許可協(xié)議 4.0 下發(fā)布，轉(zhuǎn)載需在明顯位置處注明出處
作者：coolblog
本文同步發(fā)布在我的個(gè)人博客：http://www.coolblog.xyz

本作品采用知識(shí)共享署名-非商業(yè)性使用-禁止演繹 4.0 國(guó)際許可協(xié)議進(jìn)行許可。