Java核心（二）深入理解執行緒池ThreadPool

摘要： 本文你將獲得以下資訊： 執行緒池原始碼解讀 執行緒池執行流程分析 帶返回值的執行緒池實現 延遲執行緒池實現 為了方便讀者理解，本文會由淺入深，先從執行緒池的使用開始再延伸到原始碼解讀和原始碼分析等高階內容，讀者可根據自己的情況自主選擇...

本文你將獲得以下資訊：

• 執行緒池原始碼解讀
• 執行緒池執行流程分析
• 帶返回值的執行緒池實現
• 延遲執行緒池實現

為了方便讀者理解，本文會由淺入深，先從執行緒池的使用開始再延伸到原始碼解讀和原始碼分析等高階內容，讀者可根據自己的情況自主選擇閱讀順序和需要了解的章節。

一、執行緒池優點

執行緒池能夠更加充分的利用CPU、記憶體、網路、IO等系統資源，執行緒池的主要作用如下：

• 利用執行緒池可以複用執行緒，控制最大併發數；
• 實現任務快取策略和拒絕機制；
• 實現延遲執行

阿里巴巴Java開發手冊強制規定：執行緒資源必須通過執行緒池提供，如下圖：

二、執行緒池使用

本節會介紹7種執行緒池的建立與使用，執行緒池的狀態介紹，ThreadPoolExecutor引數介紹等。

2.1 執行緒池建立

執行緒池可以使用Executors和ThreadPoolExecutor，其中使用Executors有六種建立執行緒池的方法，如下圖：

