摘要：所以也很容易想到可以利用等待通知機(jī)制來(lái)實(shí)現(xiàn)，和上文的并發(fā)包入坑指北之阻塞隊(duì)列的類(lèi)似。

在面試過(guò)程中聊到并發(fā)相關(guān)的內(nèi)容時(shí)，不少面試官都喜歡問(wèn)這類(lèi)問(wèn)題：

當(dāng) N 個(gè)線程同時(shí)完成某項(xiàng)任務(wù)時(shí)，如何知道他們都已經(jīng)執(zhí)行完畢了。

這也是本次討論的話題之一，所以本篇為『并發(fā)包入坑指北』的第二篇；來(lái)聊聊常見(jiàn)的并發(fā)工具。

其實(shí)這類(lèi)問(wèn)題的核心論點(diǎn)都是：如何在一個(gè)線程中得知其他線程是否執(zhí)行完畢。

假設(shè)現(xiàn)在有 3 個(gè)線程在運(yùn)行，需要在主線程中得知他們的運(yùn)行結(jié)果；可以分為以下幾步：

定義一個(gè)計(jì)數(shù)器為 3。

每個(gè)線程完成任務(wù)后計(jì)數(shù)減一。

一旦計(jì)數(shù)器減為 0 則通知等待的線程。

所以也很容易想到可以利用等待通知機(jī)制來(lái)實(shí)現(xiàn)，和上文的『并發(fā)包入坑指北』之阻塞隊(duì)列的類(lèi)似。

按照這個(gè)思路自定義了一個(gè) MultipleThreadCountDownKit 工具，構(gòu)造函數(shù)如下：

考慮到并發(fā)的前提，這個(gè)計(jì)數(shù)器自然需要保證線程安全，所以采用了 AtomicInteger。

所以在初始化時(shí)需要根據(jù)線程數(shù)量來(lái)構(gòu)建對(duì)象。

當(dāng)其中一個(gè)業(yè)務(wù)線程完成后需要將這個(gè)計(jì)數(shù)器減一，直到減為0為止。

    /**
     * 線程完成后計(jì)數(shù) -1
     */
    public void countDown(){

        if (counter.get() <= 0){
            return;
        }

        int count = this.counter.decrementAndGet();
        if (count < 0){
            throw new RuntimeException("concurrent error") ;
        }

        if (count == 0){
            synchronized (notify){
                notify.notify();
            }
        }

    }

利用 counter.decrementAndGet() 來(lái)保證多線程的原子性，當(dāng)減為 0 時(shí)則利用等待通知機(jī)制來(lái) notify 其他線程。

而需要知道業(yè)務(wù)線程執(zhí)行完畢的其他線程則需要在未完成之前一直處于等待狀態(tài)，直到上文提到的在計(jì)數(shù)器變?yōu)?0 時(shí)得到通知。

    /**
     * 等待所有的線程完成
     * @throws InterruptedException
     */
    public void await() throws InterruptedException {
        synchronized (notify){
            while (counter.get() > 0){
                notify.wait();
            }

            if (notifyListen != null){
                notifyListen.notifyListen();
            }

        }
    }

原理也很簡(jiǎn)單，一旦計(jì)數(shù)器還存在時(shí)則會(huì)利用 notify 對(duì)象進(jìn)行等待，直到被業(yè)務(wù)線程喚醒。

同時(shí)這里新增了一個(gè)通知接口可以自定義實(shí)現(xiàn)喚醒后的一些業(yè)務(wù)邏輯，后文會(huì)做演示。

主要就是這兩個(gè)函數(shù)，下面來(lái)做一個(gè)演示。

初始化了三個(gè)計(jì)數(shù)器的并發(fā)工具 MultipleThreadCountDownKit

創(chuàng)建了三個(gè)線程分別執(zhí)行業(yè)務(wù)邏輯，完畢后執(zhí)行 countDown()。

線程 3 休眠了 2s 用于模擬業(yè)務(wù)耗時(shí)。

主線程執(zhí)行 await() 等待他們?nèi)齻€(gè)線程執(zhí)行完畢。

通過(guò)執(zhí)行結(jié)果可以看出主線程會(huì)等待最后一個(gè)線程完成后才會(huì)退出；從而達(dá)到了主線程等待其余線程的效果。

    MultipleThreadCountDownKit multipleThreadKit = new MultipleThreadCountDownKit(3);
    multipleThreadKit.setNotify(() -> LOGGER.info("三個(gè)線程完成了任務(wù)"));

也可以在初始化的時(shí)候指定一個(gè)回調(diào)接口，用于接收業(yè)務(wù)線程執(zhí)行完畢后的通知。

當(dāng)然和在主線程中執(zhí)行這段邏輯效果是一樣的（和執(zhí)行 await() 方法處于同一個(gè)線程）。

