深入淺出Java執行緒池：原始碼篇
阿新 • • 發佈：2021-02-06
## 前言
在上一篇文章[深入淺出Java執行緒池：理論篇](https://juejin.cn/post/6923856791117758472)中，已經介紹了什麼是執行緒池以及基本的使用。（本來寫作的思路是使用篇，但經網友建議後，感覺改為理論篇會更加合適）。本文則深入執行緒池的原始碼，主要是介紹ThreadPoolExecutor內部的原始碼是如何實現的，對ThreadPoolExecutor有一個更加清晰的認識。
ThreadPoolExecutor的原始碼相對而言比較好理解，沒有特別難以讀懂的地方。相信沒有閱讀原始碼習慣的讀者，跟著本文，也可以很輕鬆地讀懂ThreadPoolExecutor的核心原始碼邏輯。
本文原始碼jdk版本為8，該類版本為jdk1.5，也就是在1.5之後，ThreadPoolExecutor的原始碼沒有做修改。
## 執行緒池家族
Java中的執行緒池繼承結構如下圖：(類圖中只寫了部分方法且省略引數)
- 頂層介面Executor表示一個執行器，他只有一個介面：`execute()` ，表示可以執行任務
- ExecutorService在Executor的基礎上拓展了更多的執行方法，如`submit()` `shutdown()` 等等,表示一個任務執行服務。
- AbstarctExecutorService是一個抽象類，他實現了ExecutorService的部分核心方法，如submit等
- ThreadPoolExecutor是最核心的類，也就是執行緒池，他繼承了抽象類AbstarctExecutorService
- 此外還有ScheduledExecutorService介面，他表示一個可以按照指定時間或週期執行的執行器服務，內部定義瞭如`schedule()` 等方法來執行任務
- ScheduledThreadPoolExecutor實現了ScheduledExecutorService介面，同時繼承於ThreadPoolExecutor，內部的執行緒池相關邏輯使用自ThreadPoolExecutor，在此基礎上拓展了延遲、週期執行等功能特性
ScheduledThreadPoolExecutor相對來說用的是比較少。延時任務在我們Android中有更加熟悉的方案：Handler；而週期任務則用的非常少。現在android的後臺限制非常嚴格，基本上一退出應用，應用程序很容易被系統幹掉。當然ScheduledThreadPoolExecutor也不是完全沒有用處，例如桌面小部件需要設定定時重新整理，那麼他就可以派上用場了。
因此，我們本文的原始碼，主要針對ThreadPoolExecutor。在閱讀原始碼之前，我們先來看一下ThreadPoolExecutor內部的結構以及關鍵角色。
## 內部結構
閱讀原始碼前，我們先把ThreadPoolExecutor整個原始碼結構講解一下，形成一個整體概念，再閱讀原始碼就不會迷失在原始碼中了。先來看一下ThreadPoolExecutor的內部結構：
- ThreadPoolExecutor內部有三個關鍵的角色：阻塞佇列、執行緒、以及RejectExecutionHandler（這裡寫個中文名純粹因為不知道怎麼翻譯這個名字），他們的作用在理論篇有詳細介紹，這裡不再贅述。
- 在ThreadPoolExecutor中，一個執行緒對應一個worker物件，工人，非常形象。每個worker內部有一個獨立的執行緒，他會不斷去阻塞佇列獲取任務來執行，也就是呼叫阻塞佇列的 `poll` 或者 `take` 方法，他們區別後面會講。如果佇列沒有任務了，那麼就會阻塞在這裡。
- workQueue，就是阻塞佇列，當核心執行緒已滿之後，任務就會被放置在這裡等待被工人worker領取執行
- RejectExecutionHandler本身是一個介面，ThreadPoolExecutor內部有這樣的一個介面物件，當任務無法被執行會呼叫這個物件的方法。ThreadPoolExecutor提供了該介面的4種實現方案，我們可以直接拿來用，或者自己繼承介面，實現自定義邏輯。在構造執行緒池的時候可以傳入RejectExecutionHandler物件。
- 整個ThreadPoolExecutor中最核心的方法就是execute，他會根據具體的情況來選擇不同的執行方案或者拒絕執行。
這樣，我們就清楚ThreadPoolExecutor的內部結構了，然後，我們開始 Read the fucking code 吧。
## 原始碼分析
#### 內部關鍵屬性
ThreadPoolExecutor內部有很多的變數，他們包含的資訊非常重要，先來了解一下。
ThreadPoolExecutor的狀態和執行緒數整合在同一個int變數中，類似於view測量中MeasureSpec。他的高三位表示執行緒池的狀態，低29位表示執行緒池中執行緒的數量,如下：
```java
// AtomicInteger物件可以利用CAS實現執行緒安全的修改，其中包含了執行緒池狀態和執行緒數量資訊
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// COUNT_BITS=29,（對於int長度為32來說）表示執行緒數量的位元組位數
private static final int COUNT_BITS = Integer.SIZE - 3;
// 狀態掩碼，高三位是1，低29位全是0，可以通過 ctl&COUNT_MASK 運算來獲取執行緒池狀態
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;
```
執行緒池的狀態一共有5個：
- 執行running：執行緒池建立之後即是執行狀態
- 關閉shutdown：呼叫shutdown方法之後執行緒池處於shutdown狀態，該狀態會停止接收任何任務，阻塞佇列中的任務執行完成之後會自動終止執行緒池
- 停止stop：呼叫shutdownNow方法之後執行緒池處於stop狀態。和shutdown的區別是這個狀態下的執行緒池不會去執行佇列中剩下的任務
- 整理tidying：線上程池stop之後，進入tidying狀態，然後執行 `terminated()` 方法，再進入terminated狀態
- 終止terminated：執行緒池中沒有任何執行緒在執行任務，執行緒池完全終止。
在原始碼中這幾個狀態分別對應：
```java
// runState is stored in the high-order bits
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
```
上面的位操作不夠直觀，轉化後如下：
```java
private static final int RUNNING = 111 00000 00000000 00000000 00000000;
private static final int SHUTDOWN = 000 00000 00000000 00000000 00000000;
private static final int STOP = 001 00000 00000000 00000000 00000000;
private static final int TIDYING = 010 00000 00000000 00000000 00000000;
private static final int TERMINATED = 011 00000 00000000 00000000 00000000;
```
可以看到除了running是負數，其他的狀態都是正數，且狀態越靠後，數值越大。因此我們可以通過判斷 `ctl&COUNT_MASK > SHUTDOWN` 來判斷狀態是否處於 stop、tidying、terminated之一。後續原始碼中會有很多的這樣的判斷，舉其中的一個方法：
```java
// 這裡來判斷執行緒池的狀態
if(runStateAtLeast(ctl,SHUTDOWN)) {
...
}
// 這裡執行邏輯，直接判斷兩個數的大小
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
ps：這裡怎麼沒有使用掩碼COUNT_MASK ？因為狀態是處於高位，低位的數值不影響高位的大小判斷。當然如果要判斷相等，就還是需要使用掩碼COUNT_MASK的。
```
接下來是ThreadPoolExecutor內部的三個關鍵角色物件：
```java
// 阻塞佇列
private final BlockingQueue workQueue;
// 儲存worker的hashSet,worker被建立之後會被儲存到這裡
private final HashSet workers = new HashSet<>();
// RejectedExecutionHandler預設的實現是AbortPolicy
private volatile RejectedExecutionHandler handler;
private static final RejectedExecutionHandler defaultHandler =
new AbortPolicy();
```
內部使用的鎖物件：
```java
// 這裡是兩個鎖。ThreadPoolExecutor內部並沒有使用Synchronize關鍵字來保持同步
// 而是使用Lock；和Synchronize的區別就是他是應用層的鎖，而synchronize是jvm層的鎖
private final ReentrantLock mainLock = new ReentrantLock();
private final Condition termination = mainLock.newCondition();
```
最後是內部一些引數的配置，前面都介紹過，把原始碼貼出來再回顧一下：
```java
// 執行緒池歷史達到的最大執行緒數
private int largestPoolSize;
// 執行緒池完成的任務數。
// 該數並不是實時更新的，在獲取執行緒池完成的任務數時，需要去統計每個worker完成的任務並累加起來
// 當一個worker被銷燬之後，他的任務數就會被累加到這個資料中
private long completedTaskCount;
// 執行緒工廠，用於建立執行緒
private volatile ThreadFactory threadFactory;
// 空閒執行緒儲存的時間
private volatile long keepAliveTime;
// 是否允許核心執行緒被回收
private volatile boolean allowCoreThreadTimeOut;
// 核心執行緒數限額
private volatile int corePoolSize;
// 執行緒總數限額
private volatile int maximumPoolSize;
```
不是吧sir？原始碼還沒看到魂呢，整出來這麼無聊的變數？
咳咳，別急嘛，原始碼解析馬上來。這些變數會貫穿整個原始碼過程始終，先對他們有個印象，後續閱讀原始碼就會輕鬆暢通很多。
#### 關鍵方法：execute()
這個方法的主要任務就是根據執行緒池的當前狀態，選擇任務的執行策略。該方法的核心邏輯思路是：
1. 線上程數沒有達到核心執行緒數時，會建立一個核心執行緒來執行任務
```java
public void execute(Runnable command) {
// 不能傳入空任務
if (command == null)
throw new NullPointerException();
// 獲取ctl變數，就是上面我們講的將狀態和執行緒數合在一起的一個變數
int c = ctl.get();
// 判斷核心執行緒數是否超過限額，否則建立一個核心執行緒來執行任務
if (workerCountOf(c) < corePoolSize) {
// addWorker方法是建立一個worker，也就是建立一個執行緒，引數true表示這是一個核心執行緒
// 如果新增成功則直接返回
// 否則意味著中間有其他的worker被添加了，導致超出核心執行緒數;或者執行緒池被關閉了等其他情況
// 需要進入下一步繼續判斷
if (addWorker(command, true))
return;
c = ctl.get();
}
...
}
```
2. 當執行緒數達到核心執行緒數時，新任務會被放入到等待佇列中等待被執行
3. 當等待佇列已經滿了之後，如果執行緒數沒有到達總的執行緒數上限，那麼會建立一個非核心執行緒來執行任務
4. 當執行緒數已經到達總的執行緒數限制時，新的任務會被拒絕策略者處理，執行緒池無法執行該任務。
```java
public void execute(Runnable command) {
...
// 如果執行緒池還在執行,則嘗試新增任務到佇列中
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 再次檢查如果執行緒池被關閉了，那麼把任務移出佇列
// 如果移除成功則拒絕本次任務
// 這裡主要是判斷在插入佇列的過程中，執行緒池有沒有被關閉了
if (! isRunning(recheck) && remove(command))
reject(command);
// 否則再次檢查執行緒數是否為0，如果是，則建立一個沒有任務的非主執行緒worker
// 這裡對應核心執行緒為0的情況，指定任務為null，worker會去佇列拿任務來執行
// 這裡表示執行緒池至少有一個執行緒來執行佇列中的任務
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 如果上面新增到佇列中失敗，則嘗試建立一個非核心執行緒來執行任務
// 如果建立失敗，則拒絕任務
else if (!addWorker(command, false))
reject(command);
}
```
原始碼中還設計到兩個關鍵方法：addWorker建立一個新的worker，也就是建立一個執行緒；reject拒絕一個任務。後者比較簡單我們先看一下。
#### 拒絕任務：reject()
```java
// 拒絕任務，呼叫rejectedExecutionHandler來處理
final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}
```
預設的實現類有4個，我們依次來看一下：
- AbortPolicy是預設實現，會丟擲一個RejectedExecutionException異常：
```java
public static class AbortPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
```
- DiscardPolicy最簡單，就是：什麼都不做，直接拋棄任務。（這是非常渣男不負責任的行為，咱們不能學他，所以也不要用它 [此處狗頭] ）
```java
public static class DiscardPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
```
- DiscardOldestPolicy會刪除佇列頭的一個任務，然後再次執行自己（擠掉原位，自己上位，綠茶行為？）
```java
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
```
- CallerRunsPolicy最猛，他乾脆在自己的執行緒執行run方法，不依靠執行緒池了，自己動手豐衣足食。
```java
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
```
上面4個ThreadPoolExecutor已經幫我們實現了，他的靜態內部類，在建立ThreadPoolExecutor的時候我們可以直接拿來用。也可以自己繼承介面實現自己的邏輯。具體選擇哪個需要根據實際的業務需求來決定。
那麼接下來看建立worker的方法。
#### 建立worker：addWorker()
方法的目的很簡單：建立一個worker。前面我們講到，worker內部建立了一個執行緒，每一個worker則代表了一個執行緒，非常類似android中的looper。looper的loop()方法會不斷地去MessageQueue獲取message，而Worker的run()方法會不斷地去阻塞佇列獲取任務，這個我們後面講。
addWorker() 方法的邏輯整體上分為兩個部分：
1. 檢查**執行緒狀態**和**執行緒數**是否滿足條件：
```java
// 第一個引數是建立的執行緒首次要執行的任務，可以是null，則表示初始化一個執行緒
// 第二引數表示是否是一個核心執行緒
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (int c = ctl.get();;) {
// 還記不記得我們前面講到執行緒池的狀態控制？
// runStateAtLeast(c, SHUTDOWN)表示狀態至少為shutdown，後面類同
// 如果執行緒池處於stop及以上，不會再建立worker
// 如果執行緒池狀態在shutdown時，如果佇列不為空或者任務!=null，則還會建立worker
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP)
|| firstTask != null
|| workQueue.isEmpty()))
// 其他情況返回false，表示拒絕建立worker
return false;
// 這裡採用CAS輪詢，也就是迴圈鎖的策略來讓執行緒總數+1
for (;;) {
// 檢查是否超出執行緒數限制
// 這裡根據core引數判斷是核心執行緒還是非核心執行緒
if (workerCountOf(c)
> shutdownNow() {
List tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 檢查是否可以關閉執行緒
checkShutdownAccess();
// 設定狀態為stop
advanceRunState(STOP);
// 嘗試中斷所有執行緒
interruptWorkers();
// 返回佇列中剩下的任務
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
```
最後再來看一下和 `execute()`不同的提交任務方法：submit。
#### 提交任務：submit()
submit方法並不是ThreadPoolExecutor實現的，而是AbstractExecutorService，如下：
```java
// runnable沒有返回值，建立FutureTask的返回引數傳入null
public Future submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
// 有引數返回值的runnable
// 最終也是構造一個callable來執行，把返回值設定為result
public Future submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
// callable本身就擁有返回值
public Future submit(Callable task) {
if (task == null) throw new NullPointerException();
RunnableFuture ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
```
他們的邏輯都幾乎一樣：呼叫newTaskFor方法來構造一個Future物件並返回。我們看到newTaskFor方法：
```java
// 建立一個FutureTask來返回
protected RunnableFuture newTaskFor(Runnable runnable, T value) {
return new FutureTask(runnable, value);
}
protected RunnableFuture newTaskFor(Callable callable) {
return new FutureTask(callable);
}
```
可以看到這個方法很簡單：構造一個FutureTask並返回，FutureTask也是Future介面目前唯一的實現類。
更加具體關於Future的內容就不展開了，有興趣的讀者可以去了解一下。
## 最後
好了到這裡，關於ThreadPoolExecutor的原始碼分析內容就講完了。最後讓我們再回顧一下吧：
- ThreadPoolExecutor的整個執行流程從execute方法開始，他會根據具體的情況，採用合適的執行方案
- 執行緒被封裝在worker物件中，worker物件通過runWorker方法，會一直不斷地呼叫getTask方法來呼叫佇列的poll或take方法獲取任務
- 當需要退出一個worker時，只要getTask方法返回null即可退出
- 當執行緒池關閉時，會根據不同的關閉方法，等待所有的執行緒執行完成，然後關閉執行緒池。
執行緒池整體的模型和handler是十分類似的：一個生產者-消費者模型。但和Handler不同的是，ThreadPoolExecutor不支援延時任務，這點在ScheduledThreadPoolExecutor得到了實現；Handler的執行緒安全採用synchronize關鍵字，而ThreadPoolExecutor採用的是Lock和一些利用CAS實現執行緒安全的整型變數；Handler無法拒絕任務，執行緒池可以；Handler丟擲異常會直接程式崩潰，而執行緒池不會等等。
瞭解了執行緒池的內部原始碼，對於他更加了解後，那麼可以根據具體的問題，做出更加合適的解決方案。ThreadPoolExecutor還有一些原始碼沒有講到，以及ScheduledThreadPoolExecutor、阻塞佇列的原始碼，有興趣讀者可以自行去深入瞭解一下，拓展關於執行緒池的一切。
全文到此，假期肝文不容易啊，如果文章對你有幫助，求一個大拇指，贊一下再走唄。
## 參考文獻
- 《Java併發程式設計的藝術》：併發程式設計必讀，作者對一些原理講的很透徹
- 《Java核心技術卷》：這系列的書主要是講解框架的使用，不會深入原理，適合入門
- [javaGuide](https://snailclimb.gitee.io/javaguide/#/./docs/java/multi-thread/java執行緒池學習總結?id=_61-%e7%ae%80%e4%bb%8b)：javaGuide，對java知識總結得很不錯的一個部落格
- [Java併發程式設計：執行緒池的使用](https://www.cnblogs.com/dolphin0520/p/3932921.html)：部落格園上一位很優秀的博主，文章寫得通俗易懂且不失深度
> 全文到此，原創不易，覺得有幫助可以點贊收藏評論轉發。
> 筆者才疏學淺，有任何想法歡迎評論區交流指正。
> 如需轉載請評論區或私信交流。
>
> 另外歡迎光臨筆者的個人部落格：[傳送門](https://qwerhuan.gi