java高并發(fā)ThreadPoolExecutor類解析線程池執(zhí)行流程
摘要
ThreadPoolExecutor是Java線程池中最核心的類之一，它能夠保證線程池按照正常的業(yè)務(wù)邏輯執(zhí)行任務(wù)，并通過原子方式更新線程池每個階段的狀態(tài)。
今天，我們通過ThreadPoolExecutor類的源碼深度解析線程池執(zhí)行任務(wù)的核心流程，小伙伴們最好是打開IDEA，按照冰河說的步驟，調(diào)試下ThreadPoolExecutor類的源碼，這樣會理解的更加深刻，好了，開始今天的主題。
核心邏輯概述
ThreadPoolExecutor是Java線程池中最核心的類之一，它能夠保證線程池按照正常的業(yè)務(wù)邏輯執(zhí)行任務(wù)，并通過原子方式更新線程池每個階段的狀態(tài)。
ThreadPoolExecutor類中存在一個workers工作線程集合，用戶可以向線程池中添加需要執(zhí)行的任務(wù)，workers集合中的工作線程可以直接執(zhí)行任務(wù)，或者從任務(wù)隊(duì)列中獲取任務(wù)后執(zhí)行。ThreadPoolExecutor類中提供了整個線程池從創(chuàng)建到執(zhí)行任務(wù)，再到消亡的整個流程方法。本文，就結(jié)合ThreadPoolExecutor類的源碼深度分析線程池執(zhí)行任務(wù)的整體流程。
在ThreadPoolExecutor類中，線程池的邏輯主要體現(xiàn)在execute(Runnable)方法，addWorker(Runnable, boolean)方法，addWorkerFailed(Worker)方法和拒絕策略上，接下來，我們就深入分析這幾個核心方法。
execute(Runnable)方法
execute(Runnable)方法的作用是提交Runnable類型的任務(wù)到線程池中。我們先看下execute(Runnable)方法的源碼，如下所示。
public void execute(Runnable command) {
//如果提交的任務(wù)為空，則拋出空指針異常
if (command == null)
throw new NullPointerException();
//獲取線程池的狀態(tài)和線程池中線程的數(shù)量
int c = ctl.get();
//線程池中的線程數(shù)量小于corePoolSize的值
if (workerCountOf(c) < corePoolSize) {
//重新開啟線程執(zhí)行任務(wù)
if (addWorker(command, true))
return;
c = ctl.get();
}
//如果線程池處于RUNNING狀態(tài)，則將任務(wù)添加到阻塞隊(duì)列中
if (isRunning(c) && workQueue.offer(command)) {
//再次獲取線程池的狀態(tài)和線程池中線程的數(shù)量，用于二次檢查
int recheck = ctl.get();
//如果線程池沒有未處于RUNNING狀態(tài)，從隊(duì)列中刪除任務(wù)
if (! isRunning(recheck) && remove(command))
//執(zhí)行拒絕策略
reject(command);
//如果線程池為空，則向線程池中添加一個線程
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//任務(wù)隊(duì)列已滿，則新增worker線程，如果新增線程失敗，則執(zhí)行拒絕策略
else if (!addWorker(command, false))
reject(command);
}
整個任務(wù)的執(zhí)行流程，我們可以簡化成下圖所示。
接下來，我們拆解execute(Runnable)方法，具體分析execute(Runnable)方法的執(zhí)行邏輯。
（1）線程池中的線程數(shù)是否小于corePoolSize核心線程數(shù)，如果小于corePoolSize核心線程數(shù)，則向workers工作線程集合中添加一個核心線程執(zhí)行任務(wù)。代碼如下所示。
//線程池中的線程數(shù)量小于corePoolSize的值
if (workerCountOf(c) < corePoolSize) {
//重新開啟線程執(zhí)行任務(wù)
if (addWorker(command, true))
return;
c = ctl.get();
}
（2）如果線程池中的線程數(shù)量大于corePoolSize核心線程數(shù)，則判斷當(dāng)前線程池是否處于RUNNING狀態(tài)，如果處于RUNNING狀態(tài)，則添加任務(wù)到待執(zhí)行的任務(wù)隊(duì)列中。注意：這里向任務(wù)隊(duì)列添加任務(wù)時，需要判斷線程池是否處于RUNNING狀態(tài)，只有線程池處于RUNNING狀態(tài)時，才能向任務(wù)隊(duì)列添加新任務(wù)。否則，會執(zhí)行拒絕策略。代碼如下所示。
if (isRunning(c) && workQueue.offer(command))
（3）向任務(wù)隊(duì)列中添加任務(wù)成功，由于其他線程可能會修改線程池的狀態(tài)，所以這里需要對線程池進(jìn)行二次檢查，如果當(dāng)前線程池的狀態(tài)不再是RUNNING狀態(tài)，則需要將添加的任務(wù)從任務(wù)隊(duì)列中移除，執(zhí)行后續(xù)的拒絕策略。如果當(dāng)前線程池仍然處于RUNNING狀態(tài)，則判斷線程池是否為空，如果線程池中不存在任何線程，則新建一個線程添加到線程池中，如下所示。
//再次獲取線程池的狀態(tài)和線程池中線程的數(shù)量，用于二次檢查 int recheck = ctl.get(); //如果線程池沒有未處于RUNNING狀態(tài)，從隊(duì)列中刪除任務(wù) if (! isRunning(recheck) && remove(command)) //執(zhí)行拒絕策略 reject(command); //如果線程池為空，則向線程池中添加一個線程 else if (workerCountOf(recheck) == 0) addWorker(null, false);
（4）如果在步驟（3）中向任務(wù)隊(duì)列中添加任務(wù)失敗，則嘗試開啟新的線程執(zhí)行任務(wù)。此時，如果線程池中的線程數(shù)量已經(jīng)大于線程池中的最大線程數(shù)maximumPoolSize，則不能再啟動新線程。此時，表示線程池中的任務(wù)隊(duì)列已滿，并且線程池中的線程已滿，需要執(zhí)行拒絕策略，代碼如下所示。
//任務(wù)隊(duì)列已滿，則新增worker線程，如果新增線程失敗，則執(zhí)行拒絕策略 else if (!addWorker(command, false)) reject(command);
這里，我們將execute(Runnable)方法拆解，結(jié)合流程圖來理解線程池中任務(wù)的執(zhí)行流程就比較簡單了?？梢赃@么說，execute(Runnable)方法的邏輯基本上就是一般線程池的執(zhí)行邏輯，理解了execute(Runnable)方法，就基本理解了線程池的執(zhí)行邏輯。
注意：有關(guān)ScheduledThreadPoolExecutor類和ForkJoinPool類執(zhí)行線程池的邏輯，在【高并發(fā)專題】系列文章中的后文中會詳細(xì)說明，理解了這些類的執(zhí)行邏輯，就基本全面掌握了線程池的執(zhí)行流程。
在分析execute(Runnable)方法的源碼時，我們發(fā)現(xiàn)execute(Runnable)方法中多處調(diào)用了addWorker(Runnable, boolean)方法，接下來，我們就一起分析下addWorker(Runnable, boolean)方法的邏輯。
addWorker(Runnable, boolean)方法
總體上，addWorker(Runnable, boolean)方法可以分為三部分，第一部分是使用CAS安全的向線程池中添加工作線程；第二部分是創(chuàng)建新的工作線程；第三部分則是將任務(wù)通過安全的并發(fā)方式添加到workers中，并啟動工作線程執(zhí)行任務(wù)。
接下來，我們看下addWorker(Runnable, boolean)方法的源碼，如下所示。
private boolean addWorker(Runnable firstTask, boolean core) {
//標(biāo)記重試的標(biāo)識
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 檢查隊(duì)列是否在某些特定的條件下為空
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
//下面循環(huán)的主要作用為通過CAS方式增加線程的個數(shù)
for (;;) {
//獲取線程池中的線程數(shù)量
int wc = workerCountOf(c);
//如果線程池中的線程數(shù)量超出限制，直接返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//通過CAS方式向線程池新增線程數(shù)量
if (compareAndIncrementWorkerCount(c))
//通過CAS方式保證只有一個線程執(zhí)行成功，跳出最外層循環(huán)
break retry;
//重新獲取ctl的值
c = ctl.get();
//如果CAS操作失敗了，則需要在內(nèi)循環(huán)中重新嘗試通過CAS新增線程數(shù)量
if (runStateOf(c) != rs)
continue retry;
}
}
//跳出最外層for循環(huán)，說明通過CAS新增線程數(shù)量成功
//此時創(chuàng)建新的工作線程
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//將執(zhí)行的任務(wù)封裝成worker
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
//獨(dú)占鎖，保證操作workers時的同步
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//此處需要重新檢查線程池狀態(tài)
//原因是在獲得鎖之前可能其他的線程改變了線程池的狀態(tài)
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive())
throw new IllegalThreadStateException();
//向worker中添加新任務(wù)
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
//將是否添加了新任務(wù)的標(biāo)識設(shè)置為true
workerAdded = true;
}
} finally {
//釋放獨(dú)占鎖
mainLock.unlock();
}
//添加新任成功，則啟動線程執(zhí)行任務(wù)
if (workerAdded) {
t.start();
//將任務(wù)是否已經(jīng)啟動的標(biāo)識設(shè)置為true
workerStarted = true;
}
}
} finally {
//如果任務(wù)未啟動或啟動失敗，則調(diào)用addWorkerFailed(Worker)方法
if (! workerStarted)
addWorkerFailed(w);
}
//返回是否啟動任務(wù)的標(biāo)識
return workerStarted;
}
乍一看，addWorker(Runnable, boolean)方法還蠻長的，這里，我們還是將addWorker(Runnable, boolean)方法進(jìn)行拆解。
（1）檢查任務(wù)隊(duì)列是否在某些特定的條件下為空，代碼如下所示。
// 檢查隊(duì)列是否在某些特定的條件下為空 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false;
（2）在通過步驟（1）的校驗(yàn)后，則進(jìn)入內(nèi)層for循環(huán)，在內(nèi)層for循環(huán)中通過CAS來增加線程池中的線程數(shù)量，如果CAS操作成功，則直接退出雙重for循環(huán)。如果CAS操作失敗，則查看當(dāng)前線程池的狀態(tài)是否發(fā)生了變化，如果線程池的狀態(tài)發(fā)生了變化，則通過continue關(guān)鍵字重新通過外層for循環(huán)校驗(yàn)任務(wù)隊(duì)列，檢驗(yàn)通過再次執(zhí)行內(nèi)層for循環(huán)的CAS操作。如果線程池的狀態(tài)沒有發(fā)生變化，此時上一次CAS操作失敗了，則繼續(xù)嘗試CAS操作。代碼如下所示。
for (;;) {
//獲取線程池中的線程數(shù)量
int wc = workerCountOf(c);
//如果線程池中的線程數(shù)量超出限制，直接返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//通過CAS方式向線程池新增線程數(shù)量
if (compareAndIncrementWorkerCount(c))
//通過CAS方式保證只有一個線程執(zhí)行成功，跳出最外層循環(huán)
break retry;
//重新獲取ctl的值
c = ctl.get();
//如果CAS操作失敗了，則需要在內(nèi)循環(huán)中重新嘗試通過CAS新增線程數(shù)量
if (runStateOf(c) != rs)
continue retry;
}
（3）CAS操作成功后，表示向線程池中成功添加了工作線程，此時，還沒有線程去執(zhí)行任務(wù)。使用全局的獨(dú)占鎖mainLock來將新增的工作線程Worker對象安全的添加到workers中。
總體邏輯就是：創(chuàng)建新的Worker對象，并獲取Worker對象中的執(zhí)行線程，如果線程不為空，則獲取獨(dú)占鎖，獲取鎖成功后，再次檢查線線程的狀態(tài)，這是避免在獲取獨(dú)占鎖之前其他線程修改了線程池的狀態(tài)，或者關(guān)閉了線程池。如果線程池關(guān)閉，則需要釋放鎖。否則將新增加的線程添加到工作集合中，釋放鎖并啟動線程執(zhí)行任務(wù)。將是否啟動線程的標(biāo)識設(shè)置為true。最后，判斷線程是否啟動，如果沒有啟動，則調(diào)用addWorkerFailed(Worker)方法。最終返回線程是否起送的標(biāo)識。
//跳出最外層for循環(huán)，說明通過CAS新增線程數(shù)量成功
//此時創(chuàng)建新的工作線程
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//將執(zhí)行的任務(wù)封裝成worker
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
//獨(dú)占鎖，保證操作workers時的同步
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//此處需要重新檢查線程池狀態(tài)
//原因是在獲得鎖之前可能其他的線程改變了線程池的狀態(tài)
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive())
throw new IllegalThreadStateException();
//向worker中添加新任務(wù)
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
//將是否添加了新任務(wù)的標(biāo)識設(shè)置為true
workerAdded = true;
}
} finally {
//釋放獨(dú)占鎖
mainLock.unlock();
}
//添加新任成功，則啟動線程執(zhí)行任務(wù)
if (workerAdded) {
t.start();
//將任務(wù)是否已經(jīng)啟動的標(biāo)識設(shè)置為true
workerStarted = true;
}
}
} finally {
//如果任務(wù)未啟動或啟動失敗，則調(diào)用addWorkerFailed(Worker)方法
if (! workerStarted)
addWorkerFailed(w);
}
//返回是否啟動任務(wù)的標(biāo)識
return workerStarted;
addWorkerFailed(Worker)方法
在addWorker(Runnable, boolean)方法中，如果添加工作線程失敗或者工作線程啟動失敗時，則會調(diào)用addWorkerFailed(Worker)方法，下面我們就來看看addWorkerFailed(Worker)方法的實(shí)現(xiàn)，如下所示。
private void addWorkerFailed(Worker w) {
//獲取獨(dú)占鎖
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//如果Worker任務(wù)不為空
if (w != null)
//將任務(wù)從workers集合中移除
workers.remove(w);
//通過CAS將任務(wù)數(shù)量減1
decrementWorkerCount();
tryTerminate();
} finally {
//釋放鎖
mainLock.unlock();
}
}
addWorkerFailed(Worker)方法的邏輯就比較簡單了，獲取獨(dú)占鎖，將任務(wù)從workers中移除，并且通過CAS將任務(wù)的數(shù)量減1，最后釋放鎖。
拒絕策略
我們在分析execute(Runnable)方法時，線程池會在適當(dāng)?shù)臅r候調(diào)用reject(Runnable)方法來執(zhí)行相應(yīng)的拒絕策略，我們看下reject(Runnable)方法的實(shí)現(xiàn)，如下所示。
final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}
通過代碼，我們發(fā)現(xiàn)調(diào)用的是handler的rejectedExecution方法，handler又是個什么鬼，我們繼續(xù)跟進(jìn)代碼，如下所示。
private volatile RejectedExecutionHandler handler;
再看看RejectedExecutionHandler是個啥類型，如下所示。
package java.util.concurrent;
public interface RejectedExecutionHandler {
void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}
可以發(fā)現(xiàn)RejectedExecutionHandler是個接口，定義了一個rejectedExecution(Runnable, ThreadPoolExecutor)方法。既然RejectedExecutionHandler是個接口，那我們就看看有哪些類實(shí)現(xiàn)了RejectedExecutionHandler接口。
看到這里，我們發(fā)現(xiàn)RejectedExecutionHandler接口的實(shí)現(xiàn)類正是線程池默認(rèn)提供的四種拒絕策略的實(shí)現(xiàn)類。
至于reject(Runnable)方法中具體會執(zhí)行哪個類的拒絕策略，是根據(jù)創(chuàng)建線程池時傳遞的參數(shù)決定的。如果沒有傳遞拒絕策略，則默認(rèn)會執(zhí)行AbortPolicy類的拒絕策略。否則會執(zhí)行傳遞的類的拒絕策略。
在創(chuàng)建線程池時，除了能夠傳遞JDK默認(rèn)提供的拒絕策略外，還可以傳遞自定義的拒絕策略。如果想使用自定義的拒絕策略，則只需要實(shí)現(xiàn)RejectedExecutionHandler接口，并重寫rejectedExecution(Runnable, ThreadPoolExecutor)方法即可。例如，下面的代碼。
public class CustomPolicy implements RejectedExecutionHandler {
public CustomPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
System.out.println("使用調(diào)用者所在的線程來執(zhí)行任務(wù)")
r.run();
}
}
}
使用如下方式創(chuàng)建線程池。
new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
Executors.defaultThreadFactory(),
new CustomPolicy());
至此，線程池執(zhí)行任務(wù)的整體核心邏輯分析結(jié)束，更多關(guān)于java ThreadPoolExecutor類解析線程池的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
JVM教程之Java代碼編譯和執(zhí)行的整個過程（二）
這篇文章主要介紹了JVM學(xué)習(xí)筆記第二篇，關(guān)于Java代碼編譯和執(zhí)行的整個過程，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-03-03
java中double轉(zhuǎn)化為BigDecimal精度缺失的實(shí)例
下面小編就為大家?guī)硪黄猨ava中double轉(zhuǎn)化為BigDecimal精度缺失的實(shí)例。小編覺得挺不錯的，現(xiàn)在就分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2017-03-03
今天給大家?guī)淼氖顷P(guān)于Springboot的相關(guān)知識,文章圍繞著RestTemplate未使用線程池展開,文中有非常詳細(xì)的介紹及代碼示例,需要的朋友可以參考下2021-06-06
安全管理是軟件系統(tǒng)必不可少的的功能。根據(jù)經(jīng)典的“墨菲定律”——凡是可能，總會發(fā)生。如果系統(tǒng)存在安全隱患，最終必然會出現(xiàn)問題，這篇文章主要介紹了SpringBoot安全管理Shiro框架的使用2022-08-08
spring?boot?validation參數(shù)校驗(yàn)與分組嵌套各種類型及使用小結(jié)
參數(shù)校驗(yàn)基本上是controller必做的事情，畢竟前端傳過來的一切都不可信，validation可以簡化這一操作，這篇文章主要介紹了spring?boot?validation參數(shù)校驗(yàn)分組嵌套各種類型及使用小結(jié),需要的朋友可以參考下2023-09-09
Java實(shí)現(xiàn)分布式系統(tǒng)限流
這篇文章主要為大家詳細(xì)介紹了Java實(shí)現(xiàn)分布式系統(tǒng)限流，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-08-08
Maven多模塊之父子關(guān)系的創(chuàng)建
這篇文章主要介紹了Maven多模塊之父子關(guān)系的創(chuàng)建，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-03-03
這篇文章主要介紹了Java抽象類原理與用法,結(jié)合實(shí)例形式詳細(xì)分析了Java抽象類相關(guān)概念、原理、定義、使用方法及操作注意事項(xiàng),需要的朋友可以參考下2019-11-11
springBoot項(xiàng)目如何實(shí)現(xiàn)啟動多個實(shí)例
這篇文章主要介紹了springBoot項(xiàng)目如何實(shí)現(xiàn)啟動多個實(shí)例的操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-08-08
java中實(shí)現(xiàn)漢字按照拼音排序(示例代碼)
這篇文章主要是對java中將漢字按照拼音排序的實(shí)現(xiàn)代碼進(jìn)行了詳細(xì)的分析介紹。需要的朋友可以過來參考下，希望對大家有所幫助2013-12-12