[[422507]]

本文轉載自微信公眾號「運維開發(fā)故事」，作者老鄭。轉載本文請聯(lián)系運維開發(fā)故事公眾號。

功能簡介

閉鎖是一種同步工具類，可以延遲線程的進度直到其到達終止狀態(tài)【CPJ 3.4.2】。閉鎖的作用相當于一扇門∶ 在閉鎖到達結束狀態(tài)之前，這扇門一直是關閉的，并且沒有任何線程能通過，當?shù)竭_結束狀態(tài)時，這扇門會打開并允許所有的線程通過。當閉鎖到達結束狀態(tài)后，將不會再改變狀態(tài)，因此這扇門將永遠保持打開狀態(tài)。閉鎖可以用來確保某些活動直到其他活動都完成后才繼續(xù)執(zhí)行，例如∶

• 確保某個計算在其需要的所有資源都被初始化之后才繼續(xù)執(zhí)行。二元閉鎖(包括兩個狀態(tài))可以用來表示"資源R已經(jīng)被初始化"，而所有需要 R 的操作都必須先在這個閉鎖上等待。
• 確保某個服務在其依賴的所有其他服務都已經(jīng)啟動之后才啟動。每個服務都有一個相關的二元閉鎖。當啟動服務S 時，將首先在S依賴的其他服務的閉鎖上等待，在所有依賴的服務都啟動后會釋放閉鎖S，這樣其他依賴 S 的服務才能繼續(xù)執(zhí)行。
• 等待直到某個操作的所有參與者(例如，在多玩家游戲中的所有玩家)都就緒再繼續(xù)執(zhí)行。在這種情況中，當所有玩家都準備就緒時，閉鎖將到達結束狀態(tài)。

CountDownLatch.jpg

CountDownLatch是一種靈活的閉鎖實現(xiàn)，可以在上述各種情況中使用，它可以使一個或多個線程等待一組事件發(fā)生。閉鎖狀態(tài)包括一個計數(shù)器，該計數(shù)器被初始化為一個正數(shù)，表示需要等待的事件數(shù)量。countDown方法遞減計數(shù)器，表示有一個事件已經(jīng)發(fā)生了，而 await方法等待計數(shù)器達到零，這表示所有需要等待的事件都已經(jīng)發(fā)生。如果計數(shù)器的值非零，那么 await 會一直阻塞直到計數(shù)器為零，或者等待中的線程中斷，或者等待超時。

使用案例

TestHarness 中給出了閉鎖的兩種常見用法。TestHarness 創(chuàng)建一定數(shù)量的線程，利用它們并發(fā)地執(zhí)行指定的任務。它使用兩個閉鎖，分別表示"起始門(Starting Gate)"和"結束門(Ending Gate)"。起始門計數(shù)器的初始值為1，而結束門計數(shù)器的初始值為工作線程的數(shù)量。每個工作線程首先要做的就是在啟動門上等待，從而確保所有線程都就緒后才開始執(zhí)行。而每個線程要做的最后一件事情是將調用結束門的 countDown 方法減1，這能使主線程高效地等待直到所有工作線程都執(zhí)行完成，因此可以統(tǒng)計所消耗的時間。

public class TestHarness { 
 
    public long timeTasks(int nThreads, final Runnable task) throws InterruptedException { 
        final CountDownLatch startGate = new CountDownLatch(1); 
        final CountDownLatch endGate = new CountDownLatch(nThreads); 
 
        for (int i = 0; i < nThreads; i++) { 
            Thread t = new Thread(() -> { 
                try { 
                    startGate.await(); 
                    try { 
                        task.run(); 
                    } finally { 
                        endGate.countDown(); 
                    } 
 
                } catch (InterruptedException ignored) { 
 
                } 
            }); 
            t.start(); 
        } 
 
        long start = System.nanoTime(); 
        startGate.countDown(); 
        endGate.await(); 
        long end = System.nanoTime(); 
        return end - start; 
    } 
 
    public static void main(String[] args) throws InterruptedException { 
        TestHarness testHarness = new TestHarness(); 
        AtomicInteger num = new AtomicInteger(0); 
        long time = testHarness.timeTasks(10, () -> System.out.println(num.incrementAndGet())); 
        System.out.println("cost time: " + time + "ms"); 
    } 
} 
 
//輸出結果 
1 
10 
9 
8 
7 
5 
6 
4 
3 
2 
cost time: 2960900ms 

為什么要在 TestHarness 中使用閉鎖，而不是在線程創(chuàng)建后就立即啟動? 或許，我們希望測試 n 個線程并發(fā)執(zhí)行某個任務時需要的時間。如果在創(chuàng)建線程后立即啟動它們，那么先啟動的線程將"領先"后啟動的線程，并且活躍線程數(shù)量會隨著時間的推移而增加或減少，競爭程度也在不斷發(fā)生變化。啟動門將使得主線程能夠實時釋放所有工作線程，而結束門則使主線程能夠等待最后一個線程執(zhí)行完成，而不是順序地等待每個線程執(zhí)行完成。

使用總結

CountDownLatch 是一次性的，計算器的值只能在構造方法中初始化一次，之后沒有任何機制再次對其設置值，當CountDownLatch 使用完畢后，它不能再次被使用。

源碼分析

代碼分析

CountDownLatch　在底層還是采用 AbstractQueuedSynchronizer 實現(xiàn)。

CountDownLatch startGate = **new **CountDownLatch(1); 

我們先看它的構造方法, 創(chuàng)建了一個 sync 對象。

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

Sync 是 AbstractQueuedSynchronizer 的一個實現(xiàn), 按照字面意思我們可以猜到它是公平方式實現(xiàn)。

private static final class Sync extends AbstractQueuedSynchronizer { 
    private static final long serialVersionUID = 4982264981922014374L; 
 
    // 構造方法 
    Sync(int count) { 
        setState(count); 
    } 
 
    // 獲取資源數(shù) 
    int getCount() { 
        return getState(); 
    } 
 
    // 獲取鎖 
    protected int tryAcquireShared(int acquires) { 
        return (getState() == 0) ? 1 : -1; 
    } 
 
    // 釋放鎖 
    protected boolean tryReleaseShared(int releases) { 
        // Decrement count; signal when transition to zero 
        for (;;) { 
            int c = getState(); 
            if (c == 0) 
                return false; 
            int nextc = c-1; 
            // CAS 解鎖 
            if (compareAndSetState(c, nextc)) 
                return nextc == 0; 
        } 
    } 
} 

在 await 方法中如果存在計算值, 那么當前線程將進入 AQS 隊列生成 Node 節(jié)點, 線程進入阻塞狀態(tài)。

public void await() throws InterruptedException { 
    sync.acquireSharedInterruptibly(1); 
} 

其實主要是獲取共享鎖。

public final void acquireSharedInterruptibly(int arg) 
    throws InterruptedException { 
    if (Thread.interrupted()) 
        throw new InterruptedException(); 
    if (tryAcquireShared(arg) < 0) 
        doAcquireSharedInterruptibly(arg); 
} 

CountDownLatch.Sync 實現(xiàn)了 tryAcquireShared 方法 ,如果 getState() == 0 返回 1 , 否則返回 -1. 也就是說創(chuàng)建 CountDownLatch 實例后再執(zhí)行 await 方法將繼續(xù)調用 doAcquireSharedInterruptibly(arg);

// 是否可獲取共享鎖 
protected int tryAcquireShared(int acquires) { 
    return (getState() == 0) ? 1 : -1; 
} 
 
 
// 嘗試獲取鎖, 或者入隊 
private void doAcquireSharedInterruptibly(int arg) 
    throws InterruptedException { 
    final Node node = addWaiter(Node.SHARED); 
    boolean failed = true; 
    try { 
        for (;;) { 
            final Node p = node.predecessor(); 
            if (p == head) { 
                int r = tryAcquireShared(arg); 
                if (r >= 0) { 
                    setHeadAndPropagate(node, r); 
                    p.next = null; // help GC 
                    failed = false; 
                    return; 
                } 
            } 
            if (shouldParkAfterFailedAcquire(p, node) && 
                parkAndCheckInterrupt()) 
                throw new InterruptedException(); 
        } 
    } finally { 
        if (failed) 
            cancelAcquire(node); 
    } 
} 

在 countDown 方法如果存在等待的線程, 將對其進行喚醒. 或者減少 CountDownLatch 資源數(shù)。

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

通過 releaseShared 對共享鎖進行解鎖。 

public final boolean releaseShared(int arg) { 
    if (tryReleaseShared(arg)) { 
        doReleaseShared(); 
        return true; 
    } 
    return false; 
} 

最終會調用 doReleaseShared 喚醒 AQS 中的頭節(jié)點。

private void doReleaseShared() { 
    /* 
         * Ensure that a release propagates, even if there are other 
         * in-progress acquires/releases.  This proceeds in the usual 
         * way of trying to unparkSuccessor of head if it needs 
         * signal. But if it does not, status is set to PROPAGATE to 
         * ensure that upon release, propagation continues. 
         * Additionally, we must loop in case a new node is added 
         * while we are doing this. Also, unlike other uses of 
         * unparkSuccessor, we need to know if CAS to reset status 
         * fails, if so rechecking. 
         */ 
    for (;;) { 
        Node h = head; 
        if (h != null && h != tail) { 
            int ws = h.waitStatus; 
            if (ws == Node.SIGNAL) { 
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) 
                    continue;            // loop to recheck cases 
                unparkSuccessor(h); 
            } 
            else if (ws == 0 && 
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) 
                continue;                // loop on failed CAS 
        } 
        if (h == head)                   // loop if head changed 
            break; 
    } 
} 

詳細流程如下圖：

源碼流程圖

CountDownLatch 閉鎖源碼分析.png

參考資料

《Java 并發(fā)編程實戰(zhàn)》 

https://www.cnblogs.com/Lee_xy_z/p/10470181.html