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

前段時(shí)間寫過一篇《線程池沒你想的那么簡(jiǎn)單》，和大家一起擼了一個(gè)基本的線程池，具備：

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

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

隊(duì)列緩存線程。

關(guān)閉線程池。

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

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

異常處理怎么辦？

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

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

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

大家在用線程池的時(shí)候或多或少都會(huì)有這樣的需求：

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

以我們之前的代碼為例：

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

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

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

public interface Notify {

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

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

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

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

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

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

而在任務(wù)執(zhí)行完畢后就將計(jì)數(shù)器 -1 ，一旦為 0 時(shí)則任務(wù)任務(wù)全部執(zhí)行完畢；這時(shí)便可回調(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í)行完畢后則會(huì)返回 true 從而退出循環(huán)。

這兩個(gè)方法的目的和原理如下：

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

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

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

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

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

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

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

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

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

這個(gè)接口有一個(gè)返回值。

同時(shí)在提交任務(wù)時(shí)也稍作改動(dòng)：

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

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

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

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

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

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

public interface Callable {

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

比如類似于這樣的場(chǎng)景：

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

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

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

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

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

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

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

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

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

沒錯(cuò)，我之前寫過兩篇：

一個(gè)線程罷工的詭異事件

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

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

所以就這次簡(jiǎn)版的代碼來看看其中的問題：

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

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

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

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

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

所以我們?cè)谑褂镁€程池時(shí)，其中的任務(wù)一定要做好異常處理。

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

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

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

最后也學(xué)會(huì)了：

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

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

線程異常捕獲的重要性。

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

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

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