Java中的CountDownLatch原理深入解析

更新時(shí)間：2024年01月22日 10:21:28 作者：我不是歐拉_

這篇文章主要介紹了Java中的CountDownLatch原理深入解析,CountDownLatch是多線程控制的一種同步工具類(lèi),它被稱為門(mén)閥、計(jì)數(shù)器或者閉鎖,這個(gè)工具經(jīng)常用來(lái)用來(lái)協(xié)調(diào)多個(gè)線程之間的同步,或者說(shuō)起到線程之間的通信,需要的朋友可以參考下

1. CountDownLatch是什么？

CountDownLatch是多線程控制的一種同步工具類(lèi)，它被稱為門(mén)閥、計(jì)數(shù)器或者閉鎖。這個(gè)工具經(jīng)常用來(lái)用來(lái)協(xié)調(diào)多個(gè)線程之間的同步，或者說(shuō)起到線程之間的通信（而不是用作互斥的作用）。

它允許一個(gè)或多個(gè)線程一直等待，直到其他線程執(zhí)行完后再執(zhí)行。例如，應(yīng)用程序的主線程希望在負(fù)責(zé)啟動(dòng)框架服務(wù)的線程已經(jīng)啟動(dòng)所有框架服務(wù)之后執(zhí)行。

當(dāng)然利用ReentrantLock + Condition也可以實(shí)現(xiàn)線程之間通信，達(dá)到同樣的效果

2. 類(lèi)圖

可以看出CountDownLatch只有一個(gè)內(nèi)部類(lèi)Sync，Sync繼承AbstractQueuedSynchronizer

3. 實(shí)現(xiàn)原理

3.1 示例用法

// N個(gè)線程等待主線程
class Driver { // ...
    void main() throws InterruptedException {
        // 開(kāi)始信號(hào)
        CountDownLatch startSignal = new CountDownLatch(1);
        // 完成信號(hào)
        CountDownLatch doneSignal = new CountDownLatch(N);
        for (int i = 0; i < N; ++i) // create and start threads
        // 創(chuàng)建N個(gè)工作線程并開(kāi)始運(yùn)行
        new Thread(new Worker(startSignal, doneSignal)).start();
        // 做準(zhǔn)備工作
        doSomethingElse();            // don't let run yet
        // 準(zhǔn)備完畢，喚醒工作線程
        startSignal.countDown();      // let all threads proceed
        doSomethingElse();
        // 等待工作線程結(jié)束
        doneSignal.await();           // wait for all to finish
    }
  }
class Worker implements Runnable {
    private final CountDownLatch startSignal;
    private final CountDownLatch doneSignal;
    // 構(gòu)造方法創(chuàng)建工作線程
    Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
        this.startSignal = startSignal;
        this.doneSignal = doneSignal;
    }
    public void run() {
        try {
            // 工作線程進(jìn)入等待狀態(tài)
            startSignal.await();
            // 工作線程工作
            doWork();
            // 完成工作后，countDown
            doneSignal.countDown();
        } catch (InterruptedException ex) {} // return;
    }
void doWork() { ... }
}

// 主線程等到N個(gè)線程
class Driver2 { // ...
    void main() throws InterruptedException {
        // 完成信號(hào)
        CountDownLatch doneSignal = new CountDownLatch(N);
        // 創(chuàng)建線程執(zhí)行器
        Executor e = ...
        for (int i = 0; i < N; ++i) // create and start threads
            // 創(chuàng)建并執(zhí)行N個(gè)準(zhǔn)備工作線程
            e.execute(new WorkerRunnable(doneSignal, i));
            // 主線程等到準(zhǔn)備工作線程執(zhí)行完畢
            doneSignal.await();           // wait for all to finish
        }
}
class WorkerRunnable implements Runnable {
    private final CountDownLatch doneSignal;
    private final int i;
    // 構(gòu)造方法
    WorkerRunnable(CountDownLatch doneSignal, int i) {
        this.doneSignal = doneSignal;
        this.i = i;
    }
    // run
    public void run() {
        try {
            // 完成準(zhǔn)備工作
            doWork(i);
            // countDown
            doneSignal.countDown();
        } catch (InterruptedException ex) {} // return;
    }
void doWork() { ... }
}

