摘要：前言前段時間寫過一篇線程池沒你想的那么簡單，和大家一起擼了一個基本的線程池，具備線程池基本調度功能。線程池自動擴容縮容。回調以上就是線程池的構造函數以及接口的定義。所以我們在使用線程池時，其中的任務一定要做好異常處理。線程異常捕獲的重要性。

前段時間寫過一篇《線程池沒你想的那么簡單》，和大家一起擼了一個基本的線程池，具備：

線程池基本調度功能。

線程池自動擴容縮容。

隊列緩存線程。

關閉線程池。

這些功能，最后也留下了三個待實現的 features 。

執行帶有返回值的線程。

異常處理怎么辦？

所有任務執行完怎么通知我？

這次就實現這三個特性來看看 j.u.c 中的線程池是如何實現這些需求的。

再看本文之前，強烈建議先查看上文《線程池沒你想的那么簡單》

大家在用線程池的時候或多或少都會有這樣的需求：

線程池中的任務執行完畢后再通知主線程做其他事情，比如一批任務都執行完畢后再執行下一波任務等等。

以我們之前的代碼為例：

總共往線程池中提交了 13 個任務，直到他們都執行完畢后再打印 “任務執行完畢” 這個日志。

執行結果如下：

為了簡單的達到這個效果，我們可以在初始化線程池的時候傳入一個接口的實現，這個接口就是用于任務完成之后的回調。

public interface Notify {

    /**
     * 回調
     */
    void notifyListen() ;
}

以上就是線程池的構造函數以及接口的定義。

所以想要實現這個功能的關鍵是在何時回調這個接口？

仔細想想其實也簡單：只要我們記錄提交到線程池中的任務及完成的數量，他們兩者的差為 0 時就認為線程池中的任務已執行完畢；這時便可回調這個接口。

所以在往線程池中寫入任務時我們需要記錄任務數量：

為了并發安全的考慮，這里的計數器采用了原子的 AtomicInteger 。

而在任務執行完畢后就將計數器 -1 ，一旦為 0 時則任務任務全部執行完畢；這時便可回調我們自定義的接口完成通知。

這樣的需求在 jdk 中的 ThreadPoolExecutor 中也有相關的 API ，只是用法不太一樣，但本質原理都大同小異。

我們使用 ThreadPoolExecutor 的常規關閉流程如下：

    executorService.shutdown();
    while (!executorService.awaitTermination(100, TimeUnit.MILLISECONDS)) {
        logger.info("thread running");
    }

線程提交完畢后執行 shutdown() 關閉線程池，接著循環調用 awaitTermination() 方法，一旦任務全部執行完畢后則會返回 true 從而退出循環。

這兩個方法的目的和原理如下：

執行 shutdown() 后會將線程池的狀態置為關閉狀態，這時將會停止接收新的任務同時會等待隊列中的任務全部執行完畢后才真正關閉線程池。

awaitTermination 會阻塞直到線程池所有任務執行完畢或者超時時間已到。

為什么要兩個 api 結合一起使用呢？

主要還在最終的目的是：所有線程執行完畢后再做某件事情，也就是在線程執行完畢之前其實主線程是需要被阻塞的。

shutdown() 執行后并不會阻塞，會立即返回，所有才需要后續用循環不停的調用 awaitTermination()，因為這個 api 才會阻塞線程。

其實我們查看源碼會發現，ThreadPoolExecutor 中的阻塞依然也是等待通知機制的運用，只不過用的是 LockSupport 的 API 而已。

接下來是帶有返回值的線程，這個需求也非常常見；比如需要線程異步計算某些數據然后得到結果最終匯總使用。

先來看看如何使用（和 jdk 的類似）：

首先任務是不能實現 Runnable 接口了，畢竟他的 run() 函數是沒有返回值的；所以我們改實現一個 Callable 的接口：

這個接口有一個返回值。

同時在提交任務時也稍作改動：

首先是執行任務的函數由 execute() 換為了 submit()，同時他會返回一個返回值 Future，通過它便可拿到線程執行的結果。

最后通過第二步將所有執行結果打印出來：

再看具體實現之前先來思考下這樣的功能如何實現？

首先受限于 jdk 的線程 api 的規范，要執行一個線程不管是實現接口還是繼承類，最終都是執行的 run() 函數。

所以我們想要一個線程有返回值無非只能是在執行 run() 函數時去調用一個有返回值的方法，再將這個返回值存放起來用于后續使用。

