追蹤解析 ThreadPoolExecutor 原始碼

摘要： 零 前期準備 0 FBI WARNING 文章異常囉嗦且繞彎。 1 版本 JDK 版本 : OpenJDK 11.0.1 IDE : idea 2018.3 2 ThreadPoolExecutor 簡介 ThreadPoolExecutor 是 jdk4 中加入的...

零 前期準備

0 FBI WARNING

文章異常囉嗦且繞彎。

1 版本

JDK 版本 : OpenJDK 11.0.1

IDE : idea 2018.3

2 ThreadPoolExecutor 簡介

ThreadPoolExecutor 是 jdk4 中加入的工具，被封裝在 jdk 自帶的 Executors 框架中，是 java 中最經典的執行緒池技術。

ThreadPoolExecutor 類在 concurrent 包下，和其它執行緒工具類一樣都由 Doug Lea 大神操刀完成。

[ 在看完 Spring ioc 和 Gson 之後有點乏了，換換口味看一些 jdk 的原始碼 ]

3 Demo

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadPoolDemo {

public static void main(String[] args){
//建立執行緒池
//這裡使用固定執行緒數的執行緒池，執行緒數為 5
ExecutorService executorService = Executors.newFixedThreadPool(5);

for(int i = 0 ; i < 100 ; i ++){
final int ii = i;
//建立 Runnable 作為執行緒池的任務
Runnable r = () -> System.out.println(ii);
//執行
executorService.execute(r);
}
}
}

一 執行緒池的初始化

執行緒池的初始化呼叫的 Executors 框架的靜態方法：

//Executors.class
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}

繼續追蹤這個構造方法：

//ThreadPoolExecutor.class
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}

繼續追蹤：

//ThreadPoolExecutor.class
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> 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.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();

//執行緒數
this.corePoolSize = corePoolSize;
//最大執行緒數
//本例中使用固定執行緒數的執行緒池，所以執行緒數和最大執行緒數相等
this.maximumPoolSize = maximumPoolSize;
//用於儲存任務的佇列
//此處使用 LinkedBlockingQueue 來儲存任務，其執行緒安全
this.workQueue = workQueue;
//keepAliveTime 引數用於表示：
//對於超出執行緒和佇列快取總和的任務，是否要臨時增加執行緒來處理
//超出的執行緒的存在時間是多少
//這裡使用的是定長執行緒池，所以 keepAliveTime = 0，即不增加執行緒
this.keepAliveTime = unit.toNanos(keepAliveTime);
//用於建立執行緒的工廠類
this.threadFactory = threadFactory;
//handler 用來處理 task 太多時候的拒絕策略
//此例中使用的是預設的，即定義在 ThreadPoolExecutor 中的 defaultHandler 物件
this.handler = handler;
}

二 Worker

Worker 是 ThreadPoolExecutor 的內部類，可以看做是 Runnable 的代理類：

//ThreadPoolExecutor.class
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{

private static final long serialVersionUID = 6138294804551838833L;
final Thread thread;
Runnable firstTask;
//完成 task 數量的計數器
volatile long completedTasks;

Worker(Runnable firstTask) {
//這個方法是 AbstractQueuedSynchronizer 中的方法，功能相當於加鎖
//-1 的意思是後續的任務會處於阻塞狀態，即為已經加鎖
setState(-1);
//在建立的時候存入一個要處理的 task
//需要注意的是每個 worker 物件被創建出來之後是可以重複利用來處理多個 task 的
this.firstTask = firstTask;
//worker 會用自身作為 Runnable 物件去建立一個執行緒
//這裡呼叫執行緒工廠進行執行緒建立
this.thread = getThreadFactory().newThread(this);
}

//對於執行緒變數來說，其啟動的就是 worker 的 run() 方法
public void run() {
//runWorker(...) 方法在 ThreadPoolExecutor 裡
runWorker(this);
}

//獲取鎖的狀態
protected boolean isHeldExclusively() {
return getState() != 0;
}
//重寫了 AbstractQueuedSynchronizer 中的 tryAcquire(...) 方法
//嘗試加鎖
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
//重寫了 AbstractQueuedSynchronizer 中的 tryRelease(...) 方法
//嘗試釋放鎖
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
//真正的加鎖方法
public void lock() { 
acquire(1); 
}
//嘗試加鎖
public boolean tryLock() { 
return tryAcquire(1); 
}
//真正的釋放鎖方法
public void unlock() { 
release(1); 
}
//判斷是否在鎖中
public boolean isLocked() { 
return isHeldExclusively(); 
}
//中斷執行緒
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}