3.2 Sync

private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;
        // State即同步狀態(tài)，在不同的實(shí)現(xiàn)中叫法不一樣，只是為了方便理解
        // 構(gòu)造方法初始化計(jì)數(shù)器計(jì)數(shù)值（即同步狀態(tài)值）
        Sync(int count) {
            setState(count);
        }
        // 獲取計(jì)數(shù)值
        int getCount() {
            return getState();
        }
        // 共享模式獲取
        protected int tryAcquireShared(int acquires) {
            // 體現(xiàn)出只有計(jì)數(shù)值為0時(shí)，才能算獲取成功
            return (getState() == 0) ? 1 : -1;
        }
        // 共享模式釋放
        protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
                int c = getState();
                // 如果計(jì)數(shù)值已經(jīng)為0，直接返回false，結(jié)束自旋
                if (c == 0)
                    return false;
                // 否則計(jì)數(shù) - 1
                int nextc = c-1;
                // 通過(guò)自旋 + CAS方式改變剩余計(jì)數(shù)
                if (compareAndSetState(c, nextc))
                    // 如果計(jì)數(shù)為0返回true，否則返回false，結(jié)束自旋
                    // 返回true表示可以喚醒等待的線程
                    return nextc == 0;
            }
        }
    }

通過(guò)上面代碼解析可知， CountDownLatch的實(shí)現(xiàn)方法都是在內(nèi)部類(lèi)Sync里面。

3.3 CountDownLatch

public class CountDownLatch {
    // 同步隊(duì)列
    private final Sync sync;
    // 構(gòu)造方法初始化計(jì)數(shù)值
    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }
    // 線程等待
    public void await() throws InterruptedException {
        // 調(diào)用AQS的acquireSharedInterruptibly方法
        // 即共享模式響應(yīng)中斷的獲取
        sync.acquireSharedInterruptibly(1);
    }
    // 計(jì)數(shù) - 1
    public void countDown() {
        sync.releaseShared(1);
    }
}

3.3.1 await() 方法解析

// CountDownLatch
public void await() throws InterruptedException {
        // 調(diào)用AQS的acquireSharedInterruptibly方法
        sync.acquireSharedInterruptibly(1);
    }
// 進(jìn)入AQS
public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        // 中斷判斷
        if (Thread.interrupted())
            throw new InterruptedException();
        // 如果沒(méi)有獲取到同步狀態(tài)，或者說(shuō)計(jì)數(shù)值不為0
        // 則調(diào)用doAcquireSharedInterruptibly方法，進(jìn)入同步隊(duì)列
        // 如果計(jì)數(shù)值為0則執(zhí)行后續(xù)業(yè)務(wù)邏輯
        if (tryAcquireShared(arg) < 0)
            // 該方法的解析參考文章結(jié)尾的鏈接，此處不再贅述
            doAcquireSharedInterruptibly(arg);
    }
// CountDownLatch 中tryAcquireShared的實(shí)現(xiàn)
protected int tryAcquireShared(int acquires) {
            // 當(dāng)計(jì)數(shù)為0時(shí)，線程才不會(huì)進(jìn)入同步隊(duì)列
            return (getState() == 0) ? 1 : -1;
        }

通過(guò)上面代碼可以知道，如果計(jì)數(shù)值為0，表示獲取成功。這就是CountDownLatch的機(jī)制，嘗試獲取latch的線程只有當(dāng)latch的值減到0的時(shí)候，才能獲取成功。

3.3.2 countDown() 方法解析

// CountDownLatch
public void countDown() {
        // 調(diào)用AQS的releaseShared方法
        sync.releaseShared(1);
    }
// 進(jìn)入AQS
public final boolean releaseShared(int arg) {
        // 共享模式釋放
        if (tryReleaseShared(arg)) {
            // 如果釋放成功則喚醒等待的線程，并返回true
            // 具體喚醒邏輯不再贅述，參考AQS解析文章
            doReleaseShared();
            return true;
        }
        return false;
    }