// 使用Executors方式建立
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(2);
ScheduledExecutorService singleThreadScheduledExecutor = Executors.newSingleThreadScheduledExecutor();
ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(2);
ExecutorService workStealingPool = Executors.newWorkStealingPool();
// 原始建立方式
ThreadPoolExecutor tp = new ThreadPoolExecutor(10, 10, 10L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
複製程式碼

2.1.1 執行緒池解讀

1. newSingleThreadExecutor()，它的特點在於工作執行緒數目被限制為 1，操作一個無界的工作佇列，所以它保證了所有任務的都是被順序執行，最多會有一個任務處於活動狀態，並且不允許使用者改動執行緒池例項，因此可以避免其改變執行緒數目。
2. newCachedThreadPool()，它是一種用來處理大量短時間工作任務的執行緒池，具有幾個鮮明特點：它會試圖快取執行緒並重用，當無快取執行緒可用時，就會建立新的工作執行緒；如果執行緒閒置的時間超過 60 秒，則被終止並移出快取；長時間閒置時，這種執行緒池，不會消耗什麼資源。其內部使用 SynchronousQueue 作為工作佇列。
3. newFixedThreadPool(int nThreads)，重用指定數目（nThreads）的執行緒，其背後使用的是無界的工作佇列，任何時候最多有 nThreads 個工作執行緒是活動的。這意味著，如果任務數量超過了活動佇列數目，將在工作佇列中等待空閒執行緒出現；如果有工作執行緒退出，將會有新的工作執行緒被建立，以補足指定的數目 nThreads。
4. newSingleThreadScheduledExecutor() 建立單執行緒池，返回 ScheduledExecutorService，可以進行定時或週期性的工作排程。
5. newScheduledThreadPool(int corePoolSize)和newSingleThreadScheduledExecutor()類似，建立的是個 ScheduledExecutorService，可以進行定時或週期性的工作排程，區別在於單一工作執行緒還是多個工作執行緒。
6. newWorkStealingPool(int parallelism)，這是一個經常被人忽略的執行緒池，Java 8 才加入這個建立方法，其內部會構建ForkJoinPool，利用Work-Stealing演算法，並行地處理任務，不保證處理順序。
7. ThreadPoolExecutor是最原始的執行緒池建立，上面1-3建立方式都是對ThreadPoolExecutor的封裝。

總結：其中newSingleThreadExecutor、newCachedThreadPool、newFixedThreadPool是對ThreadPoolExecutor的封裝實現，newSingleThreadScheduledExecutor、newScheduledThreadPool則為ThreadPoolExecutor子類ScheduledThreadPoolExecutor的封裝，用於執行延遲任務，newWorkStealingPool則為Java 8新加的方法。

2.1.2 單執行緒池的意義

從以上程式碼可以看出newSingleThreadExecutor和newSingleThreadScheduledExecutor建立的都是單執行緒池，那麼單執行緒池的意義是什麼呢？

雖然是單執行緒池，但提供了工作佇列，生命週期管理，工作執行緒維護等功能。

2.2 ThreadPoolExecutor解讀

ThreadPoolExecutor作為執行緒池的核心方法，我們來看一下ThreadPoolExecutor內部實現，以及封裝類是怎麼呼叫ThreadPoolExecutor的。

先從建構函式說起，建構函式原始碼如下：

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;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
複製程式碼

引數說明：

• corePoolSize：所謂的核心執行緒數，可以大致理解為長期駐留的執行緒數目（除非設定了 allowCoreThreadTimeOut）。對於不同的執行緒池，這個值可能會有很大區別，比如 newFixedThreadPool 會將其設定為 nThreads，而對於 newCachedThreadPool 則是為 0。
• maximumPoolSize：顧名思義，就是執行緒不夠時能夠建立的最大執行緒數。同樣進行對比，對於 newFixedThreadPool，當然就是 nThreads，因為其要求是固定大小，而 newCachedThreadPool 則是 Integer.MAX_VALUE。
• keepAliveTime：空閒執行緒的保活時間，如果執行緒的空閒時間超過這個值，那麼將會被關閉。注意此值生效條件必須滿足：空閒時間超過這個值，並且執行緒池中的執行緒數少於等於核心執行緒數corePoolSize。當然核心執行緒預設是不會關閉的，除非設定了allowCoreThreadTimeOut(true)那麼核心執行緒也可以被回收。
• TimeUnit：時間單位。
• BlockingQueue：任務丟列，用於儲存執行緒池的待執行任務的。
• threadFactory：用於生成執行緒，一般我們可以用預設的就可以了。
• handler：當執行緒池已經滿了，但是又有新的任務提交的時候，該採取什麼策略由這個來指定。有幾種方式可供選擇，像丟擲異常、直接拒絕然後返回等，也可以自己實現相應的介面實現自己的邏輯。

來看一下執行緒池封裝類對於ThreadPoolExecutor的呼叫：

newSingleThreadExecutor對ThreadPoolExecutor的封裝原始碼如下：

public static ExecutorService newSingleThreadExecutor() {
return new Executors.FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
複製程式碼

newCachedThreadPool對ThreadPoolExecutor的封裝原始碼如下：

public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
複製程式碼

newFixedThreadPool對ThreadPoolExecutor的封裝原始碼如下：

public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
複製程式碼

ScheduledExecutorService對ThreadPoolExecutor的封裝原始碼如下：

public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
return new DelegatedScheduledExecutorService
(new ScheduledThreadPoolExecutor(1));
}
複製程式碼

newSingleThreadScheduledExecutor使用的是ThreadPoolExecutor的子類ScheduledThreadPoolExecutor，如下圖所示：

newScheduledThreadPool對ThreadPoolExecutor的封裝原始碼如下：

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
複製程式碼

newScheduledThreadPool使用的也是ThreadPoolExecutor的子類ScheduledThreadPoolExecutor。

2.3 執行緒池狀態

檢視ThreadPoolExecutor原始碼可知執行緒的狀態如下：

執行緒狀態解讀（以下內容來源於： ofollow,noindex">javadoop.com/post/java-t…

• RUNNING：這個沒什麼好說的，這是最正常的狀態：接受新的任務，處理等待佇列中的任務；
• SHUTDOWN：不接受新的任務提交，但是會繼續處理等待佇列中的任務；
• STOP：不接受新的任務提交，不再處理等待佇列中的任務，中斷正在執行任務的執行緒；
• TIDYING：所有的任務都銷燬了，workCount 為 0。執行緒池的狀態在轉換為 TIDYING 狀態時，會執行鉤子方法 terminated()；
• TERMINATED：terminated() 方法結束後，執行緒池的狀態就會變成這個；

RUNNING 定義為 -1，SHUTDOWN 定義為 0，其他的都比 0 大，所以等於 0 的時候不能提交任務，大於 0 的話，連正在執行的任務也需要中斷。

看了這幾種狀態的介紹，讀者大體也可以猜到十之八九的狀態轉換了，各個狀態的轉換過程有以下幾種：

• RUNNING -> SHUTDOWN：當呼叫了 shutdown() 後，會發生這個狀態轉換，這也是最重要的；
• (RUNNING or SHUTDOWN) -> STOP：當呼叫 shutdownNow() 後，會發生這個狀態轉換，這下要清楚 shutDown() 和 shutDownNow() 的區別了；
• SHUTDOWN -> TIDYING：當任務佇列和執行緒池都清空後，會由 SHUTDOWN 轉換為 TIDYING；
• STOP -> TIDYING：當任務佇列清空後，發生這個轉換；
• TIDYING -> TERMINATED：這個前面說了，當 terminated() 方法結束後；

2.4 執行緒池執行

說了那麼多下來一起來看執行緒池的是怎麼執行任務的，執行緒池任務提交有兩個方法：

• execute
• submit

其中execute只能接受Runnable型別的任務，使用如下：

ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
singleThreadExecutor.execute(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName());
}
});
複製程式碼

