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

時間 2019-11-08

標籤 java 同步系列 countdownlatch 源碼解析欄目 Java 简体版

原文原文鏈接

問題

（1）CountDownLatch是什麼？java

（2）CountDownLatch具備哪些特性？源碼分析

（3）CountDownLatch一般運用在什麼場景中？學習

（4）CountDownLatch的初始次數是否能夠調整？ui

簡介

CountDownLatch，能夠翻譯爲倒計時器，可是彷佛不太準確，它的含義是容許一個或多個線程等待其它線程的操做執行完畢後再執行後續的操做。this

CountDownLatch的一般用法和Thread.join()有點相似，等待其它線程都完成後再執行主任務。線程

類結構

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

這裏還要注意一點，CountDownLatch沒有實現Serializable接口，因此它不是可序列化的。code

源碼分析

內部類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 &lt; 0) throw new IllegalArgumentException("count &lt; 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) &lt; 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 &lt; 5; i++) {
            new Thread(()-&gt;{
                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()方法能夠在線程執行的任意時刻調用，靈活性更大。