比如我們這里新建了一個 Callable 的接口：

public interface Callable {

    /**
     * 執行任務
     * @return 執行結果
     */
    T call() ;
}

它的 call 函數就是剛才提到的有返回值的方法，所以我們應當在線程的 run() 函數中去調用它。

接著還會有一個 Future 的接口，他的主要作用是獲取線程的返回值，也就是 再將這個返回值存放起來用于后續使用 這里提到的后續使用。

既然有了接口那自然就得有它的實現 FutureTask，它實現了 Future 接口用于后續獲取返回值。

同時實現了 Runnable 接口會把自己變為一個線程。

所以在它的 run() 函數中會調用剛才提到的具有返回值的 call() 函數。

再次結合 submit() 提交任務和 get() 獲取返回值的源碼來看會更加理解這其中的門道。

    /**
     * 有返回值
     *
     * @param callable
     * @param 
     * @return
     */
    public  Future submit(Callable callable) {
        FutureTask future = new FutureTask(callable);
        execute(future);
        return future;
    }

submit() 非常簡單，將我們丟進來的 Callable 對象轉換為一個 FutureTask 對象，然后再調用之前的 execute() 來丟進線程池（后續的流程就和一個普通的線程進入線程池的流程一樣）。

FutureTask 本身也是線程，所以可以直接使用 execute() 函數。

而 future.get() 函數中 future 對象由于在 submit() 中返回的真正對象是 FutureTask，所以我們直接看其中的源碼就好。

由于 get() 在線程沒有返回之前是一個阻塞函數，最終也是通過 notify.wait() 使線程進入阻塞狀態來實現的。

而使其從 wait() 中返回的條件必然是在線程執行完畢拿到返回值的時候才進行喚醒。

也就是圖中的第二部分；一旦線程執行完畢（callable.call()）就會喚醒 notify 對象，這樣 get 方法也就能返回了。

同樣的道理，ThreadPoolExecutor 中的原理也是類似，只不過它考慮的細節更多所以看起來很復雜，但精簡代碼后核心也就是這些。

甚至最終使用的 api 看起來都是類似的：

最后一個是一些新手使用線程池很容易踩坑的一個地方：那就是異常處理。

比如類似于這樣的場景：

創建了只有一個線程的線程池，這個線程只做一件事，就是一直不停的 while 循環。

但是循環的過程中不小心拋出了一個異常，巧的是這個異常又沒有被捕獲。你覺得后續會發生什么事情呢？

是線程繼續運行？還是線程池會退出？

通過現象來看其實哪種都不是，線程既沒有繼續運行同時線程池也沒有退出，會一直卡在這里。

當我們 dump 線程快照會發現：

這時線程池中還有一個線程在運行，通過線程名稱會發現這是新創建的一個線程（之前是Thread-0，現在是 Thread-1）。

它的線程狀態為 WAITING ，通過堆棧發現是卡在了 CustomThreadPool.java:272 處。

就是卡在了從隊列里獲取任務的地方，由于此時的任務隊列是空的，所以他會一直阻塞在這里。

看到這里，之前關注的朋友有沒有似曾相識的感覺。

沒錯，我之前寫過兩篇：

一個線程罷工的詭異事件

線程池中你不容錯過的一些細節

線程池相關的問題，當時的討論也非常“激烈”，其實最終的原因和這里是一模一樣的。

所以就這次簡版的代碼來看看其中的問題：

現在又簡化了一版代碼我覺得之前還有疑問的朋友這次應該會更加明白。

其實在線程池內部會對線程的運行捕獲異常，但它并不會處理，只是用于標記是否執行成功；

一旦執行失敗則會回收掉當前異常的線程，然后重新創建一個新的 Worker 線程繼續從隊列里取任務然后執行。

所以最終才會卡在從隊列中取任務處。

其實 ThreadPoolExecutor 的異常處理也是類似的，具體的源碼就不多分析了，在上面兩篇文章中已經說過幾次。

所以我們在使用線程池時，其中的任務一定要做好異常處理。

這一波下來我覺得線程池搞清楚沒啥問題了，總的來看它內部運用了非常多的多線程解決方案，比如：

ReentrantLock 重入鎖來保證線程寫入的并發安全。

利用等待通知機制來實現線程間通信（線程執行結果、等待線程池執行完畢等）。

最后也學會了：

標準的線程池關閉流程。

如何使用有返回值的線程。

線程異常捕獲的重要性。

最后本文所有源碼（結合其中的測試代碼使用）：

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

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