submit可以接受Runnable或Callable型別的任務，使用如下：

ExecutorService executorService = Executors.newSingleThreadExecutor();
executorService.submit(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName());
}
});
複製程式碼

2.4.1 帶返回值的執行緒池實現

使用submit傳遞Callable類可以獲取執行任務的返回值，Callable是JDK 1.5 新增的特性用於補充Runnable無返回的情況。

ExecutorService executorService = Executors.newSingleThreadExecutor();
Future<Long> result = executorService.submit(new Callable<Long>() {
@Override
public Long call() throws Exception {
return new Date().getTime();
}
});
try {
System.out.println("執行結果：" + result.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
複製程式碼

2.4.2 延遲執行緒池實現

線上程池中newSingleThreadScheduledExecutor和newScheduledThreadPool返回的是ScheduledExecutorService，用於執行延遲執行緒池的，程式碼如下：

// 延遲執行緒池
ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(2);
scheduledThreadPool.schedule(new Runnable() {
@Override
public void run() {
System.out.println("time:" + new Date().getTime());
}
}, 10, TimeUnit.SECONDS);
複製程式碼

完整示例下載地址： github.com/vipstone/ja…

三、執行緒池原始碼解讀

閱讀執行緒池的原始碼有一個小技巧，可以按照執行緒池執行的順序進行串連關聯閱讀，這樣更容易理解執行緒池的實現。

原始碼閱讀流程解讀

我們先從執行緒池的任務提交方法execute()開始閱讀，從execute()我們會發現執行緒池執行的核心方法是addWorker()，在addWorker()中我們發現啟動執行緒呼叫了start()方法，呼叫start()方法之後會執行Worker類的run()方法，run裡面呼叫runWorker()，執行程式的關鍵在於getTask()方法，getTask()方法之後就是此執行緒的關閉，整個執行緒池的工作流程也就完成了，下來一起來看吧（如果本段文章沒看懂的話也可以看完原始碼之後，回過頭來再看一遍）。

3.1 execute() 原始碼解讀

public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();

// 如果當前執行緒數少於核心執行緒數，那麼直接新增一個 worker 來執行任務，
// 建立一個新的執行緒，並把當前任務 command 作為這個執行緒的第一個任務(firstTask)
if (workerCountOf(c) < corePoolSize) {
// 新增任務成功，那麼就結束了。提交任務嘛，執行緒池已經接受了這個任務，這個方法也就可以返回了
// 至於執行的結果，到時候會包裝到 FutureTask 中。
// 返回 false 代表執行緒池不允許提交任務
if (addWorker(command, true))
return;
c = ctl.get();
}
// 到這裡說明，要麼當前執行緒數大於等於核心執行緒數，要麼剛剛 addWorker 失敗了

// 如果執行緒池處於 RUNNING 狀態，把這個任務新增到任務佇列 workQueue 中
if (isRunning(c) && workQueue.offer(command)) {
/* 這裡面說的是，如果任務進入了 workQueue，我們是否需要開啟新的執行緒
* 因為執行緒數在 [0, corePoolSize) 是無條件開啟新的執行緒
* 如果執行緒數已經大於等於 corePoolSize，那麼將任務新增到佇列中，然後進到這裡
*/
int recheck = ctl.get();
// 如果執行緒池已不處於 RUNNING 狀態，那麼移除已經入隊的這個任務，並且執行拒絕策略
if (! isRunning(recheck) && remove(command))
reject(command);
// 如果執行緒池還是 RUNNING 的，並且執行緒數為 0，那麼開啟新的執行緒
// 到這裡，我們知道了，這塊程式碼的真正意圖是：擔心任務提交到佇列中了，但是執行緒都關閉了
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 如果 workQueue 佇列滿了，那麼進入到這個分支
// 以 maximumPoolSize 為界建立新的 worker，
// 如果失敗，說明當前執行緒數已經達到 maximumPoolSize，執行拒絕策略
else if (!addWorker(command, false))
reject(command);
}
複製程式碼

3.2 addWorker() 原始碼解讀

// 第一個引數是準備提交給這個執行緒執行的任務，之前說了，可以為 null
// 第二個引數為 true 代表使用核心執行緒數 corePoolSize 作為建立執行緒的界線，也就說建立這個執行緒的時候，
//如果執行緒池中的執行緒總數已經達到 corePoolSize，那麼不能響應這次建立執行緒的請求
//如果是 false，代表使用最大執行緒數 maximumPoolSize 作為界線
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);

// 這個非常不好理解
// 如果執行緒池已關閉，並滿足以下條件之一，那麼不建立新的 worker：
// 1. 執行緒池狀態大於 SHUTDOWN，其實也就是 STOP, TIDYING, 或 TERMINATED
// 2. firstTask != null
// 3. workQueue.isEmpty()
// 簡單分析下：
// 還是狀態控制的問題，當執行緒池處於 SHUTDOWN 的時候，不允許提交任務，但是已有的任務繼續執行
// 當狀態大於 SHUTDOWN 時，不允許提交任務，且中斷正在執行的任務
// 多說一句：如果執行緒池處於 SHUTDOWN，但是 firstTask 為 null，且 workQueue 非空，那麼是允許建立 worker 的
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;

for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 如果成功，那麼就是所有建立執行緒前的條件校驗都滿足了，準備建立執行緒執行任務了
// 這裡失敗的話，說明有其他執行緒也在嘗試往執行緒池中建立執行緒
if (compareAndIncrementWorkerCount(c))
break retry;
// 由於有併發，重新再讀取一下 ctl
c = ctl.get();
// 正常如果是 CAS 失敗的話，進到下一個裡層的for迴圈就可以了
// 可是如果是因為其他執行緒的操作，導致執行緒池的狀態發生了變更，如有其他執行緒關閉了這個執行緒池
// 那麼需要回到外層的for迴圈
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}

