前言
在上一篇文章深入淺出Java線程池：理論篇中，已經(jīng)介紹了什么是線程池以及基本的使用。（本來寫作的思路是使用篇，但經(jīng)網(wǎng)友建議后，感覺改為理論篇會(huì)更加合適）。本文則深入線程池的源碼，主要是介紹ThreadPoolExecutor內(nèi)部的源碼是如何實(shí)現(xiàn)的，對ThreadPoolExecutor有一個(gè)更加清晰的認(rèn)識。
ThreadPoolExecutor的源碼相對而言比較好理解，沒有特別難以讀懂的地方。相信沒有閱讀源碼習(xí)慣的讀者，跟著本文，也可以很輕松地讀懂ThreadPoolExecutor的核心源碼邏輯。
本文源碼jdk版本為8，該類版本為jdk1.5，也就是在1.5之后，ThreadPoolExecutor的源碼沒有做修改。
線程池家族
Java中的線程池繼承結(jié)構(gòu)如下圖：(類圖中只寫了部分方法且省略參數(shù))
- 頂層接口Executor表示一個(gè)執(zhí)行器，他只有一個(gè)接口：
execute() ，表示可以執(zhí)行任務(wù)
- ExecutorService在Executor的基礎(chǔ)上拓展了更多的執(zhí)行方法，如
submit() shutdown() 等等,表示一個(gè)任務(wù)執(zhí)行服務(wù)。
- AbstarctExecutorService是一個(gè)抽象類，他實(shí)現(xiàn)了ExecutorService的部分核心方法，如submit等
- ThreadPoolExecutor是最核心的類，也就是線程池，他繼承了抽象類AbstarctExecutorService
- 此外還有ScheduledExecutorService接口，他表示一個(gè)可以按照指定時(shí)間或周期執(zhí)行的執(zhí)行器服務(wù)，內(nèi)部定義了如
schedule() 等方法來執(zhí)行任務(wù)
- ScheduledThreadPoolExecutor實(shí)現(xiàn)了ScheduledExecutorService接口，同時(shí)繼承于ThreadPoolExecutor，內(nèi)部的線程池相關(guān)邏輯使用自ThreadPoolExecutor，在此基礎(chǔ)上拓展了延遲、周期執(zhí)行等功能特性
ScheduledThreadPoolExecutor相對來說用的是比較少。延時(shí)任務(wù)在我們Android中有更加熟悉的方案：Handler；而周期任務(wù)則用的非常少?，F(xiàn)在android的后臺限制非常嚴(yán)格，基本上一退出應(yīng)用，應(yīng)用進(jìn)程很容易被系統(tǒng)干掉。當(dāng)然ScheduledThreadPoolExecutor也不是完全沒有用處，例如桌面小部件需要設(shè)置定時(shí)刷新，那么他就可以派上用場了。
因此，我們本文的源碼，主要針對ThreadPoolExecutor。在閱讀源碼之前，我們先來看一下ThreadPoolExecutor內(nèi)部的結(jié)構(gòu)以及關(guān)鍵角色。
內(nèi)部結(jié)構(gòu)
閱讀源碼前，我們先把ThreadPoolExecutor整個(gè)源碼結(jié)構(gòu)講解一下，形成一個(gè)整體概念，再閱讀源碼就不會(huì)迷失在源碼中了。先來看一下ThreadPoolExecutor的內(nèi)部結(jié)構(gòu)：
- ThreadPoolExecutor內(nèi)部有三個(gè)關(guān)鍵的角色：阻塞隊(duì)列、線程、以及RejectExecutionHandler（這里寫個(gè)中文名純粹因?yàn)椴恢涝趺捶g這個(gè)名字），他們的作用在理論篇有詳細(xì)介紹，這里不再贅述。
- 在ThreadPoolExecutor中，一個(gè)線程對應(yīng)一個(gè)worker對象，工人，非常形象。每個(gè)worker內(nèi)部有一個(gè)獨(dú)立的線程，他會(huì)不斷去阻塞隊(duì)列獲取任務(wù)來執(zhí)行，也就是調(diào)用阻塞隊(duì)列的
poll 或者 take 方法，他們區(qū)別后面會(huì)講。如果隊(duì)列沒有任務(wù)了，那么就會(huì)阻塞在這里。
- workQueue，就是阻塞隊(duì)列，當(dāng)核心線程已滿之后，任務(wù)就會(huì)被放置在這里等待被工人worker領(lǐng)取執(zhí)行
- RejectExecutionHandler本身是一個(gè)接口，ThreadPoolExecutor內(nèi)部有這樣的一個(gè)接口對象，當(dāng)任務(wù)無法被執(zhí)行會(huì)調(diào)用這個(gè)對象的方法。ThreadPoolExecutor提供了該接口的4種實(shí)現(xiàn)方案，我們可以直接拿來用，或者自己繼承接口，實(shí)現(xiàn)自定義邏輯。在構(gòu)造線程池的時(shí)候可以傳入RejectExecutionHandler對象。
- 整個(gè)ThreadPoolExecutor中最核心的方法就是execute，他會(huì)根據(jù)具體的情況來選擇不同的執(zhí)行方案或者拒絕執(zhí)行。
這樣，我們就清楚ThreadPoolExecutor的內(nèi)部結(jié)構(gòu)了，然后，我們開始 Read the fucking code 吧。
源碼分析
內(nèi)部關(guān)鍵屬性
ThreadPoolExecutor內(nèi)部有很多的變量，他們包含的信息非常重要，先來了解一下。
ThreadPoolExecutor的狀態(tài)和線程數(shù)整合在同一個(gè)int變量中，類似于view測量中MeasureSpec。他的高三位表示線程池的狀態(tài)，低29位表示線程池中線程的數(shù)量,如下：
// AtomicInteger對象可以利用CAS實(shí)現(xiàn)線程安全的修改，其中包含了線程池狀態(tài)和線程數(shù)量信息
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// COUNT_BITS=29,（對于int長度為32來說）表示線程數(shù)量的字節(jié)位數(shù)
private static final int COUNT_BITS = Integer.SIZE - 3;
// 狀態(tài)掩碼，高三位是1，低29位全是0，可以通過 ctl&COUNT_MASK 運(yùn)算來獲取線程池狀態(tài)
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;
線程池的狀態(tài)一共有5個(gè)：
- 運(yùn)行running：線程池創(chuàng)建之后即是運(yùn)行狀態(tài)
- 關(guān)閉shutdown：調(diào)用shutdown方法之后線程池處于shutdown狀態(tài)，該狀態(tài)會(huì)停止接收任何任務(wù)，阻塞隊(duì)列中的任務(wù)執(zhí)行完成之后會(huì)自動(dòng)終止線程池
- 停止stop：調(diào)用shutdownNow方法之后線程池處于stop狀態(tài)。和shutdown的區(qū)別是這個(gè)狀態(tài)下的線程池不會(huì)去執(zhí)行隊(duì)列中剩下的任務(wù)
- 整理tidying：在線程池stop之后，進(jìn)入tidying狀態(tài)，然后執(zhí)行
terminated() 方法，再進(jìn)入terminated狀態(tài)
- 終止terminated：線程池中沒有任何線程在執(zhí)行任務(wù)，線程池完全終止。
在源碼中這幾個(gè)狀態(tài)分別對應(yīng)：
// runState is stored in the high-order bits
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;
上面的位操作不夠直觀，轉(zhuǎn)化后如下：
private static final int RUNNING = 111 00000 00000000 00000000 00000000;
private static final int SHUTDOWN = 000 00000 00000000 00000000 00000000;
private static final int STOP = 001 00000 00000000 00000000 00000000;
private static final int TIDYING = 010 00000 00000000 00000000 00000000;
private static final int TERMINATED = 011 00000 00000000 00000000 00000000;
可以看到除了running是負(fù)數(shù)，其他的狀態(tài)都是正數(shù)，且狀態(tài)越靠后，數(shù)值越大。因此我們可以通過判斷 ctl&COUNT_MASK > SHUTDOWN 來判斷狀態(tài)是否處于 stop、tidying、terminated之一。后續(xù)源碼中會(huì)有很多的這樣的判斷，舉其中的一個(gè)方法：
// 這里來判斷線程池的狀態(tài)
if(runStateAtLeast(ctl,SHUTDOWN)) {
...
}
// 這里執(zhí)行邏輯，直接判斷兩個(gè)數(shù)的大小
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
ps：這里怎么沒有使用掩碼COUNT_MASK ？因?yàn)闋顟B(tài)是處于高位，低位的數(shù)值不影響高位的大小判斷。當(dāng)然如果要判斷相等，就還是需要使用掩碼COUNT_MASK的。
接下來是ThreadPoolExecutor內(nèi)部的三個(gè)關(guān)鍵角色對象：
// 阻塞隊(duì)列
private final BlockingQueue<Runnable> workQueue;
// 存儲(chǔ)worker的hashSet,worker被創(chuàng)建之后會(huì)被存儲(chǔ)到這里
private final HashSet<Worker> workers = new HashSet<>();
// RejectedExecutionHandler默認(rèn)的實(shí)現(xiàn)是AbortPolicy
private volatile RejectedExecutionHandler handler;
private static final RejectedExecutionHandler defaultHandler =
new AbortPolicy();
內(nèi)部使用的鎖對象：
// 這里是兩個(gè)鎖。ThreadPoolExecutor內(nèi)部并沒有使用Synchronize關(guān)鍵字來保持同步
// 而是使用Lock；和Synchronize的區(qū)別就是他是應(yīng)用層的鎖，而synchronize是jvm層的鎖
private final ReentrantLock mainLock = new ReentrantLock();
private final Condition termination = mainLock.newCondition();
最后是內(nèi)部一些參數(shù)的配置，前面都介紹過，把源碼貼出來再回顧一下：
// 線程池歷史達(dá)到的最大線程數(shù)
private int largestPoolSize;
// 線程池完成的任務(wù)數(shù)。
// 該數(shù)并不是實(shí)時(shí)更新的，在獲取線程池完成的任務(wù)數(shù)時(shí)，需要去統(tǒng)計(jì)每個(gè)worker完成的任務(wù)并累加起來
// 當(dāng)一個(gè)worker被銷毀之后，他的任務(wù)數(shù)就會(huì)被累加到這個(gè)數(shù)據(jù)中
private long completedTaskCount;
// 線程工廠，用于創(chuàng)建線程
private volatile ThreadFactory threadFactory;
// 空閑線程存儲(chǔ)的時(shí)間
private volatile long keepAliveTime;
// 是否允許核心線程被回收
private volatile boolean allowCoreThreadTimeOut;
// 核心線程數(shù)限額
private volatile int corePoolSize;
// 線程總數(shù)限額
private volatile int maximumPoolSize;
不是吧sir？源碼還沒看到魂呢，整出來這么無聊的變量？
咳咳，別急嘛，源碼解析馬上來。這些變量會(huì)貫穿整個(gè)源碼過程始終，先對他們有個(gè)印象，后續(xù)閱讀源碼就會(huì)輕松暢通很多。
關(guān)鍵方法：execute()
這個(gè)方法的主要任務(wù)就是根據(jù)線程池的當(dāng)前狀態(tài)，選擇任務(wù)的執(zhí)行策略。該方法的核心邏輯思路是：
-
在線程數(shù)沒有達(dá)到核心線程數(shù)時(shí)，會(huì)創(chuàng)建一個(gè)核心線程來執(zhí)行任務(wù)
public void execute(Runnable command) {
// 不能傳入空任務(wù)
if (command == null)
throw new NullPointerException();
// 獲取ctl變量，就是上面我們講的將狀態(tài)和線程數(shù)合在一起的一個(gè)變量
int c = ctl.get();
// 判斷核心線程數(shù)是否超過限額，否則創(chuàng)建一個(gè)核心線程來執(zhí)行任務(wù)
if (workerCountOf(c) < corePoolSize) {
// addWorker方法是創(chuàng)建一個(gè)worker，也就是創(chuàng)建一個(gè)線程，參數(shù)true表示這是一個(gè)核心線程
// 如果添加成功則直接返回
// 否則意味著中間有其他的worker被添加了，導(dǎo)致超出核心線程數(shù);或者線程池被關(guān)閉了等其他情況
// 需要進(jìn)入下一步繼續(xù)判斷
if (addWorker(command, true))
return;
c = ctl.get();
}
...
}
-
當(dāng)線程數(shù)達(dá)到核心線程數(shù)時(shí)，新任務(wù)會(huì)被放入到等待隊(duì)列中等待被執(zhí)行
-
當(dāng)?shù)却?duì)列已經(jīng)滿了之后，如果線程數(shù)沒有到達(dá)總的線程數(shù)上限，那么會(huì)創(chuàng)建一個(gè)非核心線程來執(zhí)行任務(wù)
-
當(dāng)線程數(shù)已經(jīng)到達(dá)總的線程數(shù)限制時(shí)，新的任務(wù)會(huì)被拒絕策略者處理，線程池?zé)o法執(zhí)行該任務(wù)。
public void execute(Runnable command) {
...
// 如果線程池還在運(yùn)行,則嘗試添加任務(wù)到隊(duì)列中
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 再次檢查如果線程池被關(guān)閉了，那么把任務(wù)移出隊(duì)列
// 如果移除成功則拒絕本次任務(wù)
// 這里主要是判斷在插入隊(duì)列的過程中，線程池有沒有被關(guān)閉了
if (! isRunning(recheck) && remove(command))
reject(command);
// 否則再次檢查線程數(shù)是否為0，如果是，則創(chuàng)建一個(gè)沒有任務(wù)的非主線程worker
// 這里對應(yīng)核心線程為0的情況，指定任務(wù)為null，worker會(huì)去隊(duì)列拿任務(wù)來執(zhí)行
// 這里表示線程池至少有一個(gè)線程來執(zhí)行隊(duì)列中的任務(wù)
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 如果上面添加到隊(duì)列中失敗，則嘗試創(chuàng)建一個(gè)非核心線程來執(zhí)行任務(wù)
// 如果創(chuàng)建失敗，則拒絕任務(wù)
else if (!addWorker(command, false))
reject(command);
}
源碼中還設(shè)計(jì)到兩個(gè)關(guān)鍵方法：addWorker創(chuàng)建一個(gè)新的worker，也就是創(chuàng)建一個(gè)線程；reject拒絕一個(gè)任務(wù)。后者比較簡單我們先看一下。
拒絕任務(wù)：reject()
// 拒絕任務(wù)，調(diào)用rejectedExecutionHandler來處理
final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}
默認(rèn)的實(shí)現(xiàn)類有4個(gè)，我們依次來看一下：
-
AbortPolicy是默認(rèn)實(shí)現(xiàn)，會(huì)拋出一個(gè)RejectedExecutionException異常：
public static class AbortPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
-
DiscardPolicy最簡單，就是：什么都不做，直接拋棄任務(wù)。（這是非常渣男不負(fù)責(zé)任的行為，咱們不能學(xué)他，所以也不要用它 [此處狗頭] ）
public static class DiscardPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
-
DiscardOldestPolicy會(huì)刪除隊(duì)列頭的一個(gè)任務(wù)，然后再次執(zhí)行自己（擠掉原位，自己上位，綠茶行為？）
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
-
CallerRunsPolicy最猛，他干脆在自己的線程執(zhí)行run方法，不依靠線程池了，自己動(dòng)手豐衣足食。
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
上面4個(gè)ThreadPoolExecutor已經(jīng)幫我們實(shí)現(xiàn)了，他的靜態(tài)內(nèi)部類，在創(chuàng)建ThreadPoolExecutor的時(shí)候我們可以直接拿來用。也可以自己繼承接口實(shí)現(xiàn)自己的邏輯。具體選擇哪個(gè)需要根據(jù)實(shí)際的業(yè)務(wù)需求來決定。
那么接下來看創(chuàng)建worker的方法。
創(chuàng)建worker：addWorker()
方法的目的很簡單：創(chuàng)建一個(gè)worker。前面我們講到，worker內(nèi)部創(chuàng)建了一個(gè)線程，每一個(gè)worker則代表了一個(gè)線程，非常類似android中的looper。looper的loop()方法會(huì)不斷地去MessageQueue獲取message，而Worker的run()方法會(huì)不斷地去阻塞隊(duì)列獲取任務(wù)，這個(gè)我們后面講。
addWorker() 方法的邏輯整體上分為兩個(gè)部分：
-
檢查線程狀態(tài)和線程數(shù)是否滿足條件：
// 第一個(gè)參數(shù)是創(chuàng)建的線程首次要執(zhí)行的任務(wù)，可以是null，則表示初始化一個(gè)線程
// 第二參數(shù)表示是否是一個(gè)核心線程
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (int c = ctl.get();;) {
// 還記不記得我們前面講到線程池的狀態(tài)控制？
// runStateAtLeast(c, SHUTDOWN)表示狀態(tài)至少為shutdown，后面類同
// 如果線程池處于stop及以上，不會(huì)再創(chuàng)建worker
// 如果線程池狀態(tài)在shutdown時(shí)，如果隊(duì)列不為空或者任務(wù)!=null，則還會(huì)創(chuàng)建worker
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP)
|| firstTask != null
|| workQueue.isEmpty()))
// 其他情況返回false，表示拒絕創(chuàng)建worker
return false;
// 這里采用CAS輪詢，也就是循環(huán)鎖的策略來讓線程總數(shù)+1
for (;;) {
// 檢查是否超出線程數(shù)限制
// 這里根據(jù)core參數(shù)判斷是核心線程還是非核心線程
if (workerCountOf(c)
>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
return false;
// 利用CAS讓ctl變量自增，表示worker+1
// 如果CAS失敗，則表示發(fā)生了競爭，則再來一次
if (compareAndIncrementWorkerCount(c))
// 成功則跳出最外層循環(huán)
break retry;
// 如果這個(gè)期間ctl被改變了，則獲取ctl，再嘗試一次
c = ctl.get();
// 如果線程池被shutdown了，那么重復(fù)最外層的循環(huán)，重新判斷狀態(tài)是否可以創(chuàng)建worker
if (runStateAtLeast(c, SHUTDOWN))
// 繼續(xù)最外層循環(huán)
continue retry;
}
}
// 創(chuàng)建worker邏輯
...
}
不知道讀者對于源碼中的retry: 有沒有疑惑，畢竟平時(shí)很少用到。他的作用是標(biāo)記一個(gè)循環(huán)，這樣我們在內(nèi)層的循環(huán)就可以跳轉(zhuǎn)到任意一個(gè)外層的循環(huán)。這里的retry只是一個(gè)名字，改成 repeat: 甚至 a: 都是可以的。他的本質(zhì)就是：一個(gè)循環(huán)的標(biāo)記 。
-
創(chuàng)建worker對象，并調(diào)用其內(nèi)部線程的start()方法來啟動(dòng)線程：
private boolean addWorker(Runnable firstTask, boolean core) {
...
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
// 創(chuàng)建一個(gè)新的worker
// 創(chuàng)建的過程中內(nèi)部會(huì)創(chuàng)建一個(gè)線程
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
// 獲得全局鎖并加鎖
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 獲取鎖之后，需要再次檢查狀態(tài)
int c = ctl.get();
// 只有運(yùn)行狀態(tài)或者shutDown&&task==null才會(huì)被執(zhí)行
if (isRunning(c) ||
(runStateLessThan(c, STOP) && firstTask == null)) {
// 如果這個(gè)線程不是剛創(chuàng)建的，則拋出異常
if (t.getState() != Thread.State.NEW)
throw new IllegalThreadStateException();
// 添加到workerSet中
workers.add(w);
workerAdded = true;
int s = workers.size();
// 跟蹤線程池到達(dá)的最多線程數(shù)量
if (s > largestPoolSize)
largestPoolSize = s;
}
} finally {
// 釋放鎖
mainLock.unlock();
}
// 如果添加成功，啟動(dòng)線程
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
// 如果線程沒有啟動(dòng)，表示添加worker失敗，可能在添加的過程中線程池被關(guān)閉了
if (! workerStarted)
// 把worker從workerSet中移除
addWorkerFailed(w);
}
return workerStarted;
}
經(jīng)過前面兩步，如果沒有出現(xiàn)異常，則創(chuàng)建worker成功。最后還涉及到一個(gè)方法： addWorkerFailed(w) ，他的內(nèi)容比較簡答，順便提一下吧：
// 添加worker失敗
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
// 加鎖
mainLock.lock();
try {
if (w != null)
workers.remove(w);
// 這里會(huì)讓線程總數(shù)-1
decrementWorkerCount();
// 嘗試設(shè)置線程池的狀態(tài)為terminad
// 因?yàn)樘砑邮∮锌赡苁蔷€程池在添加worker的過程中被shutdown
// 那么這個(gè)時(shí)候如果沒有任務(wù)正在執(zhí)行就需要設(shè)置狀態(tài)為terminad
// 這個(gè)方法后面會(huì)詳細(xì)講
tryTerminate();
} finally {
mainLock.unlock();
}
}
那么到這里，execute()方法中的一些調(diào)用方法就分析完了。阻塞隊(duì)列相關(guān)的方法不屬于本文的范疇，就不展開了。那么還有一個(gè)問題：worker是如何工作的呢？worker內(nèi)部有一個(gè)線程，當(dāng)線程啟動(dòng)時(shí)，初始化線程的runnable對象的run方法會(huì)被調(diào)用，那么這個(gè)runnable對象是什么？我直接來看worker。
打工人：Worker
首先我們看到他的構(gòu)造方法：
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
源碼很簡單，把傳進(jìn)來的任務(wù)設(shè)置給內(nèi)部變量firstTask，然后把自己傳給線程工廠去創(chuàng)建一個(gè)線程。所以線程啟動(dòng)時(shí)，Worker本身的run方法會(huì)被調(diào)用，那么我們看到Worker的 run()方法。
public void run() {
runWorker(this);
}
Worker是ThreadPoolExecutor的內(nèi)部類，這里直接調(diào)用到了ThreadPoolExecutor的方法： runWorker()來開始執(zhí)行。那么接下來，我們就看到這個(gè)方法。
啟動(dòng)worker：runWorker()
這個(gè)方法是worker執(zhí)行的方法，在線程被銷毀前他會(huì)一直執(zhí)行，類似于Handler的looper，不斷去隊(duì)列獲取消息來執(zhí)行：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
// 獲取worker初始化時(shí)設(shè)置的任務(wù)，可以為null。如果為null則表示僅僅創(chuàng)建線程
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock();
// 這個(gè)參數(shù)的作用后面解釋，需要結(jié)合其他的源碼
boolean completedAbruptly = true;
try {
// 如果自身的task不為null，那么會(huì)執(zhí)行自身的task
// 否則調(diào)用getTask去隊(duì)列獲取一個(gè)task來執(zhí)行
// 這個(gè)getTask最終會(huì)去調(diào)用隊(duì)列的方法來獲取任務(wù)
// 而隊(duì)列如果為空他的獲取方法會(huì)進(jìn)行阻塞，這里也就阻塞了,后面深入講
while (task != null || (task = getTask()) != null) {
try{
// 執(zhí)行任務(wù)
...
} finally {
// 任務(wù)執(zhí)行完成，把task設(shè)置為null
task = null;
// 任務(wù)總數(shù)+1
w.completedTasks++;
// 釋放鎖
w.unlock();
}
}
// 這里設(shè)置為false，先記住他
completedAbruptly = false;
} finally {
// 如果worker退出，那么需要執(zhí)行后續(xù)的善后工作
processWorkerExit(w, completedAbruptly);
}
}
可以看到這個(gè)方法的整體框架還是比較簡單的，核心就在于 while (task != null || (task = getTask()) != null) 這個(gè)循環(huán)中，如果 getTask() 返回null，則表示線程該結(jié)束了，這和Handler機(jī)制也是一樣的。
上面的源碼省略了具體執(zhí)行任務(wù)的邏輯，他的邏輯也是很簡單：判斷狀態(tài)+運(yùn)行任務(wù)。我們來看一下：
final void runWorker(Worker w) {
...;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
// 如果線程池已經(jīng)設(shè)置為stop狀態(tài)，那么保證線程是interrupted標(biāo)志
// 如果線程池沒有在stop狀態(tài)，那么保證線程不是interrupted標(biāo)志
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
// 回調(diào)方法，這個(gè)方法是一個(gè)空實(shí)現(xiàn)
beforeExecute(wt, task);
try {
// 運(yùn)行任務(wù)
task.run();
// 回調(diào)方法，也是一個(gè)空實(shí)現(xiàn)
afterExecute(task, null);
} catch (Throwable ex) {
afterExecute(task, ex);
throw ex;
}
}
...
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
在獲取到一個(gè)任務(wù)后，就會(huì)去執(zhí)行該任務(wù)的run方法，然后再回去繼續(xù)獲取新的任務(wù)。
我們會(huì)發(fā)現(xiàn)其中有很多的空實(shí)現(xiàn)方法，他是給子類去實(shí)現(xiàn)的，有點(diǎn)類似于Activity的生命周期，子類需要重寫這些方法，在具體的情況做一些工作。當(dāng)然，一般的使用是不需要去重寫這些方法。接下來需要來看看 getTask() 是如何獲取任務(wù)的。
獲取任務(wù)：getTask()
這個(gè)方法的內(nèi)容可以分為兩個(gè)部分：判斷當(dāng)前線程池的狀態(tài)+阻塞地從隊(duì)列中獲取一個(gè)任務(wù)。
第一部分是判斷當(dāng)前線程池的狀況，如果處于關(guān)閉狀態(tài)那么直接返回null來讓worker結(jié)束，否則需要判斷當(dāng)前線程是否超時(shí)或者超出最大限制的線程數(shù)：
private Runnable getTask() {
boolean timedOut = false;
// 內(nèi)部使用了CAS，這里需要有一個(gè)循環(huán)來不斷嘗試
for (;;) {
int c = ctl.get();
// 如果處于shutdown狀態(tài)而且隊(duì)列為空，或者處于stop狀態(tài)，返回null
// 這和前面我們討論到不同的線程池的狀態(tài)的不同行為一致
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
// 這里表示讓線程總數(shù)-1，記住他，后面會(huì)繼續(xù)聊到
decrementWorkerCount();
return null;
}
// 獲取目前的線程總數(shù)
int wc = workerCountOf(c);
// 判斷該線程在空閑情況是否可以被銷毀：允許核心線程為null或者當(dāng)前線程超出核心線程數(shù)
// 可以看到這里并沒有去區(qū)分具體的線程是核心還是非核心，只有線程數(shù)量處于核心范圍還是非核心范圍
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// 超出最大線程數(shù)或者已經(jīng)超時(shí)；
// 這里可能是用戶通過 setMaximumPoolSize 改動(dòng)了數(shù)據(jù)才會(huì)導(dǎo)致這里超出最大線程數(shù)
// 同時(shí)還必須保證當(dāng)前線程數(shù)量大于1或者隊(duì)列已經(jīng)沒有任務(wù)了
// 這樣就確保了當(dāng)有任務(wù)存在時(shí)，一定至少有一個(gè)線程在執(zhí)行任務(wù)
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
// 使用CAS嘗試讓當(dāng)前線程總數(shù)-1，失敗則從來一次上面的邏輯
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
// 獲取任務(wù)邏輯
...
}
}
第二部分是獲取一個(gè)任務(wù)并執(zhí)行。獲取任務(wù)使用的是阻塞隊(duì)列的方法，如果隊(duì)列中沒有任務(wù)，則會(huì)被阻塞：
private Runnable getTask() {
boolean timedOut = false;
// 內(nèi)部使用了CAS，這里需要有一個(gè)循環(huán)來不斷嘗試
for (;;) {
// 判斷線程池狀態(tài)邏輯
...
try {
// 獲取一個(gè)任務(wù)
// poll方法等待具體時(shí)間之后如果沒有獲取到對象，會(huì)返回null
// take方法會(huì)一直等到獲取新對象，除非被interrupt
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
// r==null,說明超時(shí)了，重新循環(huán)
timedOut = true;
} catch (InterruptedException retry) {
// 被interrupt，說明可能線程池被關(guān)閉了，重新判斷情況
timedOut = false;
}
}
}
這里需要重點(diǎn)關(guān)注的是阻塞隊(duì)列的 poll() 和 take() 方法，他們都會(huì)去隊(duì)列中獲取一個(gè)任務(wù)；但是，poll() 方法會(huì)阻塞指定時(shí)間后返回，而 take() 則是無限期阻塞。這里對應(yīng)的就是有存活時(shí)間的線程和不會(huì)被銷毀的核心線程。
同時(shí)注意 timedOut = true 是在這一部分被賦值的，當(dāng)賦值為true之后需要再執(zhí)行一次循環(huán)，在上面的判斷中就會(huì)被攔截下來并返回false，這在第一部分邏輯介紹了。而如果線程在等待的時(shí)候被 interrupt 了，說明線程池被關(guān)閉了，此時(shí)也會(huì)重走一次上面判斷狀態(tài)的邏輯。
到這里關(guān)于執(zhí)行的邏輯就講得差不多了，下面聊一聊線程池關(guān)閉以及worker結(jié)束的相關(guān)邏輯。
worker退出工作：processWorkerExit
前面已經(jīng)介紹 runWorker() 了方法，這個(gè)方法的主要任務(wù)就是讓worker動(dòng)起來，不斷去隊(duì)列獲取任務(wù)。而當(dāng)獲取任務(wù)的時(shí)候返回了null，則表示該worker可以結(jié)束了，最后會(huì)調(diào)用 processWorkerExit() 方法，如下：
final void runWorker(Worker w) {
...
try {
...
} finally {
// 如果worker退出，那么需要執(zhí)行后續(xù)的善后工作
processWorkerExit(w, completedAbruptly);
}
}
processWorkerExit() 會(huì)完成worker退出的善后工作。具體的內(nèi)容是：
- 把完成的任務(wù)數(shù)合并到總的任務(wù)數(shù),移除worker，嘗試設(shè)置線程池的狀態(tài)為terminated：
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// 如果不是經(jīng)過getTask方法返回null正常退出的，那么需要讓線程總數(shù)-1
// 這個(gè)參數(shù)前面一直讓你們注意一下不知道你們還記不記得
// 如果是在正常情況下退出，那么在getTask() 方法中就會(huì)執(zhí)行decrementWorkerCount()了
// 而如果出現(xiàn)一些特殊的情況突然結(jié)束了，并不是通過在getTask返回null結(jié)束
// Abruptly就是突然的意思，那么completedAbruptly就為true,正常情況下在runWorker方法中會(huì)被設(shè)置為false
// 那什么叫突然結(jié)束？用戶的任務(wù)拋出了異常，這個(gè)時(shí)候線程就突然結(jié)束了，沒有經(jīng)過getTask方法
// 這里就需要讓線程總數(shù)-1
if (completedAbruptly)
decrementWorkerCount();
// 獲取鎖，并累加完成的任務(wù)總數(shù)，從set中移除worker
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
// 嘗試設(shè)置線程池的狀態(tài)為terminated
// 這個(gè)方法前面我們addWorker失敗的時(shí)候提到過，后面再展開
tryTerminate();
...
}
- 移除worker之后，如果線程池還沒有被stop，那么最后必須保證隊(duì)列任務(wù)至少有一個(gè)線程在執(zhí)行隊(duì)列中的任務(wù)：
private void processWorkerExit(Worker w, boolean completedAbruptly) {
...
int c = ctl.get();
// stop及以上的狀態(tài)不需要執(zhí)行剩下的任務(wù)
if (runStateLessThan(c, STOP)) {
// 如果線程是突然終止的，那肯定需要重新創(chuàng)建一個(gè)
// 否則進(jìn)行判斷是否要保留一個(gè)線程
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return;
}
// 如果此時(shí)線程數(shù)<=核心線程數(shù)，或者當(dāng)核心線程可被銷毀時(shí)，線程數(shù)==0且隊(duì)列不為空
// 那么需要?jiǎng)?chuàng)建一個(gè)線程來執(zhí)行任務(wù)
addWorker(null, false);
}
}
代碼雖然看起來很多，但是具體的邏輯內(nèi)容還是比較簡單的。前面一直提到一個(gè)方法 tryTerminate() 但一直沒有展開解釋，下面來介紹一下。
嘗試終止線程池：tryTerminate()
這個(gè)方法出現(xiàn)在任何可能讓線程池進(jìn)入終止?fàn)顟B(tài)的地方。如添加worker失敗時(shí)，那么這個(gè)時(shí)候可能線程池已經(jīng)處于stop狀態(tài)，且已經(jīng)沒有任何正在執(zhí)行的worker了，那么此時(shí)可以進(jìn)入terminated狀態(tài)；再如worker被銷毀的時(shí)候，可能這是最后一個(gè)被銷毀的worker，那么此時(shí)線程池需要進(jìn)入terminated狀態(tài)。
根據(jù)這個(gè)方法的使用情況其實(shí)就已經(jīng)差不多可以推斷出這個(gè)方法的內(nèi)容：判斷當(dāng)前線程池的狀態(tài)，如果符合條件則設(shè)置線程池的狀態(tài)為terminated 。如果此時(shí)不能轉(zhuǎn)換為terminated狀態(tài)，則什么也不做，直接返回。
- 首先判斷當(dāng)前線程池狀態(tài)是否符合轉(zhuǎn)化為terminated。如果處于運(yùn)行狀態(tài)或者tidying以上狀態(tài)，則肯定不需要進(jìn)行狀態(tài)轉(zhuǎn)換。因?yàn)閞unning需要先進(jìn)入stop狀態(tài)，而tidying其實(shí)已經(jīng)是準(zhǔn)備進(jìn)入terminated狀態(tài)了。如果處于shutdown狀態(tài)且隊(duì)列不為空，那么需要執(zhí)行完隊(duì)列中的任務(wù)，所以也不適合狀態(tài)轉(zhuǎn)換：
final void tryTerminate() {
for (;;) {
int c = ctl.get();
// 如果處于運(yùn)行狀態(tài)或者tidying以上狀態(tài)時(shí)，直接返回，不需要修改狀態(tài)
// 如果處于stop以下狀態(tài)且隊(duì)列不為空，那么需要等隊(duì)列中的任務(wù)執(zhí)行完成，直接返回
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateLessThan(c, STOP) && ! workQueue.isEmpty()))
return;
// 到這里說明線程池肯定處于stop狀態(tài)
// 線程的數(shù)量不等于0，嘗試中斷一個(gè)空閑的worker線程
// 這里他只中斷workerSet中的其中一個(gè)，當(dāng)其中的一個(gè)線程停止時(shí)，會(huì)再次調(diào)用tryTerminate
// 然后又會(huì)再去中斷workerSet中的一個(gè)worker，不斷循環(huán)下去直到剩下最后一個(gè)，workercount==0
// 這就是 鏈?zhǔn)椒磻?yīng) 。
if (workerCountOf(c) != 0) {
interruptIdleWorkers(ONLY_ONE);
return;
}
// 設(shè)置狀態(tài)為terminated邏輯
...
}
}
- 經(jīng)過上面的判斷，能到第二部分邏輯，線程池肯定是具備進(jìn)入terminated狀態(tài)的條件了。剩下的代碼就是把線程池的狀態(tài)設(shè)置為terminated：
final void tryTerminate() {
for (;;) {
// 上一部分邏輯
...
// 首先獲取全局鎖
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 嘗試把線程池的狀態(tài)從stop修改為tidying
// 如果修改失敗，說明狀態(tài)已經(jīng)被修改了，那么外層循環(huán)再跑一個(gè)
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
// 這個(gè)方法是一個(gè)空實(shí)現(xiàn)，需要子類繼承重寫
terminated();
} finally {
// 最后再設(shè)置狀態(tài)為terminated
ctl.set(ctlOf(TERMINATED, 0));
// 喚醒所有等待終止鎖的線程
termination.signalAll();
}
return;
}
} finally {
// 釋放鎖
mainLock.unlock();
}
// CAS修改線程池的狀態(tài)失敗，重新進(jìn)行判斷
}
}
當(dāng)線程池被標(biāo)記為terminated狀態(tài)時(shí)，那么這個(gè)線程池就徹底地終止了。
好了到這里，恭喜你，關(guān)于ThreadPoolExecutor的源碼解析理解得差不多了。接下來剩下幾個(gè)常用的api方法：submit() 、 shutdown()/shutdownNow() 順便看一下吧，他們的邏輯也是都非常簡單。
關(guān)閉線程池：shutdown/shutdownNow
關(guān)閉線程池有兩個(gè)方法：
- shutdown：設(shè)置線程池的狀態(tài)為shutdown，同時(shí)嘗試中斷所有空閑線程，但是會(huì)等待隊(duì)列中的任務(wù)執(zhí)行結(jié)束再終止線程池。
- shutdownNow：設(shè)置線程池的狀態(tài)為stop，同時(shí)嘗試中斷所有空閑線程，不會(huì)等待隊(duì)列中的任務(wù)完成，正在執(zhí)行中的線程執(zhí)行結(jié)束，線程池馬上進(jìn)入terminated狀態(tài)。
我們各自來看一下：
// 關(guān)閉后隊(duì)列中的任務(wù)依舊會(huì)被執(zhí)行，但是不會(huì)再添加新的任務(wù)
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
// 設(shè)置狀態(tài)為shutdown
advanceRunState(SHUTDOWN);
// 嘗試中斷所有空閑的worker
interruptIdleWorkers();
// 回調(diào)方法，這個(gè)方法是個(gè)空方法，ScheduledThreadPoolExecutor中重寫了該方法
onShutdown();
} finally {
mainLock.unlock();
}
// 嘗試設(shè)置線程池狀態(tài)為terminated
tryTerminate();
}
再看一下另一個(gè)方法shutdownNow：
// 關(guān)閉后隊(duì)列中剩余的任務(wù)不會(huì)被執(zhí)行
// 會(huì)把剩下的任務(wù)返回交給開發(fā)者去處理
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 檢查是否可以關(guān)閉線程
checkShutdownAccess();
// 設(shè)置狀態(tài)為stop
advanceRunState(STOP);
// 嘗試中斷所有線程
interruptWorkers();
// 返回隊(duì)列中剩下的任務(wù)
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
最后再來看一下和 execute()不同的提交任務(wù)方法：submit。
提交任務(wù)：submit()
submit方法并不是ThreadPoolExecutor實(shí)現(xiàn)的，而是AbstractExecutorService，如下：
// runnable沒有返回值，創(chuàng)建FutureTask的返回參數(shù)傳入null
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
// 有參數(shù)返回值的runnable
// 最終也是構(gòu)造一個(gè)callable來執(zhí)行，把返回值設(shè)置為result
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
// callable本身就擁有返回值
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
他們的邏輯都幾乎一樣：調(diào)用newTaskFor方法來構(gòu)造一個(gè)Future對象并返回。我們看到newTaskFor方法：
// 創(chuàng)建一個(gè)FutureTask來返回
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
可以看到這個(gè)方法很簡單：構(gòu)造一個(gè)FutureTask并返回，F(xiàn)utureTask也是Future接口目前唯一的實(shí)現(xiàn)類。
更加具體關(guān)于Future的內(nèi)容就不展開了，有興趣的讀者可以去了解一下。
最后
好了到這里，關(guān)于ThreadPoolExecutor的源碼分析內(nèi)容就講完了。最后讓我們再回顧一下吧：
- ThreadPoolExecutor的整個(gè)執(zhí)行流程從execute方法開始，他會(huì)根據(jù)具體的情況，采用合適的執(zhí)行方案
- 線程被封裝在worker對象中，worker對象通過runWorker方法，會(huì)一直不斷地調(diào)用getTask方法來調(diào)用隊(duì)列的poll或take方法獲取任務(wù)
- 當(dāng)需要退出一個(gè)worker時(shí)，只要getTask方法返回null即可退出
- 當(dāng)線程池關(guān)閉時(shí)，會(huì)根據(jù)不同的關(guān)閉方法，等待所有的線程執(zhí)行完成，然后關(guān)閉線程池。
線程池整體的模型和handler是十分類似的：一個(gè)生產(chǎn)者-消費(fèi)者模型。但和Handler不同的是，ThreadPoolExecutor不支持延時(shí)任務(wù)，這點(diǎn)在ScheduledThreadPoolExecutor得到了實(shí)現(xiàn)；Handler的線程安全采用synchronize關(guān)鍵字，而ThreadPoolExecutor采用的是Lock和一些利用CAS實(shí)現(xiàn)線程安全的整型變量；Handler無法拒絕任務(wù)，線程池可以；Handler拋出異常會(huì)直接程序崩潰，而線程池不會(huì)等等。
了解了線程池的內(nèi)部源碼，對于他更加了解后，那么可以根據(jù)具體的問題，做出更加合適的解決方案。ThreadPoolExecutor還有一些源碼沒有講到，以及ScheduledThreadPoolExecutor、阻塞隊(duì)列的源碼，有興趣讀者可以自行去深入了解一下，拓展關(guān)于線程池的一切。
全文到此，假期肝文不容易啊，如果文章對你有幫助，求一個(gè)大拇指 ，贊一下再走唄。
參考文獻(xiàn)
- 《Java并發(fā)編程的藝術(shù)》：并發(fā)編程必讀，作者對一些原理講的很透徹
- 《Java核心技術(shù)卷》：這系列的書主要是講解框架的使用，不會(huì)深入原理，適合入門
- javaGuide：javaGuide，對java知識總結(jié)得很不錯(cuò)的一個(gè)博客
- Java并發(fā)編程：線程池的使用：博客園上一位很優(yōu)秀的博主，文章寫得通俗易懂且不失深度
全文到此，原創(chuàng)不易，覺得有幫助可以點(diǎn)贊收藏評論轉(zhuǎn)發(fā)。
筆者才疏學(xué)淺，有任何想法歡迎評論區(qū)交流指正。
如需轉(zhuǎn)載請?jiān)u論區(qū)或私信交流。
另外歡迎光臨筆者的個(gè)人博客：傳送門
|