追蹤一下 runWorker(...) 方法：

//ThreadPoolExecutor.class
final void runWorker(Worker w) {
//獲取當前所在的執行緒的例項物件
Thread wt = Thread.currentThread();
//獲取 task
Runnable task = w.firstTask;
//取出來之後把 task 置空
w.firstTask = null;
//此處釋放鎖
w.unlock();
//指示器，此變數為 true 的時候確認該方法已經執行完畢
boolean completedAbruptly = true;
try {
//此處為一個 while 迴圈，用於不斷的執行 task
//getTask() 方法會從佇列裡不斷抓取 task 並進行執行
//當 task 為 null，且佇列裡已經沒有更多 task 的時候，就會終止迴圈
while (task != null || (task = getTask()) != null) {
//加鎖，獨佔執行緒
w.lock();
//在這裡會判斷執行緒的狀態，如果存在符合中斷的情況，就會直接中斷掉
if ((runStateAtLeast(ctl.get(), STOP) 
|| (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted())
wt.interrupt();

try {
//beforeExecute(...) 和 afterExecute(...) 方法在 ThreadPoolExecutor 中並沒有實現
//是預留出來給使用者重寫，以達到業務需求的方法
beforeExecute(wt, task);
try {
//此處執行 task
task.run();
afterExecute(task, null);
} catch (Throwable ex) {
afterExecute(task, ex);
throw ex;
}
} finally {
//將執行的 task 置空
task = null;
//每完成一個 task 就會加 1
w.completedTasks++;
//釋放鎖
w.unlock();
}
}
completedAbruptly = false;
} finally {
//這個方法會銷燬掉 worker
//同時如果檢測到有新的 task 又會重新建立 Worker
processWorkerExit(w, completedAbruptly);
}
}

Worker 是執行緒池中真正起完成業務邏輯的元件，是任務和執行緒的封裝。

三 執行緒池的狀態控制

執行緒池的狀態主要由 ctl 變數來進行控制：

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

ctl 是一個 AtomicInteger 型別的變數，其實可以簡單理解為一個 int 值，AtomicInteger 只是能夠適應高併發的原子化操作的需要。

ctl 的前 29 位數用來表示執行緒(Worker)的數量，後面三位用來表示執行緒池的狀態。

執行緒池的狀態有五種，分別是 Running、Shutdown、Stop、Tidying、Terminate，根據單詞就能猜出大概。

注意的是，這五種狀態線上程池中都以 int 變數的形式存在，從前到後依次變大，對狀態的比較有一系列方法：

//ThreadPoolExecutor.class
private static boolean runStateLessThan(int c, int s) {
//c 的狀態值要小於 s
return c < s;
}
//ThreadPoolExecutor.class
private static boolean runStateAtLeast(int c, int s) {
//c 的狀態值要大於或等於 s
return c >= s;
}
//ThreadPoolExecutor.class
private static boolean isRunning(int c) {
//狀態裡只有 RUNNING 是小於 SHUTDOWN 的
return c < SHUTDOWN;
}

在這些方法裡，傳入的引數 c 一般指的是當前執行緒池狀態，s 是用來對比的參照狀態。

四 執行緒池的執行

該 part 的起點：

executorService.execute(r);

來追蹤 execute(...) 方法：

public void execute(Runnable command) {
//有效性驗證
if (command == null)
throw new NullPointerException();

//ctl 是一個 AtomicInteger 型別的變數，用來記錄執行緒池的狀態
int c = ctl.get();

//workerCountOf(...) 方法會返回當前執行的 Worker 的數量
if (workerCountOf(c) < corePoolSize) {
//Worker 的數量小於執行緒池容量的情況下
//直接增加 Worker 並取出 task 去執行
if (addWorker(command, true))
return;
//如果 Worker 已經順利執行了 task，應該會直接返回掉
//如果執行中出現了其它情況，則會繼續往下走
//此處重新整理狀態
c = ctl.get();
}
//當 Worker 數量已經達到執行緒池的指定數量，或者新增 Worker 的時候出問題的時候，會進入此判斷語句
//先判斷執行緒池是否處於活躍狀態，且 task 是否已經被成功新增到佇列中
//如果不滿足，會進入 else 語句中，先最後嘗試一次 addWorker(...) 方法，如果不成功就拒絕 task
//reject(...) 方法會呼叫 handler 的拒絕策略
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}else if (!addWorker(command, false))
reject(command);
}

1 reject

這裡先提及一下 reject(...) 方法：

//ThreadPoolExecutor.class
final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}