/* 
* 到這裡，我們認為在當前這個時刻，可以開始建立執行緒來執行任務了，
* 因為該校驗的都校驗了，至於以後會發生什麼，那是以後的事，至少當前是滿足條件的
*/

// worker 是否已經啟動
boolean workerStarted = false;
// 是否已將這個 worker 新增到 workers 這個 HashSet 中
boolean workerAdded = false;
Worker w = null;
try {
final ReentrantLock mainLock = this.mainLock;
// 把 firstTask 傳給 worker 的構造方法
w = new Worker(firstTask);
// 取 worker 中的執行緒物件，之前說了，Worker的構造方法會呼叫 ThreadFactory 來建立一個新的執行緒
final Thread t = w.thread;
if (t != null) {
// 這個是整個類的全域性鎖，持有這個鎖才能讓下面的操作“順理成章”，
// 因為關閉一個執行緒池需要這個鎖，至少我持有鎖的期間，執行緒池不會被關閉
mainLock.lock();
try {

int c = ctl.get();
int rs = runStateOf(c);

// 小於 SHUTTDOWN 那就是 RUNNING，這個自不必說，是最正常的情況
// 如果等於 SHUTDOWN，前面說了，不接受新的任務，但是會繼續執行等待佇列中的任務
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
// worker 裡面的 thread 可不能是已經啟動的
if (t.isAlive())
throw new IllegalThreadStateException();
// 加到 workers 這個 HashSet 中
workers.add(w);
int s = workers.size();
// largestPoolSize 用於記錄 workers 中的個數的最大值
// 因為 workers 是不斷增加減少的，通過這個值可以知道執行緒池的大小曾經達到的最大值
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 新增成功的話，啟動這個執行緒
if (workerAdded) {
// 啟動執行緒
t.start();
workerStarted = true;
}
}
} finally {
// 如果執行緒沒有啟動，需要做一些清理工作，如前面 workCount 加了 1，將其減掉
if (! workerStarted)
addWorkerFailed(w);
}
// 返回執行緒是否啟動成功
return workerStarted;
}
複製程式碼

