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

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

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

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

其實這類問題的核心論點都是：如何在一個線程中得知其他線程是否執(zhí)行完畢。

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

定義一個計數(shù)器為 3。

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

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

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

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

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

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

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

    /**
     * 線程完成后計數(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() 來保證多線程的原子性，當(dāng)減為 0 時則利用等待通知機制來 notify 其他線程。

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

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

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

        }
    }

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

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

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

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

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

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

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

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

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

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

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

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

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

通過構(gòu)造函數(shù)會發(fā)現(xiàn)有一個 內(nèi)部類 Sync，他是繼承于 AbstractQueuedSynchronizer ；這是 Java 并發(fā)包中的基礎(chǔ)框架，都可以多帶帶拿來講了，所以這次重點不是它，今后我們再著重介紹。

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

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

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

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

看到這里有沒有似曾相識的感覺。

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

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

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

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

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

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

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

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

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

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

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

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

等待線程一旦計數(shù)器 > 0 時則會利用 LockSupport.park 來等待喚醒。

這樣整個流程也就串起來了，它的使用方法也和上文的類似。

就不做過多介紹了。

同樣的來看一個實際案例。

在上一篇《一次分表踩坑實踐的探討》提到了對于全表掃描的情況下，需要利用多線程來提高查詢效率。

比如我們這里分為了 64 張表，計劃利用 8 個線程來分別處理這些表的數(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("查詢完畢");

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

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

文中涉及到的源碼：

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

你的點贊與分享是對我最大的支持