Java併發之ThreadPoolExecutor原始碼解析（三）
阿新 • • 發佈：2021-03-01
Worker
先前，筆者講解到ThreadPoolExecutor.addWorker(Runnable firstTask, boolean core)，在這個方法中工作執行緒可能建立成功，也可能建立失敗，具體視執行緒池的邊界條件，以及當前記憶體情況而定。
那麼，如果執行緒池當前的狀態，是允許建立Worker物件的，那麼建立Worker的內部流程又是怎樣呢？執行緒池為何要使用Worker包裝Thread來建立一個執行緒，為何不直接使用原生的Thread來建立執行緒？如果建立Worker的firstTask不為空，那麼Worker理所當然應該優先執行firstTask任務，如果firstTask為空，那Worker又要如何獲取任務來執行呢？我們還有一堆亟待解決的問題。
首先我們來解決前兩個問題，Worker的建立流程，以及為什麼不使用原生Thread代替Worker？首先，Doug Lea用Worker包裝Thread，意味著Worker比Thread擁有更多的功能。例如：Worker會統計它所對應的執行緒執行了多少任務、通過Worker可以知道執行緒是否已啟動、執行緒是否正在執行任務？而這些資訊都是原生Thread所沒有的，所以需要一個Worker類來擴充套件Thread。
在建立Worker時，會先設定其state的值為-1，代表Worker所對應的執行緒尚未啟動，即還沒有呼叫Worker.thread.start()，之後會進行firstTask的賦值，向執行緒工廠申請建立執行緒，建立完畢後，等待外部呼叫Worker.thread.start()啟動一個執行緒執行Worker.run()方法。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
/*
* 當初始化一個Worker時，會向執行緒工廠申請建立一個Thread物件
* 用來執行任務，為null代表執行緒工廠建立失敗。
*/
final Thread thread;
//首要執行任務，該欄位可能為null。
Runnable firstTask;
//thread已完成任務數。
volatile long completedTasks;
Worker(Runnable firstTask) {
/*
* state初始為-1，代表還未呼叫Worker.thread.start()，
* Worker對應的執行緒尚未被建立，還不能中斷。執行緒啟動後，
* 如果執行緒正在執行任務，state為1，如果執行緒啟動後沒在
* 執行任務的狀態則state為0。
*/
setState(-1);//<1>
this.firstTask = firstTask;
/*
* 建立Thread物件的時候，會把Worker物件本身傳入，而Worker
* 本身實現了Runnable介面，當呼叫thead.start()啟動一個執行緒
* 執行thread.run()時，會進而呼叫Worker.run()方法。
*/
this.thread = getThreadFactory().newThread(this);
}
//Worker物件將任務的執行委託給ThreadPoolExecutor.runWorker(Worker w).
public void run() {
runWorker(this);
}
//判斷Worker是否處於被某個執行緒持有狀態。
protected boolean isHeldExclusively() {
return getState() != 0;//<2>
}
/*
* 執行緒嘗試持有worker物件，如果worker沒有被某個執行緒
* 持有，則state為0，則用CAS的方式將worker的state
* 改為1，並設定exclusiveOwnerThread為當前執行緒。
*/
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {//<3>
setExclusiveOwnerThread(Thread.currentThread());//<4>
return true;
}
return false;
}
/*
* 執行緒釋放worker物件，將state改為0，exclusiveOwnerThread
* 改為null。
*/
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
/*
* 呼叫父類acquire(int arg)時，會進而呼叫到Worker本身實現的
* tryAcquire(int unused)。
*/
public void lock() {
acquire(1);//<5>
}
public boolean tryLock() {
return tryAcquire(1);
}
/*
* 呼叫父類的release(int arg)時，會進而呼叫到Worker本身實現的
* tryRelease(int unused)。
*/
public void unlock() {
release(1);//<6>
}
public boolean isLocked() {
return isHeldExclusively();
}
/*
* 嘗試中斷worker的對應執行緒，如果執行緒已經啟動。建立
* worker時，state為-1，直到呼叫worker.thread.start()
* 後，worker的state為0，如果worker的state>=0，則嘗試
* 中斷執行緒。
*/
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
從上面的程式碼我們可以注意到，<1>、<2>、<3>、<4>、<5>、<6>處的方法並不是Worker本身有的方法，而是Worker繼承自父類AbstractQueuedSynchronizer的方法。那麼Worker為什麼需要繼承AbstractQueuedSynchronizer（AQS）？AQS又是何方神聖呢？
這裡先簡單介紹下AQS，它定義了若干介面交由程式設計師實現，諸如：lock()、unlock()、tryAcquire(int arg)、tryRelease(int arg)……等，以保證多個執行緒不會同時訪問同一資源。ThreadPoolExecutor中的欄位mainLock為可重入鎖ReentrantLock，某種程度上來說也是實現了AQS，ThreadPoolExecutor通過mainLock的lock()、unlock()以保證執行緒池內一些非執行緒安全的物件不會出現併發讀寫，如：workers、completedTaskCount……等。
public class ReentrantLock implements Lock, java.io.Serializable {
private final Sync sync;
abstract static class Sync extends AbstractQueuedSynchronizer {//...}
static final class NonfairSync extends Sync {//...}
public ReentrantLock() {
sync = new NonfairSync();
}
//...
}
那麼Worker繼承了AQS，Worker也實現了lock()、unlock()方法，說明Worker本身也存在多執行緒訪問的可能，那是什麼時候會出現多執行緒訪問Worker呢？這裡我們先按下這個問題，在介紹完Worker.run()之後你就會明白為何Worker要繼承AQS以保證執行緒訪問的順序性。
我們知道當呼叫Worker.thread.start()方法時，會進而呼叫Worker.run()方法，而Worker.run()方法會進而呼叫ThreadPoolExecutor.runWorker(Worker w)。下面，我們來看看runWorker(Worker w)的執行流程：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
/*
* 將worker.firstTask設定為null，因為worker可能會存在較長的一段時間，
* 而task可能很快執行完畢，避免worker長時間引用已完成task，以便GC回收
* 已完成task。
*/
w.firstTask = null;
/*
* 新建立的worker物件state為-1，呼叫worker.unlock()會進而呼叫
* worker.tryRelease(int unused)，將state設定為0，代表worker
* 對應工作執行緒已啟動，執行緒處於可中斷狀態。
*/
w.unlock(); // allow interrupts
/*
* 標誌worker是否異常完成，如果在<1>處task為空，且無法通過
* getTask()從任務佇列中獲取新的任務，則會跳出迴圈，並在<3>處
* 賦值為false，代表worker並沒有異常完成。
* 不管worker是正常完成還是異常完成，最後都會將completedAbruptly的結果
* 傳給processWorkerExit(...)，如果是worker是正常退出，則將workerCount-1，
* 並將worker從workers集合中移除。如果是異常退出，則不減少workerCount，僅僅
* 是將異常worker從workers集合中移除，並嘗試新增一個worker。
*/
boolean completedAbruptly = true;
try {
/*
* 如果task不為空，或者呼叫getTask()能任務佇列中獲取到新的任務，
* 則進入while塊的程式碼。如果任務佇列中沒有待執行的任務，呼叫getTask()
* 會讓當前執行緒陷入阻塞，直到超時或者有新任務進入任務佇列。
*/
while (task != null || (task = getTask()) != null) {//<1>
/*
* 如果能進入迴圈，代表worker準備開始執行任務，但在執行任務
* 前會先上鎖，等到任務執行結束又會在<2>處釋放鎖，然而執行緒池
* 又不會讓多個執行緒同時執行同一個任務，那麼為什麼在執行任務前
* 要讓worker先上鎖，執行完畢再釋放鎖呢？
* 我們假設有一個執行緒池有5個執行緒，其中A、B執行緒正在執行任務，
* C、D、E處於空閒狀態。所以我們能確定A、B兩個worker已經
* 獲得了鎖，而C、D、E還阻塞在getTask()方法中。現線上程池
* 執行shutdown()方法，該方法會進而呼叫interruptIdleWorkers()
* 中斷處於空閒狀態的工作執行緒。而在interruptIdleWorkers()方法中
* 判斷一個worker是否處於空閒，會呼叫worker.tryLock()，如果能
* 成功獲取到鎖，則代表該worker處於空閒狀態，則中斷該worker對應的
* 執行緒。因此，一個worker是有可能被多個執行緒訪問的，比如worker本身
* 對應的執行緒，又或者關閉執行緒池的執行緒。
*/
w.lock();
/*
* 如果執行緒池的執行狀態>=STOP，則中斷當前執行緒。如果執行狀態<STOP，
* 則確保執行緒沒有被中斷。
*/
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//空方法，在執行任務前執行
beforeExecute(wt, task);
try {
task.run();//開始執行任務
afterExecute(task, null);//空方法，在執行任務完畢後執行。
} catch (Throwable ex) {
afterExecute(task, ex);
throw ex;
}
} finally {
/*
* 設定當前任務為空，這樣就可以在下一次迴圈中獲取新的任務。
* 對worker執行的任務數+1，並釋放鎖。
*/
task = null;
w.completedTasks++;
w.unlock();//<2>
}
}
completedAbruptly = false;//<3>
} finally {
processWorkerExit(w, completedAbruptly);
}
}
在上面的runWorker(Worker w)中如果執行完worker的首要任務，或者首要任務為null，便會呼叫getTask()嘗試從任務佇列中獲取任務，但呼叫getTask()可能會使當前執行緒陷入等待或者阻塞直到有任務入隊。getTask()也有可能返回null導致當前worker對應的執行緒退出，有以下幾個原因可能導致工作執行緒退出：
- 如果呼叫setMaximumPoolSize(int maximumPoolSize)改小最大執行緒數，導致工作執行緒數大於maximumPoolSize。
- 執行緒池執行狀態為STOP。
- 執行緒池執行狀態為SHUTDOWN，且佇列為空。
- 執行緒等待任務超時後，如果執行緒池當前存在可回收的空閒執行緒（即allowCoreThreadTimeOut為true或者工作執行緒數大於核心執行緒數），如果佇列為空，則可直接退出，如果佇列不為空，工作執行緒數必須大於1，即執行緒池中最少兩個工作執行緒，如果只有一個執行緒還退出的話，就會存在佇列不為空，但執行緒池中沒有一個工作執行緒的尷尬情況。
private Runnable getTask() {
//超時標誌，預設為false，獲取任務如果超時則會在<5>賦值為true。
boolean timedOut = false;
for (; ; ) {
int c = ctl.get();
/*
* 如果執行緒池處於RUNNING狀態，則runStateAtLeast(c, SHUTDOWN)
* 為false，不會再判斷之後的邏輯為true或者false。
* 如果執行緒池處於SHUTDOWN狀態，且workQueue.isEmpty()為true，即
* 任務佇列為空，則直接返回。如果任務佇列不為空，則無法進入if分支，
* 依然要返回任務，按照SHUTDOWN的要求不再接受新任務，但仍要處理佇列
* 中的任務。
* 如果執行緒池處於STOP狀態，即便任務佇列不為空，也不再處理，則直接進入
* if分支後返回。
* 所以總結一下，只有兩種情況不會進入此分支：
* 1.執行緒池處於RUNNING狀態。
* 2.執行緒池處於SHUTDOWN狀態且任務佇列不為空。
*/
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {//<1>
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
/*
* timed決定是如果任務佇列沒有任務的話，是以無限期的方式
* 等待任務入隊，或者一旦等待時間超過keepAliveTime，則
* 返回null。
* 如果allowCoreThreadTimeOut為true，則核心執行緒等待
* 任務時間超過keepAliveTime後會被回收。
* 如果當前工作執行緒數量workerCount大於核心執行緒數，也會在
* 執行緒等待任務超過keepAliveTime後回收執行緒。
*/
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
/*
* 我們先看進入此分支後會做的事，再分析如何進入這個分支。進入
* 此分支後，會用CAS減少workerCount的數量，成功則返回null。
* 否則continue重新開始新一輪的for迴圈。
* 現在，我們來分析下進入此分支的邏輯：
* 首先是wc > maximumPoolSize,一般workerCount不會大於
* maximumPoolSize，除非執行緒池執行期間通過
* setMaximumPoolSize(int maximumPoolSize)將執行緒池
* 的最大執行緒數改小。
* (timed && timedOut)在第一輪for迴圈永遠為false，因為
* timedOut要為true的條件，首先是timed為true，即執行緒池記憶體在空閒後
* 可回收的執行緒，不管是執行緒池允許回收核心執行緒，或者執行緒數大於核心執行緒數。
* 只有在<4>獲取任務超時後workQueue返回null，才有可能到達<5>處將
* timedOut賦值為true，並且開始新一輪的迴圈。
* 之後的兩個判斷wc>1和workQueue.isEmpty()，判斷佇列為空還好理解，
* 為什麼要判斷wc>1?首先我們要知道maximumPoolSize必須大於等於1，當我們
* 往執行緒池傳入的maximumPoolSize<=0會丟擲異常。其次，如果我們將<2>處改為：
* (wc >= 1 || workQueue.isEmpty())，有可能出現執行緒池內只有一個執行緒，
* 但任務佇列不為null。依舊會進入此分支內部執行<3>的程式碼以CAS的方式對
* workerCount-1，從而出現一個尷尬的情況，任務佇列中有任務，但工作執行緒數
* 為0。所以<2>處必須保證wc>1。
* 思考一種情況：假設一個執行緒池核心執行緒數為3，最大執行緒數為5，
* allowCoreThreadTimeOut為false，執行緒池當前工作執行緒數量也為5，5個執行緒
* 同時完成任務，並執行getTask()獲取任務，可想而知timed為true，因為工作
* 執行緒數（5）大於核心執行緒數（3），5個工作執行緒都是呼叫<4>處workQueue.poll(...)
* 等待任務，超時則返回null。如果超時時間到達，3個核心執行緒如何重新進入等待狀態，
* 剩餘2個執行緒如何被回收？
* 當5個執行緒超時返回後，會將timedOut賦值為true，然後重新開始新一輪的for迴圈，一直
* 執行到此分支，此時(timed && timedOut)都為true，佇列也都為null，所以5個執行緒會
* 進入此分支。用CAS成功對workerCount-1的執行緒將被回收，失敗的執行緒則continue又開始
* 新一輪的for迴圈，直到wc<=corePoolSize，timed為false，最後剩餘的工作執行緒數呼叫
* workQueue.take()無限期地等待任務的到來，除非執行緒被中斷。
*/
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {//<2>
if (compareAndDecrementWorkerCount(c))//<3>
return null;
continue;
}
try {
/*
* workQueue.poll(...)和workQueue.take()都有可能使當前執行緒陷入
* 等待，直到返回任務，只不過前者相比後者多了一個超時時間，到達超時時間
* 如果有任務入隊，則r不為null，直接返回任務。
* 執行緒等待期間，如果執行緒被中斷，則會丟擲InterruptedException異常。
* 一般關閉執行緒池時，會嘗試中斷空閒執行緒，而處於等待任務的空閒執行緒會跳到<6>處，
* 重新開始新一輪的for迴圈，並且在<1>處判斷執行緒池處於SHUTDOWN或者STOP，
* 從而退出。
*/
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) ://<4>
workQueue.take();
if (r != null)
return r;
/*
* 只有等待任務超時控制流才會執行到這裡，將超時標誌賦值為true，
* 重新開始新一輪的for迴圈。
*/
timedOut = true;//<5>
} catch (InterruptedException retry) {
timedOut = false;//<6>
}
}
}
在runWorker(Worker w)中有兩種方式可以進到下面的處理執行緒退出processWorkerExit(Worker w, boolean completedAbruptly)方法：
- 工作執行緒執行任務遇到異常，從而跳出while迴圈。
- 工作執行緒獲取任務超時等待，正常退出迴圈。
上面兩種方式傳遞給processWorkerExit(...)的completedAbruptly是不同的，第一個方式傳入的completedAbruptly為true，第二個方式為false，雖然worker是同一個。那麼當completedAbruptly為true或者false，processWorkerExit(...)的流程又是怎麼走的呢？
private void processWorkerExit(Worker w, boolean completedAbruptly) {
/*
* 從runWorker(Worker w)傳遞而來的變數，標誌worker是否意外完成，
* 當worker執行任務時丟擲異常，該變數為true。如果是意外完成，則表明
* workerCount尚未-1。如果worker獲取任務超時從而要讓執行緒被回收，在
* getTask()方法中會對workerCount-1。
*/
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
/*
* 獲取可重入鎖後，將worker已完成的任務數加到執行緒池已完成任務數，
* 並將worker從workers集合中移除。
*/
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
//嘗試終止執行緒。
tryTerminate();
int c = ctl.get();
//如果執行緒池執行狀態處於RUNNING或SHUTDOWN，則進入此分支。
if (runStateLessThan(c, STOP)) {
//如果執行緒是正常退出，則進入此分支
if (!completedAbruptly) {
/*
* min為執行緒池建立核心執行緒後，允許最小的核心執行緒數，如果
* allowCoreThreadTimeOut為true則代表核心執行緒可以被回收，
* 則min為0，否則min為核心執行緒數量。
* 如果執行緒池允許回收執行緒，且佇列不為空，則判斷在移除當前worker
* 後，執行緒池工作執行緒的數量是否還大於等於1，避免出現佇列裡有任務
* 但沒有執行緒執行的情況，如果工作執行緒數大於等於1，則退出執行緒執行緒。
* 否則呼叫addWorker(Runnable firstTask, boolean core)
* 嘗試增加新的執行緒。
*/
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && !workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}
最後，我們還需要介紹關閉執行緒池之後做的操作。關閉執行緒池會修改執行緒池執行狀態，在advanceRunState(int targetState)會使用CAS自旋的方式，將執行緒池狀態修改為SHUTDOWN。之後呼叫interruptIdleWorkers()中斷空閒執行緒，這裡我們看到中斷的時候呼叫worker的tryLock()和unlock()。Worker之所以繼承AQS就是為了方便區分哪些worker正在執行任務，哪些worker處於空閒中，以便在關閉執行緒池時中斷所有空閒的worker。
/**
* 呼叫此方法後不再接受新任務，但會執行現有佇列任務。
* 如果呼叫此方法前該方法已被呼叫，則不會有任何效果。
* 該方法不會等待所有任務完成，需要呼叫：
* awaitTermination(long timeout, TimeUnit unit)
*/
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN);//設定執行緒池狀態為SHUTDOWN
interruptIdleWorkers();//中斷空閒執行緒
onShutdown(); //鉤子方法，按使用者需要實現。
} finally {
mainLock.unlock();
}
/*
* 嘗試終止執行緒池，可能執行緒池有任務在執行，當前執行緒終止失敗。
* 但隨著工作執行緒逐個退出，最後一個工作執行緒將成功終止執行緒池。
*/
tryTerminate();
}
private void advanceRunState(int targetState) {
for (; ; ) {
int c = ctl.get();
/*
* 如果執行緒池執行狀態>=targetState，則
* runStateAtLeast(c, targetState)為true直接退出。
* 否則呼叫ctlOf(int rs, int wc)，根據執行狀態和工作
* 執行緒數生成的值以CAS自旋的方式set進ctl。
*/
if (runStateAtLeast(c, targetState) ||
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
break;
}
}
//呼叫shutdown()會進而呼叫此方法，中斷空閒執行緒。
private void interruptIdleWorkers() {
interruptIdleWorkers(false);
}
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
/*
* 如果執行緒未被中斷，則嘗試獲取worker的鎖，如果能
* 成功獲取，代表worker執行緒處於空閒中。
*/
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
//如果onlyOne為true代表最多隻中斷一個執行緒
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
在中斷空閒執行緒後shutdown()還會呼叫tryTerminate()，如果檢視tryTerminate()的引用，可以發現不單單shutdown()有呼叫，像addWorkerFailed(Worker w)、processWorkerExit(...)……都有呼叫。這個方法旨在終止執行緒池，如果當前有執行緒關閉了執行緒池，執行緒池如果沒有存活執行緒，或者執行緒都處於空閒狀態，自然而然執行嘗試終止執行緒池方法；如果關閉執行緒池時，執行緒池仍然有執行緒處於執行任務狀態，無法終止執行緒池，就要靠這些工作執行緒在退出時終止執行緒池。
final void tryTerminate() {
for (; ; ) {
int c = ctl.get();
/*
* 如果執行緒池執行狀態處於RUNNING、TIDYING則退出，
* 或者執行狀態小於STOP（即處於RUNNING、SHUTDOWN）
* 且佇列不為空，則退出。
* 總結一下，只有當執行狀態處於STOP時或者狀態處於SHUTDOWN
* 但佇列為空，才不會進此分支。
*/
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||//<1>
(runStateLessThan(c, STOP) && !workQueue.isEmpty()))
return;
//如果工作執行緒數不為0，則嘗試最多中斷一個空閒執行緒後退出。
if (workerCountOf(c) != 0) {
interruptIdleWorkers(ONLY_ONE);
return;
}
/*
* 在processWorkerExit(...)方法中如果worker是
* 異常退出會對workerCount-1，如果是正常退出，則
* workerCount在getTask()中-1。之後在processWorkerExit(...)
* 中移除worker。
* 不論worker是正常退出還是異常退出，終歸workerCount會慢慢回到0。
* 而processWorkerExit(...)中在對workerCount-1後，還會呼叫
* tryTerminate()。因此一個被關閉的執行緒池，它的最後一個執行緒，會
* 執行到此處，處理終止執行緒池的工作。
*/
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
/*
* 用CAS設定執行緒池執行狀態為TIDYING，可能存在多個執行緒同時
* 呼叫shutdown()後並依次執行到這一步（因為要獲得可重入鎖），
* 但只有一個執行緒可以CAS成功，其他執行緒CAS失敗後返回<1>處後
* 判斷執行狀態>=TIDYING則退出。
*/
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
//空函式，具體由使用者實現，主要用於終止執行緒池後的操作。
terminated();
} finally {
//最後設定執行緒池的狀態為TERMINATED
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
}
}