在這段程式碼可以看出，呼叫了t.start();

3.3 runWorker() 原始碼解讀

根據上面程式碼可知，呼叫了Worker的t.start()之後，緊接著會呼叫Worker的run()方法，run()原始碼如下：

public void run() {
runWorker(this);
}
複製程式碼

runWorker()原始碼如下：

//worker 執行緒啟動後呼叫,while 迴圈(即自旋!)不斷從等待佇列獲取任務並執行
//worker 初始化時，可指定 firstTask，那麼第一個任務也就可以不需要從佇列中獲取
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
// 該執行緒的第一個任務(若有)
Runnable task = w.firstTask;
w.firstTask = null;
// 允許中斷
w.unlock(); 
boolean completedAbruptly = true;
try {
// 迴圈呼叫 getTask 獲取任務
while (task != null || (task = getTask()) != null) {
w.lock();
// 若執行緒池狀態大於等於 STOP，那麼意味著該執行緒也要中斷
/**
* 若執行緒池STOP，請確保執行緒 已被中斷
* 如果沒有，請確保執行緒未被中斷
* 這需要在第二種情況下進行重新檢查，以便在關中斷時處理shutdownNow競爭
*/
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
// 這是一個鉤子方法，留給需要的子類實現
beforeExecute(wt, task);
Throwable thrown = null;
try {
// 到這裡終於可以執行任務了
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
// 這裡不允許丟擲 Throwable，所以轉換為 Error
thrown = x; throw new Error(x);
} finally {
// 也是一個鉤子方法，將 task 和異常作為引數，留給需要的子類實現
afterExecute(task, thrown);
}
} finally {
// 置空 task，準備 getTask 下一個任務
task = null;
// 累加完成的任務數
w.completedTasks++;
// 釋放掉 worker 的獨佔鎖
w.unlock();
}
}
completedAbruptly = false;
} finally {
// 到這裡，需要執行執行緒關閉
// 1. 說明 getTask 返回 null，也就是說，這個 worker 的使命結束了，執行關閉
// 2. 任務執行過程中發生了異常
//第一種情況，已經在程式碼處理了將 workCount 減 1，這個在 getTask 方法分析中說
//第二種情況，workCount 沒有進行處理，所以需要在 processWorkerExit 中處理
processWorkerExit(w, completedAbruptly);
}
}
複製程式碼

3.4 getTask() 原始碼解讀

runWorker裡面的有getTask()，來看下具體的實現：