本質是呼叫了 handler 物件的相關方法。在本例中，handler 物件指向了 defaultHandler：

//ThreadPoolExecutor.class
private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();

defaultHandler 是一個 AbortPolicy 型別的物件，而 AbortPolicy 是 ThreadPoolExecutor 的靜態內部類。

AbortPolicy 起作用的方法為 rejectedExecution(...) 方法：

//AbortPolicy.class
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " + e.toString());
}

也就是說，在 task 過多的情況下，AbortPolicy 的應對策略是丟擲異常。

2 addWorker

來看一下核心方法 addWorker(...)：

//ThreadPoolExecutor.class
private boolean addWorker(Runnable firstTask, boolean core) {
//先標記這個 for 迴圈，方便退出迴圈
retry:
//在每一次迴圈開始之前會重新整理一次狀態標識
for (int c = ctl.get();;) {
//這裡先進行判斷，如果執行緒池已經關閉了，或者沒有 task 了，就會返回 false
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP)
|| firstTask != null
|| workQueue.isEmpty()))
return false;

for (;;) {
//如果 Worker 數量已經超出了最大值就會直接返回 false
if (workerCountOf(c)
>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
return false;
//將 ctl 變數的值加 1，如果成功了就會跳出迴圈
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();
//在狀態值比 SHUTDOWN 大的時候會直接跳到最外頭的迴圈裡
//需要注意的是最外面的 for 迴圈會判斷狀態值是否大於 SHUTDOWN
//如果大於 SHUTDOWN 的話就返回 false 了
if (runStateAtLeast(c, SHUTDOWN))
continue retry;
}
}

boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//建立一個 Worker
w = new Worker(firstTask);
//獲取執行緒物件
final Thread t = w.thread;
if (t != null) {
//加鎖，此處加的是一把全域性的鎖
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int c = ctl.get();
//如果狀態值 c 是 RUNNING，或者 [c 是 RUNNING 或者 SHUTDOWN 且 firstTask 是 null] 就會進入這個判斷語句
//
if (isRunning(c) || (runStateLessThan(c, STOP) && firstTask == null)) {
//如果這個執行緒已經處於運作狀態，會丟擲異常
if (t.isAlive())
throw new IllegalThreadStateException();
//workers 是一個列表，用於儲存 Worker 物件
workers.add(w);
//獲取 Worker 的數量
int s = workers.size();
//largestPoolSize 用來記錄執行緒池達到過的最大執行緒數
if (s > largestPoolSize)
largestPoolSize = s;
//標記 Worker 已經被新增
workerAdded = true;
}
} finally {
//釋放鎖
mainLock.unlock();
}
//先判斷 Worker 是否已經被新增到 workers 內了
if (workerAdded) {
//這是該方法核心的啟動執行緒方法
t.start();
//標記 Worker 已經開始運行了
workerStarted = true;
}
}
} finally {
//如果沒有標記 Worker 已經開始工作，會在這裡銷燬掉 Worker
if (!workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}

五 一點嘮叨

先總結一下執行緒池的業務邏輯：

1 接收到 task (即實現了 Runnable 介面的例項物件) [execute(...) 方法]

2 用 task 去嘗試建立一個 Worker 例項 [execute(...) 方法]
2.1 如果 Worker 數量沒有達到執行緒池的指定最大值 -> 新建
2.2 如果 Worker 數量達到了執行緒池的指定最大值 -> 不會再建立，而是把 task 儲存起來等待空閒的 Worker 去提取
2.3 如果 task 佇列也已經滿了，無法再新增 -> 觸發拒絕機制(handler)

3 Worker 在執行的時候呼叫其內部的 Thread 例項物件的 start() 方法 [addWorker(...) 方法]

4 該 start() 方法會呼叫到 Worker 的 run() 方法 [Worker.class 內的 run() 方法]

5 Worker 的 run() 方法本質上是封裝了 task 的 run() 方法 [runWorker(...) 方法]

主線業務邏輯不算複雜，比較艱難的是為了保證資料的一致性，執行緒池程式碼中充斥著大量的狀態判斷和鎖機制。

並且為了考慮效能問題，執行緒池的設計沒有使用悲觀鎖(synchronized 關鍵字)，而是大量使用了 ASQ 和 ReetrentLock 機制。

本文僅為個人的學習筆記，可能存在錯誤或者表述不清的地方，有緣補充