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

在面試過程中聊到并發相關的內容時，不少面試官都喜歡問這類問題：

當 N 個線程同時完成某項任務時，如何知道他們都已經執行完畢了。

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

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

假設現在有 3 個線程在運行，需要在主線程中得知他們的運行結果；可以分為以下幾步：

定義一個計數器為 3。

每個線程完成任務后計數減一。

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

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

按照這個思路自定義了一個 MultipleThreadCountDownKit 工具，構造函數如下：

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

所以在初始化時需要根據線程數量來構建對象。

當其中一個業務線程完成后需要將這個計數器減一，直到減為0為止。

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

而需要知道業務線程執行完畢的其他線程則需要在未完成之前一直處于等待狀態，直到上文提到的在計數器變為 0 時得到通知。

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

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

        }
    }

原理也很簡單，一旦計數器還存在時則會利用 notify 對象進行等待，直到被業務線程喚醒。

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

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

初始化了三個計數器的并發工具 MultipleThreadCountDownKit

創建了三個線程分別執行業務邏輯，完畢后執行 countDown()。

線程 3 休眠了 2s 用于模擬業務耗時。

主線程執行 await() 等待他們三個線程執行完畢。

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

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

也可以在初始化的時候指定一個回調接口，用于接收業務線程執行完畢后的通知。

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

當然我們自己實現的代碼沒有經過大量生產環境的驗證，所以主要的目的還是嘗試窺探官方的實現原理。

所以我們現在來看看 juc 下的 CountDownLatch 是如何實現的。

通過構造函數會發現有一個 內部類 Sync，他是繼承于 AbstractQueuedSynchronizer ；這是 Java 并發包中的基礎框架，都可以多帶帶拿來講了，所以這次重點不是它，今后我們再著重介紹。

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

其實他的核心邏輯和我們自己實現的區別不大。

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

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

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

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

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

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

所以其實本質上還是相同的。

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

    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);
    }

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

一旦還存在未完成的線程時，則會調用 doAcquireSharedInterruptibly 進入阻塞狀態。

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

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

所有的業務線程執行完畢后會在計數器減為 0 時調用 LockSupport.unpark 來喚醒線程。

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

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

就不做過多介紹了。

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

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

比如我們這里分為了 64 張表，計劃利用 8 個線程來分別處理這些表的數據，偽代碼如下：

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("查詢完畢");

這樣就可以實現所有數據都查詢完畢后再做統一匯總；代碼挺簡單，也好理解（當然也可以使用線程池的 API）。

CountDownLatch 算是 juc 中一個高頻使用的工具，學會和理解他的使用會幫助我們更容易編寫并發應用。

文中涉及到的源碼：

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

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