死磕 java同步系列之CountDownLatch源碼解析

問題

（1）CountDownLatch是什么？

（2）CountDownLatch具有哪些特性？

（3）CountDownLatch通常運(yùn)用在什么場景中？

（4）CountDownLatch的初始次數(shù)是否可以調(diào)整？

簡介

CountDownLatch，可以翻譯為倒計(jì)時(shí)器，但是似乎不太準(zhǔn)確，它的含義是允許一個(gè)或多個(gè)線程等待其它線程的操作執(zhí)行完畢后再執(zhí)行后續(xù)的操作。

CountDownLatch的通常用法和Thread.join()有點(diǎn)類似，等待其它線程都完成后再執(zhí)行主任務(wù)。

類結(jié)構(gòu)

CountDownLatch中只包含了Sync一個(gè)內(nèi)部類，它沒有公平/非公平模式，所以它算是一個(gè)比較簡單的同步器了。

這里還要注意一點(diǎn)，CountDownLatch沒有實(shí)現(xiàn)Serializable接口，所以它不是可序列化的。

源碼分析

內(nèi)部類Sync

private static final class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 4982264981922014374L;

    // 傳入初始次數(shù)
    Sync(int count) {
        setState(count);
    }
    // 獲取還剩的次數(shù)
    int getCount() {
        return getState();
    }
    // 嘗試獲取共享鎖
    protected int tryAcquireShared(int acquires) {
        // 注意，這里state等于0的時(shí)候返回的是1，也就是說count減為0的時(shí)候獲取總是成功
        // state不等于0的時(shí)候返回的是-1，也就是count不為0的時(shí)候總是要排隊(duì)
        return (getState() == 0) ? 1 : -1;
    }
    // 嘗試釋放鎖
    protected boolean tryReleaseShared(int releases) {
        for (;;) {
            // state的值
            int c = getState();
            // 等于0了，則無法再釋放了
            if (c == 0)
                return false;
            // 將count的值減1
            int nextc = c-1;
            // 原子更新state的值
            if (compareAndSetState(c, nextc))
                // 減為0的時(shí)候返回true，這時(shí)會喚醒后面排隊(duì)的線程
                return nextc == 0;
        }
    }
}

Sync重寫了tryAcquireShared()和tryReleaseShared()方法，并把count存到state變量中去。

這里要注意一下，上面兩個(gè)方法的參數(shù)并沒有什么卵用。

構(gòu)造方法

public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}

構(gòu)造方法需要傳入一個(gè)count，也就是初始次數(shù)。

await()方法

// java.util.concurrent.CountDownLatch.await()
public void await() throws InterruptedException {
    // 調(diào)用AQS的acquireSharedInterruptibly()方法
    sync.acquireSharedInterruptibly(1);
}
// java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly()
public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // 嘗試獲取鎖，如果失敗則排隊(duì)
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

await()方法是等待其它線程完成的方法，它會先嘗試獲取一下共享鎖，如果失敗則進(jìn)入AQS的隊(duì)列中排隊(duì)等待被喚醒。

根據(jù)上面Sync的源碼，我們知道，state不等于0的時(shí)候tryAcquireShared()返回的是-1，也就是說count未減到0的時(shí)候所有調(diào)用await()方法的線程都要排隊(duì)。

countDown()方法

