硬核乾貨：4W字從原始碼上分析JUC執行緒池ThreadPoolExecutor的實現原理
阿新 • • 發佈：2020-08-28
## 前提
很早之前就打算看一次JUC執行緒池`ThreadPoolExecutor`的原始碼實現，由於近段時間比較忙，一直沒有時間整理出原始碼分析的文章。之前在分析擴充套件執行緒池實現可回撥的`Future`時候曾經提到併發大師`Doug Lea`在設計執行緒池`ThreadPoolExecutor`的提交任務的頂層介面`Executor`只有一個無狀態的執行方法：
```java
public interface Executor {
void execute(Runnable command);
}
```
而`ExecutorService`提供了很多擴充套件方法底層基本上是基於`Executor#execute()`方法進行擴充套件。本文著重分析`ThreadPoolExecutor#execute()`的實現，筆者會從實現原理、原始碼實現等角度結合簡化例子進行詳細的分析。`ThreadPoolExecutor`的原始碼從`JDK8`到`JDK11`基本沒有變化，本文編寫的時候使用的是`JDK11`。
## ThreadPoolExecutor的原理
`ThreadPoolExecutor`裡面使用到JUC同步器框架`AbstractQueuedSynchronizer`（俗稱`AQS`）、大量的位操作、`CAS`操作。`ThreadPoolExecutor`提供了固定活躍執行緒（核心執行緒）、額外的執行緒（執行緒池容量 - 核心執行緒數這部分額外建立的執行緒，下面稱為非核心執行緒）、任務佇列以及拒絕策略這幾個重要的功能。
### JUC同步器框架
`ThreadPoolExecutor`裡面使用到JUC同步器框架，主要用於四個方面：
- 全域性鎖`mainLock`成員屬性，是可重入鎖`ReentrantLock`型別，主要是用於訪問工作執行緒`Worker`集合和進行資料統計記錄時候的加鎖操作。
- 條件變數`termination`，`Condition`型別，主要用於執行緒進行等待終結`awaitTermination()`方法時的帶期限阻塞。
- 任務佇列`workQueue`，`BlockingQueue`型別，任務佇列，用於存放待執行的任務。
- 工作執行緒，內部類`Worker`型別，是執行緒池中真正的工作執行緒物件。
關於`AQS`筆者之前寫過一篇相關原始碼分析的文章：[JUC同步器框架AbstractQueuedSynchronizer原始碼圖文分析](http://www.throwable.club/2019/04/07/java-juc-aqs-source-code)。
### 核心執行緒
這裡先參考`ThreadPoolExecutor`的實現並且進行簡化，實現一個只有核心執行緒的執行緒池，要求如下：
- 暫時不考慮任務執行異常情況下的處理。
- 任務佇列為無界佇列。
- 執行緒池容量固定為核心執行緒數量。
- 暫時不考慮拒絕策略。
```java
public class CoreThreadPool implements Executor {
private BlockingQueue workQueue;
private static final AtomicInteger COUNTER = new AtomicInteger();
private int coreSize;
private int threadCount = 0;
public CoreThreadPool(int coreSize) {
this.coreSize = coreSize;
this.workQueue = new LinkedBlockingQueue<>();
}
@Override
public void execute(Runnable command) {
if (++threadCount <= coreSize) {
new Worker(command).start();
} else {
try {
workQueue.put(command);
} catch (InterruptedException e) {
throw new IllegalStateException(e);
}
}
}
private class Worker extends Thread {
private Runnable firstTask;
public Worker(Runnable runnable) {
super(String.format("Worker-%d", COUNTER.getAndIncrement()));
this.firstTask = runnable;
}
@Override
public void run() {
Runnable task = this.firstTask;
while (null != task || null != (task = getTask())) {
try {
task.run();
} finally {
task = null;
}
}
}
}
private Runnable getTask() {
try {
return workQueue.take();
} catch (InterruptedException e) {
throw new IllegalStateException(e);
}
}
public static void main(String[] args) throws Exception {
CoreThreadPool pool = new CoreThreadPool(5);
IntStream.range(0, 10)
.forEach(i -> pool.execute(() ->
System.out.println(String.format("Thread:%s,value:%d", Thread.currentThread().getName(), i))));
Thread.sleep(Integer.MAX_VALUE);
}
}
```
某次執行結果如下：
```java
Thread:Worker-0,value:0
Thread:Worker-3,value:3
Thread:Worker-2,value:2
Thread:Worker-1,value:1
Thread:Worker-4,value:4
Thread:Worker-1,value:5
Thread:Worker-2,value:8
Thread:Worker-4,value:7
Thread:Worker-0,value:6
Thread:Worker-3,value:9
```
設計此執行緒池的時候，核心執行緒是懶建立的，如果執行緒空閒的時候則阻塞在任務佇列的`take()`方法，其實對於`ThreadPoolExecutor`也是類似這樣實現，只是如果使用了`keepAliveTime`並且允許核心執行緒超時（`allowCoreThreadTimeOut`設定為`true`）則會使用`BlockingQueue#poll(keepAliveTime)`進行輪詢代替永久阻塞。
### 其他附加功能
構建`ThreadPoolExecutor`例項的時候，需要定義`maximumPoolSize`（執行緒池最大執行緒數）和`corePoolSize`（核心執行緒數）。當任務佇列是有界的阻塞佇列，核心執行緒滿負載，任務佇列已經滿的情況下，會嘗試建立額外的`maximumPoolSize - corePoolSize`個執行緒去執行新提交的任務。當`ThreadPoolExecutor`這裡實現的兩個主要附加功能是：
- 一定條件下會建立非核心執行緒去執行任務，非核心執行緒的回收週期（執行緒生命週期終結時刻）是`keepAliveTime`，執行緒生命週期終結的條件是：下一次通過任務佇列獲取任務的時候並且存活時間超過`keepAliveTime`。
- 提供拒絕策略，也就是在核心執行緒滿負載、任務佇列已滿、非核心執行緒滿負載的條件下會觸發拒絕策略。
## 原始碼分析
先分析執行緒池的關鍵屬性，接著分析其狀態控制，最後重點分析`ThreadPoolExecutor#execute()`方法。
### 關鍵屬性
```java
public class ThreadPoolExecutor extends AbstractExecutorService {
// 控制變數-存放狀態和執行緒數
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 任務佇列，必須是阻塞佇列
private final BlockingQueue workQueue;
// 工作執行緒集合，存放執行緒池中所有的（活躍的）工作執行緒，只有在持有全域性鎖mainLock的前提下才能訪問此集合
private final HashSet workers = new HashSet<>();
// 全域性鎖
private final ReentrantLock mainLock = new ReentrantLock();
// awaitTermination方法使用的等待條件變數
private final Condition termination = mainLock.newCondition();
// 記錄峰值執行緒數
private int largestPoolSize;
// 記錄已經成功執行完畢的任務數
private long completedTaskCount;
// 執行緒工廠，用於建立新的執行緒例項
private volatile ThreadFactory threadFactory;
// 拒絕執行處理器，對應不同的拒絕策略
private volatile RejectedExecutionHandler handler;
// 空閒執行緒等待任務的時間週期，單位是納秒
private volatile long keepAliveTime;
// 是否允許核心執行緒超時，如果為true則keepAliveTime對核心執行緒也生效
private volatile boolean allowCoreThreadTimeOut;
// 核心執行緒數
private volatile int corePoolSize;
// 執行緒池容量
private volatile int maximumPoolSize;
// 省略其他程式碼
}
```
下面看引數列表最長的建構函式：
```java
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
```
可以自定義核心執行緒數、執行緒池容量（最大執行緒數）、空閒執行緒等待任務週期、任務佇列、執行緒工廠、拒絕策略。下面簡單分析一下每個引數的含義和作用：
- **`corePoolSize`**：int型別，核心執行緒數量。
- **`maximumPoolSize`**：int型別，最大執行緒數量，也就是執行緒池的容量。
- **`keepAliveTime`**：long型別，執行緒空閒等待時間，也和工作執行緒的生命週期有關，下文會分析。
- **`unit`**：`TimeUnit`型別，`keepAliveTime`引數的時間單位，實際上`keepAliveTime`最終會轉化為納秒。
- **`workQueue`**：`BlockingQueue`型別，等待佇列或者叫任務佇列。
- **`threadFactory`**：`ThreadFactory`型別，執行緒工廠，用於建立工作執行緒（包括核心執行緒和非核心執行緒），預設使用`Executors.defaultThreadFactory()`作為內建執行緒工廠例項，一般自定義執行緒工廠才能更好地跟蹤工作執行緒。
- **`handler`**：`RejectedExecutionHandler`型別，執行緒池的拒絕執行處理器，更多時候稱為拒絕策略，拒絕策略執行的時機是當阻塞佇列已滿、沒有空閒的執行緒（包括核心執行緒和非核心執行緒）並且繼續提交任務。提供了4種內建的拒絕策略實現：
- `AbortPolicy`：直接拒絕策略，也就是不會執行任務，直接丟擲`RejectedExecutionException`，這是**預設的拒絕策略**。
- `DiscardPolicy`：拋棄策略，也就是直接忽略提交的任務（通俗來說就是空實現）。
- `DiscardOldestPolicy`：拋棄最老任務策略，也就是通過`poll()`方法取出任務佇列隊頭的任務拋棄，然後執行當前提交的任務。
- `CallerRunsPolicy`：呼叫者執行策略，也就是當前呼叫`Executor#execute()`的執行緒直接呼叫任務`Runnable#run()`，**一般不希望任務丟失會選用這種策略，但從實際角度來看，原來的非同步呼叫意圖會退化為同步呼叫**。
### 狀態控制
狀態控制主要圍繞原子整型成員變數`ctl`：
```java
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;
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;
// 通過ctl值獲取執行狀態
private static int runStateOf(int c) { return c & ~COUNT_MASK; }
// 通過ctl值獲取工作執行緒數
private static int workerCountOf(int c) { return c & COUNT_MASK; }
// 通過執行狀態和工作執行緒數計算ctl的值，或運算
private static int ctlOf(int rs, int wc) { return rs | wc; }
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
// CAS操作執行緒數增加1
private boolean compareAndIncrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect + 1);
}
// CAS操作執行緒數減少1
private boolean compareAndDecrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect - 1);
}
// 執行緒數直接減少1
private void decrementWorkerCount() {
ctl.addAndGet(-1);
}
```
接下來分析一下執行緒池的狀態變數，工作執行緒上限數量位的長度是`COUNT_BITS`，它的值是`Integer.SIZE - 3`，也就是正整數29：
> shutdownNow() {
List tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 許可權校驗，安全策略相關判斷
checkShutdownAccess();
// 設定STOP狀態
advanceRunState(STOP);
// 中斷所有的工作執行緒
interruptWorkers();
// 清空工作佇列並且取出所有的未執行的任務
tasks = drainQueue();
} finally {
mainLock.unlock();
}
// 呼叫上面分析果敢的嘗試terminate方法，使狀態更變為TIDYING，執行鉤子方法terminated()後，最終狀態更新為TERMINATED
tryTerminate();
return tasks;
}
// 遍歷所有的工作執行緒，如果state > 0（啟動狀態）則進行中斷
private void interruptWorkers() {
// assert mainLock.isHeldByCurrentThread();
for (Worker w : workers)
w.interruptIfStarted();
}
```
`shutdownNow()`方法會把執行緒池狀態先更變為`STOP`，中斷所有 的工作執行緒（`AbstractQueuedSynchronizer`的`state`值大於0的`Worker`例項，也就是包括正在執行任務的`Worker`和空閒的`Worker`），然後遍歷任務佇列，取出（移除）所有任務存放在一個列表中返回。
最後看`awaitTermination()`方法：
```java
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
// 轉換timeout的單位為納秒
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 迴圈等待直到執行緒池狀態更變為TERMINATED，每輪迴圈等待nanos納秒
while (runStateLessThan(ctl.get(), TERMINATED)) {
if (nanos <= 0L)
return false;
nanos = termination.awaitNanos(nanos);
}
return true;
} finally {
mainLock.unlock();
}
}
```
`awaitTermination()`雖然不是`shutdown()`方法體系，但是它的處理邏輯就是確保呼叫此方法的執行緒會阻塞到`tryTerminate()`方法成功把執行緒池狀態更新為`TERMINATED`後再返回，可以使用在某些需要感知執行緒池終結時刻的場景。
有一點值得關注的是：`shutdown()`方法**只會中斷空閒的工作執行緒**，如果工作執行緒正在執行任務物件`Runnable#run()`，這種情況下的工作執行緒不會中斷，而是等待下一輪執行`getTask()`方法的時候通過執行緒池狀態判斷正常終結該工作執行緒。
### 理解可重入鎖mainLock成員變數
> private final ReentrantLock mainLock = new ReentrantLock();
> private final Condition termination = mainLock.newCondition();
先看了`ThreadPoolExecutor`內部成員屬性`mainLock`的引用情況：
歸結一下`mainLock`的使用場景：
|方法|主要作用|
|:-:|:-:|
|`tryTerminate`|保證狀態`TIDYING -> TERMINATED`，鉤子方法`terminated()`回撥和條件變數喚醒|
|`interruptIdleWorkers`|保護工作執行緒中斷的序列化，避免"中斷風暴"|
|`addWorker`|保護工作執行緒集合避免併發增加工作執行緒、保護度量統計資料變更|
|`processWorkerExit`|保護度量統計資料變更|
|`shutdown`、`shutdownNow`和`awaitTermination`|見下文分析|
|`getPoolSize`、`getActiveCount`、`getLargestPoolSize`、`getTaskCount`和`getCompletedTaskCount`|保護度量統計資料讀取，這些統計資料來一般源於`Worker`集合的屬性統計|
這裡分析一下執行緒池如何通過可重入鎖和條件變數實現相對優雅地關閉。先看`shutdown()`方法：
```java
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN);
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}
```
這裡`shutdown()`中除了`tryTerminate()`，其他它方法都是包裹在鎖裡面執行，**確保工作執行緒集合穩定性以及關閉許可權、確保狀態變更序列化，中斷所有工作執行緒並且避免工作執行緒"中斷風暴"**（多次併發呼叫`shutdown()`如果不加鎖，會反覆中斷工作執行緒）。
```java
public List shutdownNow() {
List tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);
interruptWorkers();
tasks = drainQueue(); # <--- 多了這一步
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
```
`shutdownNow()`方法其實加鎖的目的和`shutdown()`差不多，不過多了一步：匯出任務佇列中的剩餘的任務例項列表。`awaitTermination()`方法中使用到前面提到過的條件變數`termination`：
```java
// 條件變數必須在鎖程式碼塊中執行，和synchronized關鍵字用法差不多
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 死迴圈確保等待執行和狀態變更為TERMINATED
while (runStateLessThan(ctl.get(), TERMINATED)) {
if (nanos <= 0L)
return false;
nanos = termination.awaitNanos(nanos); # <-- 確保當前呼叫執行緒阻塞等待對應的時間或者執行緒池狀態變更為TERMINATED，再退出等待
}
return true;
} finally {
mainLock.unlock();
}
}
```
`awaitTermination()`方法的核心功能是：確保當前呼叫`awaitTermination()`方法的執行緒阻塞等待對應的時間或者執行緒池狀態變更為`TERMINATED`，再退出等待返回結果，這樣能夠讓使用者輸入一個可以接受的等待時間進行阻塞等待，或者執行緒池在其他執行緒中被呼叫了`shutdown()`方法狀態變更為`TERMINATED`就能正常解除阻塞。`awaitTermination()`方法的返回值為布林值，`true`代表執行緒池狀態變更為`TERMINATED`或者等待了輸入時間範圍內的時間週期被喚醒，意味則執行緒池正常退出，結果為`false`代表等待了超過輸入時間範圍內的時間週期，執行緒池的狀態依然沒有更變為`TERMINATED`。
> 執行緒池中的工作執行緒如何優雅地退出，不導致當前任務執行丟失、任務狀態異常或者任務持有的資料異常，是一個很值得探討的專題，以後有機會一定會分析一下這個專題。
### reject方法原始碼分析
`reject(Runnable command)`方法很簡單：
```java
final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}
```
呼叫執行緒池持有的成員`RejectedExecutionHandler`例項回撥任務例項和當前執行緒池例項。
### 鉤子方法分析
到`JDK11`為止，`ThreadPoolExecutor`提供的鉤子方法沒有增加，有以下幾個：
- `beforeExecute(Thread t, Runnable r)`：任務物件`Runnable#run()`執行之前觸發回撥。
- `afterExecute(Runnable r, Throwable t)`：任務物件`Runnable#run()`執行之後（包括異常完成情況和正常完成情況）觸發回撥。
- `terminated()`：執行緒池關閉的時候，狀態更變為`TIDYING`成功之後會回撥此方法，執行此方法完畢後，執行緒池狀態會更新為`TERMINATED`。
- `onShutdown()`：`shutdown()`方法執行時候會回撥此方法，API註釋中提到此方法主要提供給`ScheduledThreadPoolExecutor`使用。
其中`onShutdown()`的方法修飾符為`default`，其他三個方法的修飾符為`protected`，必要時候可以自行擴充套件這些方法，可以實現監控、基於特定時機觸發具體操作等等。
### 其他方法
執行緒池本身提供了大量資料統計相關的方法、擴容方法、預建立方法等等，這些方法的原始碼並不複雜，這裡不做展開分析。
**核心執行緒相關：**
- `getCorePoolSize()`：獲取核心執行緒數。
- `setCorePoolSize()`：重新設定執行緒池的核心執行緒數。
- `prestartCoreThread()`：預啟動一個核心執行緒，當且僅當工作執行緒數量小於核心執行緒數量。
- `prestartAllCoreThreads()`：預啟動所有核心執行緒。
**執行緒池容量相關：**
- `getMaximumPoolSize()`：獲取執行緒池容量。
- `setMaximumPoolSize()`：重新設定執行緒池的最大容量。
**執行緒存活週期相關：**
- `setKeepAliveTime()`：設定空閒工作執行緒的存活週期。
- `getKeepAliveTime()`：獲取空閒工作執行緒的存活週期。
**其他監控統計相關方法：**
- `getTaskCount()`：獲取所有已經被執行的任務總數的近似值。
- `getCompletedTaskCount()`：獲取所有已經執行完成的任務總數的近似值。
- `getLargestPoolSize()`：獲取執行緒池的峰值執行緒數（最大池容量）。
- `getActiveCount()`：獲取所有活躍執行緒總數（正在執行任務的工作執行緒）的近似值。
- `getPoolSize()`：獲取工作執行緒集合的容量（當前執行緒池中的總工作執行緒數）。
**任務佇列操作相關方法：**
- `purge()`：移除任務佇列中所有是`Future`型別並且已經處於`Cancelled`狀態的任務。
- `remove()`：從任務佇列中移除指定的任務。
- `BlockingQueue getQueue()`：獲取任務佇列的引用。
有部分屬性值的設定有可能影響到執行緒池中的狀態或者工作執行緒的增減等，例如核心執行緒數改變，有可能會直接增減`Worker`，這裡就以`ThreadPoolExecutor#setCorePoolSize()`為例：
```java
// 設定核心執行緒數量
public void setCorePoolSize(int corePoolSize) {
// 輸入值不能小於0或者大於執行緒池的容量
if (corePoolSize < 0 || maximumPoolSize < corePoolSize)
throw new IllegalArgumentException();
// delta = 傳入核心執行緒數和現存的核心執行緒數的差值
int delta = corePoolSize - this.corePoolSize;
this.corePoolSize = corePoolSize;
// 如果當前執行緒池工作執行緒的總量大於傳入核心執行緒數，則中斷所有的工作執行緒
if (workerCountOf(ctl.get()) > corePoolSize)
interruptIdleWorkers();
else if (delta > 0) {
// 傳入核心執行緒數和現存的核心執行緒數的差值大於0，也就是核心執行緒擴容
// 計算傳入核心執行緒數和現存的核心執行緒數的差值和任務佇列中任務個數的最小值，並且新增這個最小值個數的工作執行緒池
// 任務佇列為空的情況下，k === 0，此時第一個條件 k--> 0就不滿足，不會進入迴圈，那麼這delta個需要建立的工作執行緒應該是在提交新任務的時候懶建立
int k = Math.min(delta, workQueue.size());
while (k-- > 0 && addWorker(null, true)) {
// 如果任務佇