// 此方法有三種可能
// 1. 阻塞直到獲取到任務返回。預設 corePoolSize 之內的執行緒是不會被回收的，它們會一直等待任務
// 2. 超時退出。keepAliveTime 起作用的時候，也就是如果這麼多時間內都沒有任務，那麼應該執行關閉
// 3. 如果發生了以下條件，須返回 null
//池中有大於 maximumPoolSize 個 workers 存在(通過呼叫 setMaximumPoolSize 進行設定)
//執行緒池處於 SHUTDOWN，而且 workQueue 是空的，前面說了，這種不再接受新的任務
//執行緒池處於 STOP，不僅不接受新的執行緒，連 workQueue 中的執行緒也不再執行
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
// 允許核心執行緒數內的執行緒回收，或當前執行緒數超過了核心執行緒數，那麼有可能發生超時關閉
// 這裡 break，是為了不往下執行後一個 if (compareAndDecrementWorkerCount(c))
// 兩個 if 一起看：如果當前執行緒數 wc > maximumPoolSize，或者超時，都返回 null
// 那這裡的問題來了，wc > maximumPoolSize 的情況，為什麼要返回 null？
// 換句話說，返回 null 意味著關閉執行緒。
// 那是因為有可能開發者呼叫了 setMaximumPoolSize 將執行緒池的 maximumPoolSize 調小了
// 如果此 worker 發生了中斷，採取的方案是重試
// 解釋下為什麼會發生中斷，這個讀者要去看 setMaximumPoolSize 方法，
// 如果開發者將 maximumPoolSize 調小了，導致其小於當前的 workers 數量，
// 那麼意味著超出的部分執行緒要被關閉。重新進入 for 迴圈，自然會有部分執行緒會返回 null
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
// CAS 操作，減少工作執行緒數
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
// 如果此 worker 發生了中斷，採取的方案是重試
// 解釋下為什麼會發生中斷，這個讀者要去看 setMaximumPoolSize 方法，
// 如果開發者將 maximumPoolSize 調小了，導致其小於當前的 workers 數量，
// 那麼意味著超出的部分執行緒要被關閉。重新進入 for 迴圈，自然會有部分執行緒會返回 null
timedOut = false;
}
}
}
複製程式碼

四、執行緒池執行流程

執行緒池的執行流程如下圖：

五、總結

本文總結以問答的形式展示，引自《深度解讀 java 執行緒池設計思想及原始碼實現》，最下方附參考地址。

1、執行緒池有哪些關鍵屬性？

• corePoolSize 到 maximumPoolSize 之間的執行緒會被回收，當然 corePoolSize 的執行緒也可以通過設定而得到回收（allowCoreThreadTimeOut(true)）。

• workQueue 用於存放任務，新增任務的時候，如果當前執行緒數超過了 corePoolSize，那麼往該佇列中插入任務，執行緒池中的執行緒會負責到佇列中拉取任務。

• keepAliveTime 用於設定空閒時間，如果執行緒數超出了 corePoolSize，並且有些執行緒的空閒時間超過了這個值，會執行關閉這些執行緒的操作

• rejectedExecutionHandler 用於處理當執行緒池不能執行此任務時的情況，預設有丟擲 RejectedExecutionException 異常、忽略任務、使用提交任務的執行緒來執行此任務和將佇列中等待最久的任務刪除，然後提交此任務這四種策略，預設為丟擲異常。

2、執行緒池中的執行緒建立時機？

• 如果當前執行緒數少於 corePoolSize，那麼提交任務的時候建立一個新的執行緒，並由這個執行緒執行這個任務；

• 如果當前執行緒數已經達到 corePoolSize，那麼將提交的任務新增到佇列中，等待執行緒池中的執行緒去佇列中取任務；

• 如果佇列已滿，那麼建立新的執行緒來執行任務，需要保證池中的執行緒數不會超過 maximumPoolSize，如果此時執行緒數超過了 maximumPoolSize，那麼執行拒絕策略。

3、任務執行過程中發生異常怎麼處理？

如果某個任務執行出現異常，那麼執行任務的執行緒會被關閉，而不是繼續接收其他任務。然後會啟動一個新的執行緒來代替它。

4、什麼時候會執行拒絕策略？

• workers 的數量達到了 corePoolSize，任務入隊成功，以此同時執行緒池被關閉了，而且關閉執行緒池並沒有將這個任務出隊，那麼執行拒絕策略。這裡說的是非常邊界的問題，入隊和關閉執行緒池併發執行，讀者仔細看看 execute 方法是怎麼進到第一個 reject(command) 裡面的。
• workers 的數量大於等於 corePoolSize，準備入隊，可是佇列滿了，任務入隊失敗，那麼準備開啟新的執行緒，可是執行緒數已經達到 maximumPoolSize，那麼執行拒絕策略。