當(dāng)然我們自己實(shí)現(xiàn)的代碼沒(méi)有經(jīng)過(guò)大量生產(chǎn)環(huán)境的驗(yàn)證，所以主要的目的還是嘗試窺探官方的實(shí)現(xiàn)原理。

所以我們現(xiàn)在來(lái)看看 juc 下的 CountDownLatch 是如何實(shí)現(xiàn)的。

通過(guò)構(gòu)造函數(shù)會(huì)發(fā)現(xiàn)有一個(gè) 內(nèi)部類(lèi) Sync，他是繼承于 AbstractQueuedSynchronizer ；這是 Java 并發(fā)包中的基礎(chǔ)框架，都可以多帶帶拿來(lái)講了，所以這次重點(diǎn)不是它，今后我們?cè)僦亟榻B。

這里就可以把他簡(jiǎn)單理解為提供了和上文類(lèi)似的一個(gè)計(jì)數(shù)器及線程通知工具就行了。

其實(shí)他的核心邏輯和我們自己實(shí)現(xiàn)的區(qū)別不大。

    public void countDown() {
        sync.releaseShared(1);
    }
    
    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

利用這個(gè)內(nèi)部類(lèi)的 releaseShared 方法，我們可以理解為他想要將計(jì)數(shù)器減一。

看到這里有沒(méi)有似曾相識(shí)的感覺(jué)。

沒(méi)錯(cuò)，在 JDK1.7 中的 AtomicInteger 自減就是這樣實(shí)現(xiàn)的（利用 CAS 保證了線程安全）。

只是一旦計(jì)數(shù)器減為 0 時(shí)則會(huì)執(zhí)行 doReleaseShared 喚醒其他的線程。

這里我們只需要關(guān)心紅框部分（其他的暫時(shí)不用關(guān)心，這里涉及到了 AQS 中的隊(duì)列相關(guān)），最終會(huì)調(diào)用 LockSupport.unpark 來(lái)喚醒線程；就相當(dāng)于上文調(diào)用 object.notify()。

所以其實(shí)本質(zhì)上還是相同的。

其中的 await() 也是借用 Sync 對(duì)象的方法實(shí)現(xiàn)的。

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

    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        //判斷計(jì)數(shù)器是否還未完成    
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

    protected int tryAcquireShared(int acquires) {
        return (getState() == 0) ? 1 : -1;
    }

一旦還存在未完成的線程時(shí)，則會(huì)調(diào)用 doAcquireSharedInterruptibly 進(jìn)入阻塞狀態(tài)。

    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

同樣的由于這也是 AQS 中的方法，我們只需要關(guān)心紅框部分；其實(shí)最終就是調(diào)用了 LockSupport.park 方法，也就相當(dāng)于執(zhí)行了 object.wait() 。

所有的業(yè)務(wù)線程執(zhí)行完畢后會(huì)在計(jì)數(shù)器減為 0 時(shí)調(diào)用 LockSupport.unpark 來(lái)喚醒線程。

等待線程一旦計(jì)數(shù)器 > 0 時(shí)則會(huì)利用 LockSupport.park 來(lái)等待喚醒。

這樣整個(gè)流程也就串起來(lái)了，它的使用方法也和上文的類(lèi)似。

就不做過(guò)多介紹了。

同樣的來(lái)看一個(gè)實(shí)際案例。

在上一篇《一次分表踩坑實(shí)踐的探討》提到了對(duì)于全表掃描的情況下，需要利用多線程來(lái)提高查詢(xún)效率。

比如我們這里分為了 64 張表，計(jì)劃利用 8 個(gè)線程來(lái)分別處理這些表的數(shù)據(jù)，偽代碼如下：

CountDownLatch count = new CountDownLatch(64);
ConcurrentHashMap total = new ConcurrentHashMap();
for(Integer i=0;i<=63;i++){
    executor.execute(new Runnable(){
        @Override
        public void run(){
            List value = queryTable(i);
            total.put(value,NULL);
            count.countDown();
        }
    }) ;
    
}

count.await();
System.out.println("查詢(xún)完畢");

這樣就可以實(shí)現(xiàn)所有數(shù)據(jù)都查詢(xún)完畢后再做統(tǒng)一匯總；代碼挺簡(jiǎn)單，也好理解（當(dāng)然也可以使用線程池的 API）。

CountDownLatch 算是 juc 中一個(gè)高頻使用的工具，學(xué)會(huì)和理解他的使用會(huì)幫助我們更容易編寫(xiě)并發(fā)應(yīng)用。

文中涉及到的源碼：

https://github.com/crossoverJie/JCSprout/blob/master/src/main/java/com/crossoverjie/concurrent/communication/MultipleThreadCountDownKit.java

你的點(diǎn)贊與分享是對(duì)我最大的支持