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

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

線程池基本調(diào)度功能。

線程池自動擴(kuò)容縮容。

隊(duì)列緩存線程。

關(guān)閉線程池。

這些功能，最后也留下了三個待實(shí)現(xiàn)的 features 。

執(zhí)行帶有返回值的線程。

異常處理怎么辦？

所有任務(wù)執(zhí)行完怎么通知我？

這次就實(shí)現(xiàn)這三個特性來看看 j.u.c 中的線程池是如何實(shí)現(xiàn)這些需求的。

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

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

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

以我們之前的代碼為例：

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

執(zhí)行結(jié)果如下：

為了簡單的達(dá)到這個效果，我們可以在初始化線程池的時候傳入一個接口的實(shí)現(xiàn)，這個接口就是用于任務(wù)完成之后的回調(diào)。

public interface Notify {

    /**
     * 回調(diào)
     */
    void notifyListen() ;
}

以上就是線程池的構(gòu)造函數(shù)以及接口的定義。

所以想要實(shí)現(xiàn)這個功能的關(guān)鍵是在何時回調(diào)這個接口？

仔細(xì)想想其實(shí)也簡單：只要我們記錄提交到線程池中的任務(wù)及完成的數(shù)量，他們兩者的差為 0 時就認(rèn)為線程池中的任務(wù)已執(zhí)行完畢；這時便可回調(diào)這個接口。

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

為了并發(fā)安全的考慮，這里的計數(shù)器采用了原子的 AtomicInteger 。

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

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

我們使用 ThreadPoolExecutor 的常規(guī)關(guān)閉流程如下：

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

線程提交完畢后執(zhí)行 shutdown() 關(guān)閉線程池，接著循環(huán)調(diào)用 awaitTermination() 方法，一旦任務(wù)全部執(zhí)行完畢后則會返回 true 從而退出循環(huán)。

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

執(zhí)行 shutdown() 后會將線程池的狀態(tài)置為關(guān)閉狀態(tài)，這時將會停止接收新的任務(wù)同時會等待隊(duì)列中的任務(wù)全部執(zhí)行完畢后才真正關(guān)閉線程池。

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

為什么要兩個 api 結(jié)合一起使用呢？

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

shutdown() 執(zhí)行后并不會阻塞，會立即返回，所有才需要后續(xù)用循環(huán)不停的調(diào)用 awaitTermination()，因?yàn)檫@個 api 才會阻塞線程。

其實(shí)我們查看源碼會發(fā)現(xiàn)，ThreadPoolExecutor 中的阻塞依然也是等待通知機(jī)制的運(yùn)用，只不過用的是 LockSupport 的 API 而已。

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

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

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

這個接口有一個返回值。

同時在提交任務(wù)時也稍作改動：

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

最后通過第二步將所有執(zhí)行結(jié)果打印出來：

再看具體實(shí)現(xiàn)之前先來思考下這樣的功能如何實(shí)現(xiàn)？

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

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

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

public interface Callable {

    /**
     * 執(zhí)行任務(wù)
     * @return 執(zhí)行結(jié)果
     */
    T call() ;
}

它的 call 函數(shù)就是剛才提到的有返回值的方法，所以我們應(yīng)當(dāng)在線程的 run() 函數(shù)中去調(diào)用它。

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

既然有了接口那自然就得有它的實(shí)現(xiàn) FutureTask，它實(shí)現(xiàn)了 Future 接口用于后續(xù)獲取返回值。

同時實(shí)現(xiàn)了 Runnable 接口會把自己變?yōu)橐粋€線程。

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

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

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

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

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

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

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

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

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

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

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

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

比如類似于這樣的場景：

創(chuàng)建了只有一個線程的線程池，這個線程只做一件事，就是一直不停的 while 循環(huán)。

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

是線程繼續(xù)運(yùn)行？還是線程池會退出？

通過現(xiàn)象來看其實(shí)哪種都不是，線程既沒有繼續(xù)運(yùn)行同時線程池也沒有退出，會一直卡在這里。

當(dāng)我們 dump 線程快照會發(fā)現(xiàn)：

這時線程池中還有一個線程在運(yùn)行，通過線程名稱會發(fā)現(xiàn)這是新創(chuàng)建的一個線程（之前是Thread-0，現(xiàn)在是 Thread-1）。

它的線程狀態(tài)為 WAITING ，通過堆棧發(fā)現(xiàn)是卡在了 CustomThreadPool.java:272 處。

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

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

沒錯，我之前寫過兩篇：

一個線程罷工的詭異事件

線程池中你不容錯過的一些細(xì)節(jié)

線程池相關(guān)的問題，當(dāng)時的討論也非常“激烈”，其實(shí)最終的原因和這里是一模一樣的。

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

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

其實(shí)在線程池內(nèi)部會對線程的運(yùn)行捕獲異常，但它并不會處理，只是用于標(biāo)記是否執(zhí)行成功；

一旦執(zhí)行失敗則會回收掉當(dāng)前異常的線程，然后重新創(chuàng)建一個新的 Worker 線程繼續(xù)從隊(duì)列里取任務(wù)然后執(zhí)行。

所以最終才會卡在從隊(duì)列中取任務(wù)處。

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

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

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

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

利用等待通知機(jī)制來實(shí)現(xiàn)線程間通信（線程執(zhí)行結(jié)果、等待線程池執(zhí)行完畢等）。

最后也學(xué)會了：

標(biāo)準(zhǔn)的線程池關(guān)閉流程。

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

線程異常捕獲的重要性。

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

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

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