開篇-為什么要使用線程池？
? Java 中的線程池是運(yùn)用場景最多的并發(fā)框架，幾乎所有需要異步或并發(fā)執(zhí)行任務(wù)的程序都可以使用線程池。在開發(fā)過程中，合理地使用線程池能夠帶來 3 個好處。
? 第一：降低資源消耗。通過重復(fù)利用已創(chuàng)建的線程降低線程創(chuàng)建和銷毀造成的消耗。
? 第二：提高響應(yīng)速度。當(dāng)任務(wù)到達(dá)時，任務(wù)可以不需要等到線程創(chuàng)建就能立即執(zhí)行。
? 第三：提高線程的可管理性。線程是稀缺資源，如果無限制地創(chuàng)建，不僅會消耗系統(tǒng)資源，還會降低系統(tǒng)的穩(wěn)定性，使用線程池可以進(jìn)行統(tǒng)一分配、調(diào)優(yōu)和監(jiān)控。但是，要做到合理利用線程池，必須對其實(shí)現(xiàn)原理了如指掌。
1. 線程池的任務(wù)執(zhí)行步驟
? 當(dāng)向線程池提交一個任務(wù)之后，線程池是如何處理這個任務(wù)的呢？處理流程圖如圖 1-1所示。
? 從圖中可以看出，當(dāng)提交一個新任務(wù)到線程池時，線程池的處理流程如下。
? 1.線程池判斷核心線程池里的線程是否都在執(zhí)行任務(wù)。如果不是，則創(chuàng)建一個新的工作線程來執(zhí)行任務(wù)。如果核心線程池里的線程都在執(zhí)行任務(wù)，則進(jìn)入下個流程。
? 2.線程池判斷工作隊(duì)列是否已經(jīng)滿。如果工作隊(duì)列沒有滿，則將新提交的任務(wù)存儲在這個工作隊(duì)列里。如果工作隊(duì)列滿了，則進(jìn)入下個流程。
? 3.線程池判斷線程池的線程是否都處于工作狀態(tài)。如果沒有，則創(chuàng)建一個新的工作線程來執(zhí)行任務(wù)。如果已經(jīng)滿了，則交給飽和策略來處理這個任務(wù)。
? ThreadPoolExecutor 執(zhí)行 execute()方法的示意圖，如圖 1-2 所示。
-
如果當(dāng)前運(yùn)行的線程少于
corePoolSize，則創(chuàng)建新線程來執(zhí)行任務(wù)（注意，執(zhí)行這一步驟需要獲取全局鎖）。 -
如果運(yùn)行的線程等于或多于
corePoolSize，則將任務(wù)加入BlockingQueue。 -
如果無法將任務(wù)加入
BlockingQueue（隊(duì)列已滿），則創(chuàng)建新的線程來處理任務(wù)（注意，執(zhí)行這一步驟需要獲取全局鎖）。 -
如果創(chuàng)建新線程將使當(dāng)前運(yùn)行的線程超出
maximumPoolSize，任務(wù)將被拒絕，并調(diào)用RejectedExecutionHandler.rejectedExecution()方法。ThreadPoolExecutor采取上述步驟的總體設(shè)計(jì)思路，是為了在執(zhí)行 execute()方法時，盡可能地避免獲取全局鎖（那將會是一個嚴(yán)重的可伸縮瓶頸）。在ThreadPoolExecutor完成預(yù)熱之后（當(dāng)前運(yùn)行的線程數(shù)大于等于corePoolSize），幾乎所有的 execute()方法調(diào)用都是執(zhí)行步驟 2，而步驟 2 不需要獲取全局鎖。
2. 線程池的使用
2.1 線程池的創(chuàng)建
通過ThreadPoolExecutor可以創(chuàng)建一個線程池。
ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
創(chuàng)建一個線程池時需要輸入幾個參數(shù)，如下。
-
corePoolSize（線程池的基本大小,也可以稱之為核心線程數(shù)大小）：當(dāng)提交一個任務(wù)到線程池時，線程池會創(chuàng)建一個線程來執(zhí)行任務(wù)，即使其他空閑的基本線程能夠執(zhí)行新任務(wù)也會創(chuàng)建線程，等到需要執(zhí)行的任務(wù)數(shù)大于線程池基本大小時就不再創(chuàng)建。如果調(diào)用了線程池的prestartAllCoreThreads()方法，線程池會提前創(chuàng)建并啟動所有基本線程。 -
maximumPoolSize（線程池最大數(shù)量）：線程池允許創(chuàng)建的最大線程數(shù)。如果隊(duì)列滿了，并且已創(chuàng)建的線程數(shù)小于最大線程數(shù)，則線程池會創(chuàng)建新的線程執(zhí)行任務(wù)。值得注意的是，如果使用了無界的任務(wù)隊(duì)列，這個參數(shù)就沒有什么效果了。 -
keepAliveTime（線程活動保持時間）：線程池的工作線程空閑后，保持存活的時間。所以，如果任務(wù)很多，并且每個任務(wù)執(zhí)行的時間比較短，可以調(diào)大時間，提高線程的利用率。默認(rèn)情況下該參數(shù)針對的是非核心線程，如果將參數(shù)allowCoreThreadTimeOut設(shè)置為true，那么核心線程也會受這個參數(shù)影響。 -
TimeUnit（線程活動保持時間的單位）：可選的單位有天（DAYS）、小時（HOURS）、分鐘（MINUTES）、毫秒（MILLISECONDS）、微秒（MICROSECONDS，千分之一毫秒）和納秒（NANOSECONDS，千分之一微秒）。 -
workQueue（任務(wù)隊(duì)列）：用于保存等待執(zhí)行的任務(wù)的阻塞隊(duì)列。具體可以參考Java阻塞隊(duì)列。 -
threadFactory：用于設(shè)置創(chuàng)建線程的工廠，可以通過線程工廠給每個創(chuàng)建出來的線程設(shè)置更有意義的名字。 -
RejectedExecutionHandler（飽和策略）：當(dāng)隊(duì)列和線程池都滿了，說明線程池處于飽和狀態(tài)，那么必須采取一種策略處理提交的新任務(wù)。這個策略默認(rèn)情況下是AbortPolicy，表示無法處理新任務(wù)時拋出異常。在 JDK 1.5 中 Java 線程池框架提供了以下 4 種策略。?
AbortPolicy：直接拋出異常。?
CallerRunsPolicy：只用調(diào)用者所在線程來運(yùn)行任務(wù)。?
DiscardOldestPolicy：丟棄隊(duì)列里最近的一個任務(wù)，并執(zhí)行當(dāng)前任務(wù)。?
DiscardPolicy：不處理，丟棄掉。當(dāng)然，也可以根據(jù)應(yīng)用場景需要來實(shí)現(xiàn) RejectedExecutionHandler 接口自定義策略。如記錄日志或持久化存儲不能處理的任務(wù)。
2.2 向線程池中提交任務(wù)
? 可以使用兩個方法向線程池提交任務(wù)，分別為 execute()和 submit()方法。
? execute()方法用于提交不需要返回值的任務(wù)，所以無法判斷任務(wù)是否被線程池執(zhí)行成功。通過以下代碼可知 execute()方法輸入的任務(wù)是一個 Runnable 類的實(shí)例。
executorService.execute(()-> Thread.currentThread().getName());
? submit()方法用于提交需要返回值的任務(wù)。線程池會返回一個 future 類型的對象，通過這個 future 對象可以判斷任務(wù)是否執(zhí)行成功，并且可以通過 future 的 get()方法來獲取返回值，get()方法會阻塞當(dāng)前線程直到任務(wù)完成，而使用 get（long timeout，TimeUnit unit）方法則會阻塞當(dāng)前線程一段時間后立即返回，這時候有可能任務(wù)沒有執(zhí)行完。
Future<String> submit = executorService.submit(() -> Thread.currentThread().getName());
System.out.println(submit.get());
2.3 關(guān)閉線程池
? 可以通過調(diào)用線程池的 shutdown 或 shutdownNow 方法來關(guān)閉線程池。它們的原理是遍歷線程池中的工作線程，然后逐個調(diào)用線程的 interrupt 方法來中斷線程，所以無法響應(yīng)中斷的任務(wù)可能永遠(yuǎn)無法終止。但是它們存在一定的區(qū)別，shutdownNow 首先將線程池的狀態(tài)設(shè)置成 STOP，然后嘗試停止所有的正在執(zhí)行或暫停任務(wù)的線程，并返回等待執(zhí)行任務(wù)的列表，而 shutdown 只是將線程池的狀態(tài)設(shè)置成 SHUTDOWN 狀態(tài)，然后中斷所有沒有正在執(zhí)行任務(wù)的線程。
? 只要調(diào)用了這兩個關(guān)閉方法中的任意一個，isShutdown 方法就會返回 true。當(dāng)所有的任務(wù)都已關(guān)閉后，才表示線程池關(guān)閉成功，這時調(diào)用 isTerminaed 方法會返回 true。至于應(yīng)該調(diào)用哪一種方法來關(guān)閉線程池，應(yīng)該由提交到線程池的任務(wù)特性決定，通常調(diào)shutdown 方法來關(guān)閉線程池，如果任務(wù)不一定要執(zhí)行完，則可以調(diào)用shutdownNow方法。
2.4 合理地配置線程池
? 要想合理地配置線程池，就必須首先分析任務(wù)特性，可以從以下幾個角度來分析。
? ? 任務(wù)的性質(zhì)：CPU 密集型任務(wù)、IO 密集型任務(wù)和混合型任務(wù)。
? ? 任務(wù)的優(yōu)先級：高、中和低。
? ? 任務(wù)的執(zhí)行時間：長、中和短。
? ? 任務(wù)的依賴性：是否依賴其他系統(tǒng)資源，如數(shù)據(jù)庫連接。
? 性質(zhì)不同的任務(wù)可以用不同規(guī)模的線程池分開處理。CPU 密集型任務(wù)應(yīng)配置盡可能小的線程，如配置 N+1（其中N是CPU合適）個線程的線程池。由于 IO 密集型任務(wù)線程并不是一直在執(zhí)行任務(wù)，則應(yīng)配置盡可能多的線程，如 2*N（其中N是CPU合適）。混合型的任務(wù)，如果可以拆分，將其拆分成一個 CPU 密集型任務(wù)和一個 IO 密集型任務(wù)，只要這兩個任務(wù)執(zhí)行的時間相差不是太大，那么分解后執(zhí)行的吞吐量將高于串行執(zhí)行的吞吐量。如果這兩個任務(wù)執(zhí)行時間相差太大，則沒必要進(jìn)行分解。可以通過 Runtime.getRuntime().availableProcessors()方法獲得當(dāng)前設(shè)備的 CPU 個數(shù)。
? 優(yōu)先級不同的任務(wù)可以使用優(yōu)先級隊(duì)列 PriorityBlockingQueue 來處理。它可以讓優(yōu)先級高的任務(wù)先執(zhí)行。
? 執(zhí)行時間不同的任務(wù)可以交給不同規(guī)模的線程池來處理，或者可以使用優(yōu)先級隊(duì)列，讓執(zhí)行時間短的任務(wù)先執(zhí)行。
? 依賴數(shù)據(jù)庫連接池的任務(wù)，因?yàn)榫€程提交 SQL 后需要等待數(shù)據(jù)庫返回結(jié)果，等待的時間越長，則 CPU 空閑時間就越長，那么線程數(shù)應(yīng)該設(shè)置得越大，這樣才能更好地利用CPU。
? 建議使用有界隊(duì)列。有界隊(duì)列能增加系統(tǒng)的穩(wěn)定性和預(yù)警能力，可以根據(jù)需要設(shè)大一點(diǎn)兒，比如幾千。有一次，我們系統(tǒng)里后臺任務(wù)線程池的隊(duì)列和線程池全滿了，不斷拋出拋棄任務(wù)的異常，通過排查發(fā)現(xiàn)是數(shù)據(jù)庫出現(xiàn)了問題，導(dǎo)致執(zhí)行 SQL 變得非常緩慢，因?yàn)楹笈_任務(wù)線程池里的任務(wù)全是需要向數(shù)據(jù)庫查詢和插入數(shù)據(jù)的，所以導(dǎo)致線程池里的工作線程全部阻塞，任務(wù)積壓在線程池里。如果當(dāng)時我們設(shè)置成無界隊(duì)列，那么線程池的隊(duì)列就會越來越多，有可能會撐滿內(nèi)存，導(dǎo)致整個系統(tǒng)不可用，而不只是后臺任務(wù)出現(xiàn)問題。當(dāng)然，我們的系統(tǒng)所有的任務(wù)是用單獨(dú)的服務(wù)器部署的，我們使用不同規(guī)模的線程池完成不同類型的任務(wù)，但是出現(xiàn)這樣問題時也會影響到其他任務(wù)。
2.5 線程池的監(jiān)控
? 如果在系統(tǒng)中大量使用線程池，則有必要對線程池進(jìn)行監(jiān)控，方便在出現(xiàn)問題時，可以根據(jù)線程池的使用狀況快速定位問題。可以通過線程池提供的參數(shù)進(jìn)行監(jiān)控，在監(jiān)控線程池的時候可以使用以下屬性。
? ? taskCount：線程池需要執(zhí)行的任務(wù)數(shù)量。
? ? completedTaskCount：線程池在運(yùn)行過程中已完成的任務(wù)數(shù)量，小于或等于taskCount。
? ? largestPoolSize：線程池里曾經(jīng)創(chuàng)建過的最大線程數(shù)量。通過這個數(shù)據(jù)可以知道線程池是否曾經(jīng)滿過。如該數(shù)值等于線程池的最大大小，則表示線程池曾經(jīng)滿過。
? ? getPoolSize：線程池的線程數(shù)量。如果線程池不銷毀的話，線程池里的線程不會自動銷毀，所以這個大小只增不減。
? ? getActiveCount：獲取活動的線程數(shù)。
? 通過擴(kuò)展線程池進(jìn)行監(jiān)控。可以通過繼承線程池來自定義線程池，重寫線程池的beforeExecute、afterExecute 和 terminated 方法，也可以在任務(wù)執(zhí)行前、執(zhí)行后和線程池關(guān)閉前執(zhí)行一些代碼來進(jìn)行監(jiān)控。例如，監(jiān)控任務(wù)的平均執(zhí)行時間、最大執(zhí)行時間和最小執(zhí)行時間等。
3.線程池的生命周期
線程池的生命周期包括以下幾個狀態(tài)：
-
初始狀態(tài)（
NEW）：線程池被創(chuàng)建后處于初始狀態(tài)。此時線程池沒有包含任何線程，也沒有開始執(zhí)行任務(wù)。 -
運(yùn)行狀態(tài)（
RUNNING）：通過調(diào)用線程池的execute()或submit()方法，線程池開始接受任務(wù)并創(chuàng)建線程執(zhí)行。線程池可以動態(tài)地調(diào)整線程數(shù)量來適應(yīng)任務(wù)的需求。 -
關(guān)閉狀態(tài)（
SHUTDOWN）：當(dāng)調(diào)用線程池的shutdown()方法后，線程池進(jìn)入關(guān)閉狀態(tài)。此時線程池不會再接受新的任務(wù)提交，但會繼續(xù)處理已經(jīng)提交的任務(wù)直到完成。處于關(guān)閉狀態(tài)的線程池仍然可以調(diào)用execute()方法來提交任務(wù)，但會拋出RejectedExecutionException。 -
停止?fàn)顟B(tài)（
STOP）：通過調(diào)用線程池的shutdownNow()方法可以使線程池進(jìn)入停止?fàn)顟B(tài)。此時線程池會立即停止，取消所有正在執(zhí)行的任務(wù)，并且丟棄所有等待執(zhí)行的任務(wù)。 -
整理狀態(tài)（
TIDYING）：當(dāng)線程池處于STOP狀態(tài)或者SHUTDOWN后，并且所有任務(wù)都已經(jīng)完成，線程池會進(jìn)入整理狀態(tài)。在整理狀態(tài)中，線程池會清理已終止的工作線程。當(dāng)線程池變?yōu)門IDYING狀態(tài)時，會執(zhí)行鉤子函數(shù)terminated()。 -
終止?fàn)顟B(tài)（
TERMINATED）：當(dāng)線程池完成整理操作后，最終進(jìn)入終止?fàn)顟B(tài)。此時線程池徹底終止，不再接受任務(wù)和執(zhí)行任務(wù)。
注意，線程池的狀態(tài)可以通過isShutdown() 和 isTerminated() 方法進(jìn)行查詢，以確定線程池當(dāng)前所處的狀態(tài)。
4.代碼分析線程池的運(yùn)行原理
4.1 線程池控制狀態(tài)ctl
? ctl 是線程池源碼中常常用到的一個變量，它的主要作用是記錄線程池的生命周期狀態(tài)和當(dāng)前工作的線程數(shù)。它是一個原子整型變量。ctl是一個32位的整數(shù)，高3位用于表示線程池的運(yùn)行狀態(tài)，低29位用于表示線程池中的線程數(shù)量。具體的結(jié)構(gòu)如下所示：
31-29 | 線程池運(yùn)行狀態(tài)（用來保存線程池的狀態(tài) RUNNING，SHUTDOWN，STOP，STOP，STOP）
28-0 | 線程池中線程數(shù)量
ctl在ThreadPoolExecutor中的聲明和初始化
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
ctl(線程池控制狀態(tài))是原子整型的，這意味這對它進(jìn)行的操作具有原子性。- 如此一來，作為
ctl組成部分的runState(線程池生命周期狀態(tài))和workerCount(工作線程數(shù)) 也將同時具有原子性。 ThreadPoolExecutor使用ctlOf方法來將runState和workerCount兩個變量(都是整型)打包成一個ctl變量。
4.1.1 工具人常量COUNT_BITS 和 CAPACITY
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
Integer.SIZE為整型的位數(shù)32位 ,COUNT_BITS位32 - 3 = 29。- (1 << COUNT_BITS) - 1 的計(jì)算過程如下：
int 類型的1用二進(jìn)制表示為：00000000 00000000 00000000 00000001
1 << COUNT_BITS，其中COUNT_BITS為29，1 << 29,代表將數(shù)字1向左移動29位。結(jié)果為：00100000 00000000 00000000 00000000
將上一步計(jì)算出的結(jié)果減1，最終CAPACITY用二進(jìn)制表示為：00011111 11111111 11111111 11111111
- 因此在接下來的代碼中， COUNT_BITS 就用來表示分隔
runState和workerCount的位數(shù)。 - 而CAPACITY 則作為取變量
runState和workerCount的工具。
4.1.2 線程池的生命周期常量
// 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;
代碼中的五個變量分別對應(yīng)著線程池中的五個生命周期，他們的二進(jìn)制表示如下：
RUNNING 二進(jìn)制為 11100000 00000000 00000000 00000000
SHUTDOWN 二進(jìn)制為 00000000 00000000 00000000 00000000
STOP 二進(jìn)制為 00100000 00000000 00000000 00000000
TIDYING 二進(jìn)制為 01000000 00000000 00000000 00000000
TERMINATED 二進(jìn)制為 01100000 00000000 00000000 00000000
通過這里可以看出為什么將COUNT_BITS 設(shè)置為32 - 3，而不是將4或者其它的數(shù)。目的是將32位的高三位用來表示線程池的狀態(tài)，后29位用來表示線程池的最大容量。
4.1.3 方法runStateOf，workerCountOf,ctlOf
// Packing and unpacking ctl
private static int runStateOf(int c) { return c & ~CAPACITY; }
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
runStateOf方法用來獲取當(dāng)前線程池的狀態(tài)，~CAPACITY就是11100000 00000000 00000000 00000000。再通過ctl與其進(jìn)行&運(yùn)算就可以獲取高三位的值，獲取線程池的狀態(tài)。workerCountOf方法通過將ctl與CAPACITY&運(yùn)算可以計(jì)算出線程池中的工作線程數(shù)。ctlOf方法runState和workerCount進(jìn)行 | (按位或)操作來得到ctl變量，就是因?yàn)?runState的高 3 位為有效信息，而workerCount的低 29 位為有效信息，合起來正好得到一個含 32 位有效信息的整型變量。
4.2 源碼分析線程池的執(zhí)行流程
4.2.1 ThreadPoolExecutor.execute方法
execute用于向線程池中提交任務(wù)
方法源碼如下：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
// 獲取當(dāng)前線程池的控制狀態(tài)ctl。
int c = ctl.get();
// workerCountOf(c)獲取工作線程數(shù)，判斷工作線程數(shù)是否小于核心線程數(shù)
if (workerCountOf(c) < corePoolSize) {
// 工作線程數(shù)小于核心線程數(shù)，調(diào)用addWorker(command, true) 添加核心線程并且執(zhí)行任務(wù)，放回true直接結(jié)束方法。addWorker方法見4.2.2中的代碼清單4-2
if (addWorker(command, true))
return;
// addWorker(command, true) 調(diào)用失敗，更新ctl的臨時變量c,重新獲取線程池控制狀態(tài).
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
// 如果線程池是Running狀態(tài)，并且往阻塞隊(duì)列中添加任務(wù)完成。進(jìn)入這里
// 考慮并發(fā)帶來的影響，可能往同步隊(duì)列中添加元素之前，線程池被調(diào)用了shutDown，或者shutDownNow等方法想關(guān)閉線程池。這里再一次的確定線程池的狀態(tài)。如果還是Running，那代表線程池的狀態(tài)沒變，一直是Running。
int recheck = ctl.get();
// 線程池的狀態(tài)變動了，調(diào)用remove方法從隊(duì)列中移除任務(wù)。
if (! isRunning(recheck) && remove(command))
// 拒絕任務(wù)
reject(command);
// 走到這里，代表1.往同步隊(duì)列中添加元素成功，并且2.線程池的狀態(tài)一直是Running狀態(tài)。
else if (workerCountOf(recheck) == 0)
// 到了這里，阻塞隊(duì)列中有任務(wù)了，但是工作線程數(shù)為0，這種情況一般是臨界點(diǎn)，舉個例子，線程池中創(chuàng)建時，corePoolSize為 1，maximumPoolSize為 1，阻塞隊(duì)列使用無界隊(duì)列，再調(diào)用此excute方法之前，線程池一直是有個線程A在執(zhí)行任務(wù)的，此時調(diào)用完workQueue.offer(command)往隊(duì)列中添加元素之后，突然，執(zhí)行任務(wù)的線程A突然遇到了一個空指針異常，線程A走終止邏輯。這是線程池中就需要一個新的線程來繼續(xù)執(zhí)行阻塞隊(duì)列中的任務(wù)了。線程A終止的時候會調(diào)用processWorkerExit方法，該方法中也有addWorker(null, false);這個代碼片段。
addWorker(null, false);
}
// 走到這里說明，線程池中的工作線程總數(shù)已經(jīng)大于等于corePoolSize；線程池不是RUNNING狀態(tài)，或者說線程池是RUNNING狀態(tài)但無法繼續(xù)往隊(duì)列中添加元素了，這時候就需要創(chuàng)建非核心線程去執(zhí)行任務(wù)了。
else if (!addWorker(command, false))
// 創(chuàng)建非核心任務(wù)執(zhí)行
reject(command);
}
4.2.2 addWorker 方法
addWorker被用于向線程池添加工作線程，該方法的用于將線程池的工作線程數(shù)的變量加1，然后創(chuàng)建一個線程并且運(yùn)行。
private boolean addWorker(Runnable firstTask, boolean core) {
// for循環(huán)，這一層的功能用于將線程池的工作線程數(shù)變量workerCountOf加1
retry:
for (;;) {
// 獲取線程池控制狀態(tài)ctl
int c = ctl.get();
// 獲取線程池的狀態(tài)
int rs = runStateOf(c);
// Check if queue empty only if necessary.
// rs >= SHUTDOWN 的狀態(tài)有SHUTDOWN,STOP,TIDYING,TERMINATED 代表線程池不在接收新的任務(wù)了
// ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()) 該方法要返回true，里面的3個判斷條件至少有一個為false，rs == SHUTDOWN 判斷線程狀態(tài)為SHUTDOWN，firstTask == null 代表創(chuàng)建工作線程時指定的任務(wù)為null，這種情況是線程池主動調(diào)用addWorker方法才會出現(xiàn)的，比如代碼清單4-1 execute方法 代碼里面的代碼片段addWorker(null, false)，接著是 ! workQueue.isEmpty() 判斷工作隊(duì)列是否為空。如果上述條件判斷中的任何條件為真，就表示不能滿足添加工作線程的條件，會返回 false。這意味著在rs >= SHUTDOWN，只有當(dāng)線程池狀態(tài)為 SHUTDOWN、沒有待執(zhí)行的任務(wù)并且工作隊(duì)列不為空時，才能添加工作線程，添加線程去執(zhí)行同步隊(duì)列中剩下的任務(wù)。
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
// 獲取當(dāng)前線程池的工作線程數(shù)
int wc = workerCountOf(c);
// wc >= CAPACITY 返回true代表線程池中的線程已經(jīng)>=2^29-1（CAPACITY 設(shè)定最大值為2^29-1，CAPACITY用來表示線程池工作線程的數(shù)量）,wc >= (core ? corePoolSize : maximumPoolSize) 返回true代表當(dāng)前要添加類型的線程數(shù)已經(jīng)滿了。
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 將工作線程數(shù)加1
if (compareAndIncrementWorkerCount(c))
// 成功就跳出外層循環(huán)retry
break retry;
// CAS失敗重新獲取ctl的值
c = ctl.get(); // Re-read ctl
// 檢查重新讀取的runStateOf(c)是否與之前的狀態(tài)rs不同，如果不同則繼續(xù)外層循環(huán)retry
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
// 下面的代碼用于往線程池中添加工作線程。
// workerStarted 工作線程啟動狀態(tài)位，true代表啟動成功。
boolean workerStarted = false;
// workerAdded 工作線程添加標(biāo)志位，true代表添加成功。
boolean workerAdded = false;
Worker w = null;
try {
// 創(chuàng)建一個Worker，詳細(xì)見4.2.3 代碼清單4-3
w = new Worker(firstTask);
// 獲得工作線程
final Thread t = w.thread;
if (t != null) {
// 加鎖，防止以下操作并發(fā)
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int rs = runStateOf(ctl.get());
// rs < SHUTDOWN 如果線程池的狀態(tài)小于SHUTDOWN，這里可以取到的狀態(tài)枚舉有RUNNING，rs == SHUTDOWN && firstTask == null，這段返回true說明線程池的狀態(tài)雖然是SHUTDOWN狀態(tài)，但需要處理阻塞隊(duì)列中剩下的任務(wù)。
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
// 如果線程t是活動的拋出異常。這邊還沒有啟動線程t，線程t就已經(jīng)活動了，說明有問題。
throw new IllegalThreadStateException();
// 將worker放入workers，workers是一個HashSet,用來存放所有的工作線程Worker。
workers.add(w);
int s = workers.size();
// largestPoolSize，曾經(jīng)出現(xiàn)過的最大線程數(shù)，如果s > largestPoolSize，重新賦值，更新線程池峰值
if (s > largestPoolSize)
largestPoolSize = s;
// 工作線程添加成功。
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
// 開啟工作線程。工作線程執(zhí)行邏輯見代碼4.2.3 代碼清單4-3
t.start();
// 工作線程開啟成功
workerStarted = true;
}
}
} finally {
if (! workerStarted)
// 工作線程啟動失敗，見如下的代碼片段1
addWorkerFailed(w);
}
return workerStarted;
}
// 代碼片段1 java.util.concurrent.ThreadPoolExecutor#addWorkerFailed
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 從工作線程w從工作線程集合中移除
if (w != null)
workers.remove(w);
// 線程工作workerCount數(shù)量減1
decrementWorkerCount();
// 基于狀態(tài)判斷嘗試終結(jié)線程池，下面的小結(jié)會分析該方法
tryTerminate();
} finally {
mainLock.unlock();
}
}
4.2.3 runWorker方法
addWorker方法 完成了工作線程創(chuàng)建和啟動。addWorker方法中的t.start()方法調(diào)用完成后會調(diào)用Worker的run方法，然后調(diào)用runWorker方法來進(jìn)行工作。
Worker(Runnable firstTask) {
// 設(shè)置同步狀態(tài)為 -1，禁止線程中斷，直到runWorker()方法執(zhí)行，結(jié)合java.util.concurrent.ThreadPoolExecutor.Worker#interruptIfStarted 方法可以知道原因，interruptIfStarted放在代碼清單 4-3。
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// Worker 實(shí)現(xiàn)了Runnable接口，將Worker自身傳入到了newThread中，到時候線程啟動時調(diào)用Worker.run()。
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this);
}
# java.util.concurrent.ThreadPoolExecutor#runWorker
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
// 獲取當(dāng)前工作線程要執(zhí)行的任務(wù)
Runnable task = w.firstTask;
// 將firstTask置為null，表示任務(wù)被取出
w.firstTask = null;
// 因?yàn)槌跏蓟疻orker的時候?qū)⒕€程標(biāo)志設(shè)置為了-1，這里把state更新為0，允許線程中斷
w.unlock(); // allow interrupts
// 記錄線程是否因?yàn)橛脩舢惓＝K結(jié)，默認(rèn)是true
boolean completedAbruptly = true;
try {
// 一個循環(huán),當(dāng)任務(wù)非空，或者通過getTask()方法可以從同步隊(duì)列中獲取任務(wù)時進(jìn)入循環(huán)，如果while命中后半段當(dāng)前線程會處于阻塞或者超時阻塞狀態(tài)。關(guān)于getTask方法見4.2.4 代碼清單4-4.
while (task != null || (task = getTask()) != null) {
// 可能會疑問這一步為什么要上鎖，這塊根本不存在并發(fā)啊！！！其實(shí)是線程池在執(zhí)行shutdown等方法時會調(diào)用interruptIdleWorkers方法來中斷空閑的線程，interruptIdleWorkers方法會使用tryLock方法來判斷線程池中的線程是否是空閑狀態(tài)；獲取不了鎖說明線程正在執(zhí)行任務(wù)，否則就是處于等待任務(wù)或者空閑狀態(tài)，可以被中斷。
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
// 這個條件判斷工作線程是否需要中斷。如果線程池處于STOP,TIDYING,TERMINATED狀態(tài)，或者當(dāng)前調(diào)用該方法的線程已經(jīng)被中斷并且線程池處于STOP，TIDYING，TERMINATED狀態(tài)，且當(dāng)前工作線程未被中斷（如果isInterrupted返回true，代表已被中斷，那就沒必要再調(diào)用wt.interrupt()中斷一次了），則中斷當(dāng)前工作線程，這里的中斷僅僅是將中斷標(biāo)志位設(shè)為true了，具體的中斷邏輯還需要用戶實(shí)現(xiàn)。
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
// 鉤子函數(shù)，執(zhí)行任務(wù)前的準(zhǔn)備工作，開發(fā)者可以去重寫他的實(shí)現(xiàn)。
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) {
thrown = x; throw new Error(x);
} finally {
// 鉤子函數(shù)，任務(wù)執(zhí)行完畢時調(diào)用，開發(fā)者可以去重新他的實(shí)現(xiàn)
afterExecute(task, thrown);
}
} finally {
// 將臨時表量task置為null
task = null;
// 累加Worker完成的任務(wù)數(shù)
w.completedTasks++;
// 解鎖
w.unlock();
}
}
// 走到這里說明getTask()返回為null，線程正常退出，移除的話就直接調(diào)用下面的finally了
completedAbruptly = false;
} finally {
// 處理線程退出，completedAbruptly為true說明工作線程非正常退出，false表示正常退出。processWorkerExit見4.2.5 代碼清單4-5
processWorkerExit(w, completedAbruptly);
}
}
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
4.2.4 getTask方法
工作線程嘗試從隊(duì)列中獲取任務(wù)
private Runnable getTask() {
// 上一次調(diào)用poll從隊(duì)列中拉取任務(wù)是否超時，默認(rèn)false，沒超時。該狀態(tài)用于配合線程池的超時退出機(jī)制使用。
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
// 判斷：
// 1.rs >= SHUTDOWN，線程池狀態(tài)至少是SHUTDOWN
// 2.在滿足條件1的情況下，rs >= STOP 線程池的狀態(tài)已經(jīng)至少是stop了，stop狀態(tài)不需要繼續(xù)處理隊(duì)列中的任務(wù)了；workQueue.isEmpty()說明隊(duì)列已經(jīng)空了，說明工作線程可以退出了。
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
// 參數(shù)allowCoreThreadTimeOut是用來控制核心線程是否需要超時退出的一個變量，默認(rèn)情況下該變量為false（表示核心線程不需要超時退出），wc > corePoolSize 說明有非核心線程加入了線程池，這些線程是需要超時退出的。
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// wc > maximumPoolSize 當(dāng)前的工作線程總數(shù)大于maximumPoolSize，說明了通過setMaximumPoolSize()方法減少了線程池容量
// timed && timedOut 返回true 說明了線程是需要超時退出的，并且上一輪循環(huán)通過poll()方法從任務(wù)隊(duì)列中拉取任務(wù)為null
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
// cas WorkerCount - 1，成功返回null，失敗繼續(xù)下一輪。
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
// 如果timed為true，通過poll()方法做超時拉取，keepAliveTime時間內(nèi)沒有等待到有效的任務(wù)，則返回null，如果timed為false，通過take()做阻塞拉取，會阻塞到有下一個有效的任務(wù)時候再返回（一般不會是null）
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take(); // 注意，這里是能被中斷喚醒的，中斷喚醒會繼續(xù)執(zhí)行循環(huán)。
if (r != null)
return r;
// 代表調(diào)用 workQueue.poll() 超時返回null
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
4.2.5 processWorkerExit方法
工作線程退出時走的方法
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// 因?yàn)閽伋鲇脩舢惓?dǎo)致線程終結(jié)，直接使工作線程數(shù)減1即可
// 如果沒有任何異常拋出的情況下是通過getTask()返回null引導(dǎo)線程正常跳出runWorker()方法的while死循環(huán)從而正常終結(jié)，這種情況下，在getTask()中已經(jīng)把線程數(shù)減1
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 線程退出時，加上他完成的任務(wù)
completedTaskCount += w.completedTasks;
// 將線程從workers集合中移除。
workers.remove(w);
} finally {
mainLock.unlock();
}
// 用于根據(jù)當(dāng)前線程池的狀態(tài)判斷是否需要進(jìn)行線程池terminate處理，見4.2.6 代碼清單4-6
tryTerminate();
int c = ctl.get();
// 如果線程池的狀態(tài)小于STOP，也就是處于RUNNING或者SHUTDOWN狀態(tài)的前提下：
// 1.如果線程不是由于拋出用戶異常終結(jié)，如果允許核心線程超時，則保持線程池中至少存在一個工作線程
// 2.如果線程由于拋出用戶異常終結(jié)，或者當(dāng)前工作線程數(shù)為0，那么直接添加一個新的非核心線程
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
// 理論上線程池中的最小線程數(shù)，允許線程池超時，最小為0，否則為corePoolSize
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
// 如果min == 0，隊(duì)列中還有任務(wù)，那么將min == 1
if (min == 0 && ! workQueue.isEmpty())
min = 1;
// 工作線程數(shù)大于等于最小值，直接返回不新增線程
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}
4.2.6 tryTerminate方法
根據(jù)當(dāng)前線程池的狀態(tài)判斷是否需要進(jìn)行線程池terminate處理，在工作線程退出時，調(diào)用shutdown方法，
shutdownNow方法方法時都會調(diào)用這個方法。
final void tryTerminate() {
for (;;) {
int c = ctl.get();
// 判斷線程池的狀態(tài)，如果是下面三種情況下的任意一種則直接返回：
// 1.線程池處于RUNNING狀態(tài)（RUNNING不需要結(jié)束線程池）
// 2.線程池至少為TIDYING狀態(tài)，也就是TIDYING或者TERMINATED狀態(tài)，意味著已經(jīng)走到了下面的步驟，線程池即將終結(jié)
// 3.線程池狀態(tài)為SHUTDOWN狀態(tài)并且任務(wù)隊(duì)列不為空（SHUTDOWN狀態(tài)還需要處理完任務(wù)隊(duì)列中的任務(wù)）
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
// 工作線程數(shù)不為0，則中斷工作線程集合中的第一個空閑的工作線程
if (workerCountOf(c) != 0) { // Eligible to terminate
// 到了這一步，說明工作線程中還有工作線程，不允許結(jié)束線程池
// 調(diào)用interruptIdleWorkers方法中斷工作線程，interruptIdleWorkers方法代碼在下面，interruptIdleWorkers(ONLY_ONE)只會中斷一個線程，被中斷的線程會繼續(xù)進(jìn)入tryTerminate方法去判斷workerCountOf是否為0，最終總會有一個線程會執(zhí)行到下面的代碼去。
interruptIdleWorkers(ONLY_ONE);
// 說明該線程不是最后一個線程，執(zhí)行上面的interruptIdleWorkers去中斷其它線程（有這么一種情況，非空閑的線程在執(zhí)行某個任務(wù)，執(zhí)行任務(wù)完畢之后，如果它剛好是核心線程，就會在下一輪循環(huán)阻塞在任務(wù)隊(duì)列的take()方法，如果不做額外的干預(yù)，它甚至?xí)诰€程池關(guān)閉之后永久阻塞在任務(wù)隊(duì)列的take()方法中。為了避免這種情況，每個工作線程退出的時候都會嘗試中斷工作線程集合中的某一個空閑的線程，確保所有空閑的線程都能夠正常退出。）
return;
}
//注意-方法走到了這里，說明結(jié)束線程池中的工作線程為0了-
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// CAS設(shè)置線程池狀態(tài)為TIDYING，如果設(shè)置成功則執(zhí)行鉤子方法terminated()
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated();
} finally {
// 最后更新線程池狀態(tài)為TERMINATED
ctl.set(ctlOf(TERMINATED, 0));
// 喚醒阻塞在termination條件的所有線程，這個變量的await()方法在awaitTermination()中調(diào)用
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
// 中斷空閑的工作線程，onlyOne為true的時候，只會中斷工作線程集合中的某一個線程
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
// t.isInterrupted() 為 true表示已經(jīng)被中斷了，這里是去尋找第一個沒有被中斷的空閑線程。w.tryLock() 加鎖成功表示線程是空閑的，否則表示線程在執(zhí)行任務(wù)，不允許中斷。
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
// 這里跳出循環(huán)，也就是只中斷集合中第一個工作線程，集合中的第一個線程被中斷
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
到這里，淺析了execute方法的執(zhí)行流程，從工作線程的創(chuàng)建，工作線程的結(jié)束，阻塞隊(duì)列在線程池中的作用等等，下面將介紹線程池的關(guān)閉方法shutdown和shutdownNow
4.3 線程池的關(guān)閉
關(guān)于
shutdown和shutdownNow方法的使用可參考本篇 2.3 關(guān)閉線程池
4.3.1 shutdown方法
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 權(quán)限校驗(yàn)，安全策略相關(guān)判斷
checkShutdownAccess();
// 設(shè)置SHUTDOWN狀態(tài)
advanceRunState(SHUTDOWN);
// 中斷所有的空閑的工作線程
interruptIdleWorkers();
// 鉤子方法
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
// 調(diào)用上面分析果敢的嘗試terminate方法，使?fàn)顟B(tài)更變?yōu)門IDYING，執(zhí)行鉤子方法terminated()后，最終狀態(tài)更新為TERMINATED
tryTerminate();
}
// 中斷所有的空閑的工作線程
private void interruptIdleWorkers() {
interruptIdleWorkers(false);
}
4.3.2 shutdownNow方法
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 權(quán)限校驗(yàn)，安全策略相關(guān)判斷
checkShutdownAccess();
// 設(shè)置STOP狀態(tài)
advanceRunState(STOP);
// 中斷所有的工作線程
interruptWorkers();
// 清空工作隊(duì)列并且取出所有的未執(zhí)行的任務(wù)
tasks = drainQueue();
} finally {
mainLock.unlock();
}
// 調(diào)用上面分析果敢的嘗試terminate方法，使?fàn)顟B(tài)更變?yōu)門IDYING，執(zhí)行鉤子方法terminated()后，最終狀態(tài)更新為TERMINATED
tryTerminate();
return tasks;
}
// 遍歷所有的工作線程，如果state > 0（啟動狀態(tài)）則進(jìn)行中斷
private void interruptWorkers() {
// assert mainLock.isHeldByCurrentThread();
for (Worker w : workers)
w.interruptIfStarted();
}
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
參考
書籍《Java并發(fā)編程的藝術(shù)》--Java中的線程池
浙公網(wǎng)安備 33010602011771號