// java.util.concurrent.CountDownLatch.countDown()
public void countDown() {
    // 調(diào)用AQS的釋放共享鎖方法
    sync.releaseShared(1);
}
// java.util.concurrent.locks.AbstractQueuedSynchronizer.releaseShared()
public final boolean releaseShared(int arg) {
    // 嘗試釋放共享鎖，如果成功了，就喚醒排隊(duì)的線程
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

countDown()方法，會釋放一個(gè)共享鎖，也就是count的次數(shù)會減1。

根據(jù)上面Sync的源碼，我們知道，tryReleaseShared()每次會把count的次數(shù)減1，當(dāng)其減為0的時(shí)候返回true，這時(shí)候才會喚醒等待的線程。

注意，doReleaseShared()是喚醒等待的線程，這個(gè)方法我們在前面的章節(jié)中分析過了。

使用案例

這里我們模擬一個(gè)使用場景，我們有一個(gè)主線程和5個(gè)輔助線程，等待主線程準(zhǔn)備就緒了，5個(gè)輔助線程開始運(yùn)行，等待5個(gè)輔助線程運(yùn)行完畢了，主線程繼續(xù)往下運(yùn)行，大致的流程圖如下：

我們一起來看看這段代碼應(yīng)該怎么寫：

public class CountDownLatchTest {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch startSignal = new CountDownLatch(1);
        CountDownLatch doneSignal = new CountDownLatch(5);

        for (int i = 0; i < 5; i++) {
            new Thread(()->{
                try {
                    System.out.println("Aid thread is waiting for starting.");
                    startSignal.await();
                    // do sth
                    System.out.println("Aid thread is doing something.");
                    doneSignal.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }

        // main thread do sth
        Thread.sleep(2000);
        System.out.println("Main thread is doing something.");
        startSignal.countDown();

        // main thread do sth else
        System.out.println("Main thread is waiting for aid threads finishing.");
        doneSignal.await();

        System.out.println("Main thread is doing something after all threads have finished.");

    }
}

這段代碼分成兩段：

第一段，5個(gè)輔助線程等待開始的信號，信號由主線程發(fā)出，所以5個(gè)輔助線程調(diào)用startSignal.await()方法等待開始信號，當(dāng)主線程的事兒干完了，調(diào)用startSignal.countDown()通知輔助線程開始干活。

第二段，主線程等待5個(gè)輔助線程完成的信號，信號由5個(gè)輔助線程發(fā)出，所以主線程調(diào)用doneSignal.await()方法等待完成信號，5個(gè)輔助線程干完自己的活兒的時(shí)候調(diào)用doneSignal.countDown()方法發(fā)出自己的完成的信號，當(dāng)完成信號達(dá)到5個(gè)的時(shí)候，喚醒主線程繼續(xù)執(zhí)行后續(xù)的邏輯。

總結(jié)

（1）CountDownLatch表示允許一個(gè)或多個(gè)線程等待其它線程的操作執(zhí)行完畢后再執(zhí)行后續(xù)的操作；

（2）CountDownLatch使用AQS的共享鎖機(jī)制實(shí)現(xiàn)；

（3）CountDownLatch初始化的時(shí)候需要傳入次數(shù)count；

（4）每次調(diào)用countDown()方法count的次數(shù)減1；

（5）每次調(diào)用await()方法的時(shí)候會嘗試獲取鎖，這里的獲取鎖其實(shí)是檢查AQS的state變量的值是否為0；

（6）當(dāng)count的值（也就是state的值）減為0的時(shí)候會喚醒排隊(duì)著的線程（這些線程調(diào)用await()進(jìn)入隊(duì)列）；

彩蛋

（1）CountDownLatch的初始次數(shù)是否可以調(diào)整？

答：前面我們學(xué)習(xí)Semaphore的時(shí)候發(fā)現(xiàn)，它的許可次數(shù)是可以隨時(shí)調(diào)整的，那么，CountDownLatch的初始次數(shù)能隨時(shí)調(diào)整嗎？答案是不能的，它沒有提供修改（增加或減少）次數(shù)的方法，除非使用反射作弊。

（2）CountDownLatch為什么使用共享鎖？

答：前面我們分析ReentrantReadWriteLock的時(shí)候?qū)W習(xí)過AQS的共享鎖模式，比如當(dāng)前鎖是由一個(gè)線程獲取為互斥鎖，那么這時(shí)候所有需要獲取共享鎖的線程都要進(jìn)入AQS隊(duì)列中進(jìn)行排隊(duì)，當(dāng)這個(gè)互斥鎖釋放的時(shí)候，會一個(gè)接著一個(gè)地喚醒這些連續(xù)的排隊(duì)的等待獲取共享鎖的線程，注意，這里的用語是“一個(gè)接著一個(gè)地喚醒”，也就是說這些等待獲取共享鎖的線程不是一次性喚醒的。

說到這里，是不是很明白了？因?yàn)镃ountDownLatch的await()多個(gè)線程可以調(diào)用多次，當(dāng)調(diào)用多次的時(shí)候這些線程都要進(jìn)入AQS隊(duì)列中排隊(duì)，當(dāng)count次數(shù)減為0的時(shí)候，它們都需要被喚醒，繼續(xù)執(zhí)行任務(wù)，如果使用互斥鎖則不行，互斥鎖在多個(gè)線程之間是互斥的，一次只能喚醒一個(gè)，不能保證當(dāng)count減為0的時(shí)候這些調(diào)用了await()方法等待的線程都被喚醒。

（3）CountDownLatch與Thread.join()有何不同？

答：Thread.join()是在主線程中調(diào)用的，它只能等待被調(diào)用的線程結(jié)束了才會通知主線程，而CountDownLatch則不同，它的countDown()方法可以在線程執(zhí)行的任意時(shí)刻調(diào)用，靈活性更大。

推薦閱讀

1、 死磕 java同步系列之開篇

2、 死磕 java魔法類之Unsafe解析

3、 死磕 java同步系列之JMM（Java Memory Model）

4、 死磕 java同步系列之volatile解析

5、 死磕 java同步系列之synchronized解析

6、 死磕 java同步系列之自己動手寫一個(gè)鎖Lock

7、 死磕 java同步系列之AQS起篇

8、 死磕 java同步系列之ReentrantLock源碼解析（一）——公平鎖、非公平鎖

9、 死磕 java同步系列之ReentrantLock源碼解析（二）——條件鎖

10、 死磕 java同步系列之ReentrantLock VS synchronized

11、 死磕 java同步系列之ReentrantReadWriteLock源碼解析

12、 死磕 java同步系列之Semaphore源碼解析

歡迎關(guān)注我的公眾號“彤哥讀源碼”，查看更多源碼系列文章, 與彤哥一起暢游源碼的海洋。