// CountDownLatch 中tryReleaseShared的實(shí)現(xiàn)
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;
                 // 通過(guò)自旋 + CAS方式改變剩余計(jì)數(shù)
                if (compareAndSetState(c, nextc))
                    // 如果計(jì)數(shù)為0返回true，否則返回false，結(jié)束自旋
                    // 返回true表示可以喚醒等待的線程
                    return nextc == 0;
            }
        }

3.3.3 CountDownLatch如何喚醒所有調(diào)用 await() 等待的線程呢？

當(dāng)調(diào)用doReleaseShared()喚醒后繼節(jié)點(diǎn)后，回到線程被掛起的地方，也就是doAcquireSharedInterruptibly(int arg)方法中

    private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        // 將當(dāng)前線程加入同步隊(duì)列的尾部
        final Node node = addWaiter(Node.SHARED);
        try {
            // 自旋
            for (;;) {
                // 獲取當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)
                final Node p = node.predecessor();
                // 如果前驅(qū)節(jié)點(diǎn)是頭結(jié)點(diǎn)，則嘗試獲取同步狀態(tài)
                if (p == head) {
                    // 當(dāng)前節(jié)點(diǎn)嘗試獲取同步狀態(tài)
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        // 如果獲取成功，則設(shè)置當(dāng)前節(jié)點(diǎn)為頭結(jié)點(diǎn)
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        return;
                    }
                }
                // 如果當(dāng)前節(jié)點(diǎn)的前驅(qū)不是頭結(jié)點(diǎn)，嘗試掛起當(dāng)前線程
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } catch (Throwable t) {
            cancelAcquire(node);
            throw t;
        }
    }

當(dāng)頭結(jié)點(diǎn)的后繼節(jié)點(diǎn)被喚醒后，線程將從掛起的地方醒來(lái)，繼續(xù)執(zhí)行，因?yàn)闆](méi)有return，所以進(jìn)入下一次循環(huán)。

此時(shí)，獲取同步狀態(tài)成功，執(zhí)行setHeadAndPropagate(node, r)。

    // 如果執(zhí)行這個(gè)函數(shù)，那么propagate一定等于1
    private void setHeadAndPropagate(Node node, int propagate) {
        // 獲取頭結(jié)點(diǎn)
        Node h = head;
        // 因?yàn)楫?dāng)前節(jié)點(diǎn)被喚醒，設(shè)置當(dāng)前節(jié)點(diǎn)為頭結(jié)點(diǎn)
        setHead(node);
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            // 獲取當(dāng)前節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)
            Node s = node.next;
            // 如果下一個(gè)節(jié)點(diǎn)為null或者節(jié)點(diǎn)為shared節(jié)點(diǎn)
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }
    private void doReleaseShared() {
    // 自旋
    for (;;) {
        Node h = head;
        // 如果隊(duì)列存在排隊(duì)的節(jié)點(diǎn)
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                // CAS設(shè)置不成功則不斷循環(huán)
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                // CAS操作成功后釋放后繼節(jié)點(diǎn)，并喚醒線程
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        // 隊(duì)列不存在排隊(duì)的節(jié)點(diǎn)，直接結(jié)束自旋
        if (h == head)                   // loop if head changed
            break;
    }

調(diào)用doReleaseShared方法喚醒后繼節(jié)點(diǎn)，后繼節(jié)點(diǎn)又回到線程被掛起的地方，也就是doAcquireSharedInterruptibly(int arg)方法中，實(shí)現(xiàn)循環(huán)喚醒所有await的線程。

此篇文章只解析了CountDownLatch的實(shí)現(xiàn)，它就是一個(gè)基于 AQS 的計(jì)數(shù)器，它內(nèi)部的方法都是圍繞 AQS 框架來(lái)實(shí)現(xiàn)的。

建議感興趣的同學(xué)先去了解AQS原理，只要明白了AQS的實(shí)現(xiàn)原理，再來(lái)看CountDownLatch、Semaphore、ReentrantLock等實(shí)現(xiàn)原理就一目了然了。

到此這篇關(guān)于Java中的CountDownLatch原理深入解析的文章就介紹到這了,更多相關(guān)CountDownLatch原理內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: