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

問題

（1）CountDownLatch是什么？

（2）CountDownLatch具有哪些特性？

（3）CountDownLatch通常運用在什么場景中？

（4）CountDownLatch的初始次數是否可以調整？

簡介

CountDownLatch，可以翻譯為倒計時器，但是似乎不太準確，它的含義是允許一個或多個線程等待其它線程的操作執行完畢后再執行后續的操作。

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

類結構

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

這里還要注意一點，CountDownLatch沒有實現Serializable接口，所以它不是可序列化的。

源碼分析

內部類Sync

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

    // 傳入初始次數
    Sync(int count) {
        setState(count);
    }
    // 獲取還剩的次數
    int getCount() {
        return getState();
    }
    // 嘗試獲取共享鎖
    protected int tryAcquireShared(int acquires) {
        // 注意，這里state等于0的時候返回的是1，也就是說count減為0的時候獲取總是成功
        // state不等于0的時候返回的是-1，也就是count不為0的時候總是要排隊
        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的時候返回true，這時會喚醒后面排隊的線程
                return nextc == 0;
        }
    }
}

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

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

構造方法

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

構造方法需要傳入一個count，也就是初始次數。

await()方法

// java.util.concurrent.CountDownLatch.await()
public void await() throws InterruptedException {
    // 調用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();
    // 嘗試獲取鎖，如果失敗則排隊
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

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

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

countDown()方法

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

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

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

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

使用案例

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

我們一起來看看這段代碼應該怎么寫：

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個輔助線程等待開始的信號，信號由主線程發出，所以5個輔助線程調用startSignal.await()方法等待開始信號，當主線程的事兒干完了，調用startSignal.countDown()通知輔助線程開始干活。

第二段，主線程等待5個輔助線程完成的信號，信號由5個輔助線程發出，所以主線程調用doneSignal.await()方法等待完成信號，5個輔助線程干完自己的活兒的時候調用doneSignal.countDown()方法發出自己的完成的信號，當完成信號達到5個的時候，喚醒主線程繼續執行后續的邏輯。

總結

（1）CountDownLatch表示允許一個或多個線程等待其它線程的操作執行完畢后再執行后續的操作；

（2）CountDownLatch使用AQS的共享鎖機制實現；

（3）CountDownLatch初始化的時候需要傳入次數count；

（4）每次調用countDown()方法count的次數減1；

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

（6）當count的值（也就是state的值）減為0的時候會喚醒排隊著的線程（這些線程調用await()進入隊列）；

彩蛋

（1）CountDownLatch的初始次數是否可以調整？

答：前面我們學習Semaphore的時候發現，它的許可次數是可以隨時調整的，那么，CountDownLatch的初始次數能隨時調整嗎？答案是不能的，它沒有提供修改（增加或減少）次數的方法，除非使用反射作弊。

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

答：前面我們分析ReentrantReadWriteLock的時候學習過AQS的共享鎖模式，比如當前鎖是由一個線程獲取為互斥鎖，那么這時候所有需要獲取共享鎖的線程都要進入AQS隊列中進行排隊，當這個互斥鎖釋放的時候，會一個接著一個地喚醒這些連續的排隊的等待獲取共享鎖的線程，注意，這里的用語是“一個接著一個地喚醒”，也就是說這些等待獲取共享鎖的線程不是一次性喚醒的。

說到這里，是不是很明白了？因為CountDownLatch的await()多個線程可以調用多次，當調用多次的時候這些線程都要進入AQS隊列中排隊，當count次數減為0的時候，它們都需要被喚醒，繼續執行任務，如果使用互斥鎖則不行，互斥鎖在多個線程之間是互斥的，一次只能喚醒一個，不能保證當count減為0的時候這些調用了await()方法等待的線程都被喚醒。

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

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

推薦閱讀

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

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

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

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

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

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

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

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

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

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

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

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

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