# Chapter 2-1. Thread Pool Analysis

## 前言

合理地利用thread pool可以帶來以下好處:

1. **降低資源的消耗**: 通過重複利用已經創造好的執行緒降低執行緒的創造/銷毀所造成的消耗
2. **提高回應速度**: 當task到達後, task不用等待執行緒創造就能立即執行
3. **提高執行緒的可管理性**: 執行緒是很珍貴的資源, 若無限制的創造, 不止消耗系統資源, 同時也會降低系統的穩定性, 使用thread pool可以進行統一的分配, 最佳化與監控. 但是要做到合理的利用thread pool, 必須先對其原理有一定程度的瞭解.

## Thread Pool的使用

* **Thread pool的建立**
  * 我們可以通過ThreadPoolExecutor來建立一個thread pool:

    ![](https://250052241-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-M523o32Rq-1H1Odaot0%2F-M523qKctpHr_YtEUjVW%2F-M523rrrqxRy38AqPSTg%2Fjmm-97.png?generation=1587042030877606\&alt=media)
  * 建立一個thread pool需要以下參數:
    * **corePoolSize**\
      Thread pool的基本大小, 當提交一個task到thread pool時, thread pool會建立一個執行緒來執行任務, 即使其它空閒的基本執行緒能夠執行新的task, thread pool還是會建立執行緒, 等到需要執行的task數量大於thread pool的基本大小(即corePoolSize)時就不會再建立了. 若呼叫了thread pool的prestartAllCoreThreads方法, thread pool會提前建立並啟動所有基本執行緒:

      ![](https://250052241-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-M523o32Rq-1H1Odaot0%2F-M523qKctpHr_YtEUjVW%2F-M523rrtm8wW5jdMiJTg%2Fjmm-98.png?generation=1587042030452174\&alt=media)
    * **maximumPoolSize**\
      Thread pool最大的size, 即其允許建立的最大執行緒數目. 若queue滿了, 並且已經建立的執行緒數目小於最大執行緒數目, 則thread pool會再建立新的執行緒來執行task. 要注意的是若使用了unbounded task queue, 這個參數就沒什麼效果了.
    * **keepAliveTime**\
      Thread pool的工作執行緒空閑後, 保持存活的時間. 若task很多, 並且每個task的執行時間都不長, 就可以調大這個時間, 提高執行緒的利用頻率.
    * **unit**\
      執行緒活動保持時間的單位, 可選的時間單位如下
      * DAYS
      * HOURS
      * MINUTES
      * MILLISECONDS
      * MICROSECONDS
      * NANOSECONDS
    * **workQueue**\
      用以保存等待執行的task之blocking queue, 大致上有以下幾種選擇:
      * **ArrayBlockingQueue**: 是一種基於陣列結構的bounded blocking queue(有界阻塞佇列), 此queue按FIFO原則對元素進行排序.
      * **LinkedBlockingQueue**: 是一種基於連結串列結構的optionally-bounded blocking queue(即可以自己指定界限大小的佇列), 此queue也是以FIFO排序元素, 吞吐量通常要高於ArrayBlockingQueue. 靜態工廠方法Executors.newFixedThreadPool()就是使用這個queue:

        ![](https://250052241-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-M523o32Rq-1H1Odaot0%2F-M523qKctpHr_YtEUjVW%2F-M523rrvOC6CFb3FzeYJ%2Fjmm-99.png?generation=1587042030237258\&alt=media)
      * **SynchronousQueue**: 一個不儲存元素的blocking queue, 每個插入操作必須等到另一個執行緒呼叫移除操作, 否則插入操作會一直處於blocking狀態, 吞吐量通常要高於LinkedBlockingQueue, 靜態工廠方法Executors.newCachedThreadPool就是用這個queue:

        ![](https://250052241-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-M523o32Rq-1H1Odaot0%2F-M523qKctpHr_YtEUjVW%2F-M523rrxk1cFu7QFmCOj%2Fjmm-100.png?generation=1587042030161211\&alt=media)
      * **PriorityBlockingQueue**: 一個具有優先級別的unbounded blocking queue(無界阻塞佇列).
    * **threadFactory**\
      用於設置建立執行緒的工廠, 可以通過執行緒工廠給每個建立出來的執行緒設定更有意義的名字
    * **handler**\
      RejectedExecutionHandler(**飽和策略**), 即**當queue跟thread pool都滿了, 表示thread pool處於飽和狀態, 那麼必須採取一種策略去處理提交的新task**. 這個策略預設是AbortPolicy, 表示無法處理新task時拋出exception. 下面是JDK5提供的四種策略:
      * **ThreadPoolExecutor.AbortPolicy**: 直接拋出runtime exception (**RejectedExecutionException**).
      * **ThreadPoolExecutor.CallerRunsPolicy**: 只有呼叫execute()的那個執行緒來執行task, 這其實提供了一種回饋控制機制(feedback control mechanism), 讓提交新task的速率得以減緩.
      * **ThreadPoolExecutor.DiscardPolicy**: 就不處理了, 直接丟掉(dropped).
      * **ThreadPoolExecutor.DiscardOldestPolicy**: 若executor當下沒有被shut down, 位於work queue的**head**之task就會被丟掉(因為它是最老的, oldest).
* **向thread pool提交task**

  我們可以使用execute提交task, 但是execute方法是沒有回傳值的, 所以無法判斷task是否被thread pool執行成功. 通過以下原始碼(java.util.concurrent.Executor, 所有executor service的根介面)可以知道execute方法輸入的task是一個Runnable的instance:

  ![](https://250052241-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-M523o32Rq-1H1Odaot0%2F-M523qKctpHr_YtEUjVW%2F-M523rrzFfiHxV4HCyjL%2Fjmm-101.png?generation=1587042030803722\&alt=media)\
  當然我們也可以使用submit方法來提交task, 其會回傳一個future, 然後我們就可以通過這個傳回來的future物件來判斷task有沒有執行成功, 通過future的get()來獲得回傳值, get()會block, 直到task完成為止, 而使用get(long timeout, TimeUnit unit)則可以指定block一定時間後就回傳, 這時候task就有可能沒有執行完, 以下只截取**java.util.concurrent.ExecutorService**的其中一個submit方法, 其實submit除了**Callable**, 也是可以接收**Runnable**的(自己翻, 我就不貼在這了):

  ![](https://250052241-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-M523o32Rq-1H1Odaot0%2F-M523qKctpHr_YtEUjVW%2F-M523rs04267kxB7_s_0%2Fjmm-102.png?generation=1587042031434483\&alt=media)
* **Thread pool的關閉**

  我們可以通過呼叫thread pool的**shutdown()/shutdownNow()**&#x4F86;關閉thread pool, 其原理是迭代thread pool中的worker thread, 然後逐一呼叫執行緒的interrupt()來中斷執行緒, 所以無法回應/中斷的task可能就永遠無法終止. 然而, 這兩種關閉的方式其實存在著一定的區別, **shutdownNow()**&#x9996;先將thread pool的狀態設成**STOP**, 然後嘗試停止所有正在執行或著暫停task的執行緒, 並**回傳等待執行task的list**, 而**shutdown()**&#x53EA;是將thread pool的狀態設成**SHUTDOWN**狀態, 然後中斷所有閒置的執行緒(呼叫interruptIdleWorkers()).

  只要呼叫了這兩個關閉方法的任一個, isShutdown()就會回傳true. 當所有的task都已經關閉後, 才意味著thread pool關閉成功, 這時呼叫isTerminated()會回傳true. 至於應該呼叫哪一種方法來關閉thread pool, 應該由提交到thread pool的task之特性來決定, 一般來說都會呼叫shutdown來關閉, 若task不一定要執行完, 則可以呼叫shutdownNow().

  ![](https://250052241-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-M523o32Rq-1H1Odaot0%2F-M523qKctpHr_YtEUjVW%2F-M523rs219w-NImDgvNk%2Fjmm-103.png?generation=1587042031298862\&alt=media)

## Thread Pool分析

### 流程分析

Thread pool的主要工作流程如下圖:\
![](https://250052241-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-M523o32Rq-1H1Odaot0%2F-M523qKctpHr_YtEUjVW%2F-M523rs4tyZUvH4yWGh-%2Fjmm-104.png?generation=1587042031072465\&alt=media)

從上圖可以看出, 在提交一個新的task至thread pool時, thread pool的處理流程如下:

1. 首先, thread pool判斷基本的thread pool是否已滿, 若沒有則建立一個worker thread來執行任務, 反之則進入下一道流程.
2. 其次, thread pool判斷work queue是否已滿, 若沒有就將新提交的task儲存在work queue裡, 滿了則進入下一道流程.
3. 最後thread pool會判斷整個pool是否滿了, 若沒滿就建立一個新的worker thread來執行task, 反之交給飽和策略來處理此task.

### 原始碼分析

以上的流程分析已經很直觀的闡述了thread pool的工作原理, 再來稍微看一下原始碼是怎麼實作的:

![](https://250052241-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-M523o32Rq-1H1Odaot0%2F-M523qKctpHr_YtEUjVW%2F-M523rs6kfA8iDmjW-Ud%2Fjmm-105.png?generation=1587042030270047\&alt=media)\
這部分是擷取自java.util.concurrent.ThreadPoolExecutor.java

### Worker Thread

Thread pool在建立執行緒時, 會將執行緒封裝成worker thread (Worker), Worker在執行完task後, 還會無限迭代獲取workQueue裡的task來執行. 我們可以從Worker的run方法看到這點:

![](https://250052241-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-M523o32Rq-1H1Odaot0%2F-M523qKctpHr_YtEUjVW%2F-M523rs8_NjNdlYz8Zs6%2Fjmm-106.png?generation=1587042030525818\&alt=media)\
這部分是擷取自java.util.concurrent.ThreadPoolExecutor.Worker.

## 合理的組態Thread Pool

要想合理的組態你的thread pool, 可以先從task的特性分析著手, 這邊有幾個常見的切入點可以作為分析的依據:

#### Task的性質: CPU密集型/IO密集型或著是混合型的task

* 這種性質的task可以用不同規模的thread pool分開處理.
  * **CPU密集型**: 盡可能配少量一點的執行緒, 譬如配置**cpu數量+1**個執行緒的thread pool.
  * **IO密集型**: 由於這種task的執行緒並不是一直在執行, 則可以盡可能配制多一點執行緒, 如**cpu數量的兩倍**.
  * **混合型**: 若可以拆分, 則將其拆分成一個CPU密集型task和一個IO密集型task, 只要這兩個task執行的時間相差不是太大, 那麼分解後執行的吞吐量應該可以高於串連執行的吞吐量; 倘若執行時間相差過大, 就沒必要分解了. 我們可以通過Runtime.getRuntime().availableProcessors()方法得到當前設備上的CPU數量.

#### Task的優先程度: 由高到低

* 這種類型可以使用**PriorityBlockingQueue**來處理: 其可以讓優先度高的task先執行, 但要注意的是若一直都有高優先度的task被提交到queue裡, 那麼低優先度的task可能永遠都執行不到.

#### Task的執行時間: 由長到短

* 這種有時間差異的task可以交給不同規模的thread pool來處理, 或是使用PriorityBlockingQueue也可以, 讓執行時間短的任務先執行.

#### Task的相依性: 是否相依於其它系統資源, 如資料庫

* 相依於DB connection pool的task, 因為執行緒提交SQL後還要等DB回傳結果, 若等待的時間越長, CPU的空閑時間就越長, 那麼執行緒數量應該要設置得越大, 這樣才能更好地利用CPU.

除了上述的手段, 還有一個建議就是: **使用bounded queue, 因為這種queue能增加系統的穩定性與預警能力**, 可以根據需求調大一點, 譬如幾千. 試想以下情境: Thread pool跟workQueue都滿了, 不斷地拋出拋棄task的exception(RejectedExecutionException). 這時若發現是DB出了問題, 導致執行SQL變得異常緩慢, 因為thread pool裡的task全都是需要向DB進行query或是insert/update資料的, 所以導致thread pool裡的worker都塞住了, 因而讓task都積在thread pool中. **若這時候的workQueue是unbounded queue, thread pool的workQueue就會越長越大, 進而可能撐爆memory**, 導致系統不能用, 而不再只是task出問題而已了. 當然, 儘管系統所有的task是用單獨的server去做部署的, 且針對不同類型的task使用不同規模的thread pool, 但出現了這種問題時, 還是有可能影響到其它有相依性的task.

## Thread Pool的監控

在thread pool裡, 有一些方法在監控的時候可以參考:

* getTaskCount: Thread pool所需要執行的task之近似值數量(have ever been scheduled), 注意, 這是近似值, 因為task的狀態在計算過程中可能會改變.
* getCompletedTaskCount: Thread pool在運作過程中已經完成的task數量, 小於等於taskCount, 這也是個近似值.
* getLargestPoolSize: Thread pool曾經建立過的最大執行緒數量, 通過這個資料可以知道thread pool是否滿過. 若等於thread pool的最大size, 表示其曾經滿過.
* getPoolSize: Thread pool當前的執行緒數量. 若thread pool不銷毀的話, pool裡的執行緒也不會自動銷毀, 所以這個大小只增不減.
* getActiveCount: 獲得正在活動中(actively executing)的執行緒數量, 這也是個近似值.

通過繼承thread pool並且覆寫其中的beforeExecute/afterExecute以及terminated方法, 我們就可以在task執行的前/後與thread pool關閉前做點什麼事情.譬如監控task的平均執行時間, 最長執行時間與最短執行時間等等. 這幾個方法在thread pool裡是空方法:

![](https://250052241-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-M523o32Rq-1H1Odaot0%2F-M523qKctpHr_YtEUjVW%2F-M523rsAkCFkWC9kbzsP%2Fjmm-107.png?generation=1587042030566311\&alt=media)

![](https://250052241-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-M523o32Rq-1H1Odaot0%2F-M523qKctpHr_YtEUjVW%2F-M523rsC7x5fS0B6-9CD%2Fjmm-108.png?generation=1587042031020547\&alt=media)

![](https://250052241-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-M523o32Rq-1H1Odaot0%2F-M523qKctpHr_YtEUjVW%2F-M523rsEfaKDLTSd3s3T%2Fjmm-109.png?generation=1587042030836274\&alt=media)

## 參考資料

* [InfoQ: Java執行緒池的分析和使用](http://www.infoq.com/cn/articles/java-threadPool?utm_source=infoq\&utm_campaign=user_page\&utm_medium=link)
* JDK9 source code (build 9+181)
