Java併發：執行緒池實現原理
一、總覽
執行緒池類ThreadPoolExecutor的相關類需要先了解：
Executor：位於最頂層，只有一個 execute(Runnable runnable) 方法，用於提交任務。
ExecutorService ：在 Executor 介面的基礎上添加了很多的介面方法，提交任務，獲取結果，關閉執行緒池。
AbstractExecutorService：實現了ExecutorService 介面，然後在其基礎上實現了幾個實用的方法，這些方法提供給子類進行呼叫。
ThreadPoolExecutor：執行緒池類
Executors：最常用的用於生成 ThreadPoolExecutor 的例項的工具類
FutureTask：Runnable, Future -> RunnableFuture -> FutureTask
FutureTask 通過 RunnableFuture 間接實現了 Runnable 介面， 所以每個 Runnable 通常都先包裝成 FutureTask， 然後呼叫 executor.execute(Runnable command) 將其提交給執行緒池
Runnable 的 void run() 方法是沒有返回值的，如果我們需要的話，會在 submit 中指定第二個引數作為返回值。
Callable：Callable 也是因為執行緒池的需要，所以才有了這個介面。它和 Runnable 的區別在於 run() 沒有返回值，而 Callable 的 call() 方法有返回值
二、執行緒池狀態
執行緒池中的各個狀態：
RUNNING：這個沒什麼好說的，這是最正常的狀態：接受新的任務，處理等待佇列中的任務
SHUTDOWN：不接受新的任務提交，但是會繼續處理等待佇列中的任務
STOP：不接受新的任務提交，不再處理等待佇列中的任務，中斷正在執行任務的執行緒
TIDYING：所有的任務都銷燬了，workCount 為 0。執行緒池的狀態在轉換為 TIDYING 狀態時，會執行鉤子方法 terminated()
TERMINATED：terminated() 方法結束後，執行緒池的狀態就會變成這個
狀態轉換：
RUNNING -> SHUTDOWN：當呼叫了 shutdown() 後，會發生這個狀態轉換，這也是最重要的
(RUNNING or SHUTDOWN) -> STOP：當呼叫 shutdownNow() 後，會發生這個狀態轉換，這下要清楚 shutDown() 和 shutDownNow() 的區別了
SHUTDOWN -> TIDYING：當任務佇列和執行緒池都清空後，會由 SHUTDOWN 轉換為 TIDYING
STOP -> TIDYING：當任務佇列清空後，發生這個轉換
TIDYING -> TERMINATED：這個前面說了，當 terminated() 方法結束後
ThreadPoolExecutor採用一個 32 位的整數來存放執行緒池的狀態和當前池中的執行緒數，其中高 3 位用於存放執行緒池狀態，低 29 位表示執行緒數。
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 這裡 COUNT_BITS 設定為 29(32-3)，意味著前三位用於存放執行緒狀態，後29位用於存放執行緒數
// 很多初學者很喜歡在自己的程式碼中寫很多 29 這種數字，或者某個特殊的字串，然後分佈在各個地方，這是非常糟糕的
private static final int COUNT_BITS = Integer.SIZE - 3;
// 000 11111111111111111111111111111
// 這裡得到的是 29 個 1，也就是說執行緒池的最大執行緒數是 2^29-1=536870911
// 以我們現在計算機的實際情況，這個數量還是夠用的
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// 我們說了，執行緒池的狀態存放在高 3 位中
// 運算結果為 111跟29個0：111 00000000000000000000000000000
private static final int RUNNING = -1 << COUNT_BITS;
// 000 00000000000000000000000000000
private static final int SHUTDOWN = 0 << COUNT_BITS;
// 001 00000000000000000000000000000
private static final int STOP = 1 << COUNT_BITS;
// 010 00000000000000000000000000000
private static final int TIDYING = 2 << COUNT_BITS;
// 011 00000000000000000000000000000
private static final int TERMINATED = 3 << COUNT_BITS;
// 將整數 c 的低 29 位修改為 0，就得到了執行緒池的狀態
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 將整數 c 的高 3 為修改為 0，就得到了執行緒池中的執行緒數
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
三、執行緒池引數
通過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.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
corePoolSize：執行緒池中核心執行緒的數量。當提交一個任務時，執行緒池會新建一個執行緒來執行任務，直到當前執行緒數等於corePoolSize。如果呼叫了執行緒池的prestartAllCoreThreads()方法，執行緒池會提前建立並啟動所有基本執行緒。
maximumPoolSize：執行緒池中允許的最大執行緒數。執行緒池的阻塞佇列滿了之後，如果還有任務提交，如果當前的執行緒數小於maximumPoolSize，則會新建執行緒來執行任務。注意，如果使用的是×××佇列，該引數也就沒有什麼效果了。
keepAliveTime：空閒執行緒的保活時間，如果某執行緒的空閒時間超過這個值都沒有任務給它做，那麼可以被關閉了。注意這個值並不會對所有執行緒起作用，如果執行緒池中的執行緒數少於等於核心執行緒數 corePoolSize，那麼這些執行緒不會因為空閒太長時間而被關閉，當然，也可以通過呼叫 allowCoreThreadTimeOut(true)使核心執行緒數內的執行緒也可以被回收
unit：keepAliveTime的單位。TimeUnit
workQueue：
用來儲存等待執行的任務的阻塞佇列，等待的任務必須實現Runnable介面。我們可以選擇如下幾種：
ArrayBlockingQueue：基於陣列結構的有界阻塞佇列，FIFO。
LinkedBlockingQueue：基於連結串列結構的有界阻塞佇列，FIFO。
SynchronousQueue：不儲存元素的阻塞佇列，每個插入操作都必須等待一個移出操作，反之亦然。
threadFactory：用於設定建立執行緒的工廠。
handler：
RejectedExecutionHandler，執行緒池的拒絕策略。
所謂拒絕策略，是指將任務新增到執行緒池中時，執行緒池拒絕該任務所採取的相應策略。當向執行緒池中提交任務時，如果此時執行緒池中的執行緒已經飽和了，而且阻塞佇列也已經滿了，則執行緒池會選擇一種拒絕策略來處理該任務。
執行緒池提供了四種拒絕策略：（重寫RejectedExecutionHandler.rejectedExecution(Runnable, ThreadPoolExecutor)）
AbortPolicy：直接丟擲異常，預設策略；
CallerRunsPolicy：用呼叫者所在的執行緒來執行任務；
DiscardOldestPolicy：丟棄阻塞佇列中靠最前的任務，並執行當前任務；
DiscardPolicy：直接丟棄任務；
當然我們也可以實現自己的拒絕策略，例如記錄日誌等等，實現RejectedExecutionHandler介面寫rejectedExecution方法即可。
四、執行緒池建立
Executor工具類提供了三種執行緒池建立方式：
FixedThreadPool ：可重用固定執行緒數的執行緒池
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
corePoolSize 和 maximumPoolSize都設定為建立FixedThreadPool時指定的引數nThreads，意味著當執行緒池滿時且阻塞佇列也已經滿時，如果繼續提交任務，則會直接走拒絕策略，該執行緒池不會再新建執行緒來執行任務，而是直接走拒絕策略。FixedThreadPool使用的是預設的拒絕策略，即AbortPolicy，則直接丟擲異常。
keepAliveTime設定為0L，表示空閒的執行緒會立刻終止。
workQueue則是使用LinkedBlockingQueue，但是沒有設定範圍，那麼則是最大值（Integer.MAX_VALUE），這基本就相當於一個×××隊列了。使用該“×××佇列”則會帶來哪些影響呢？當執行緒池中的執行緒數量等於corePoolSize 時，如果繼續提交任務，該任務會被新增到阻塞佇列workQueue中，當阻塞佇列也滿了之後，則執行緒池會新建執行緒執行任務直到maximumPoolSize。由於FixedThreadPool使用的是“×××佇列”LinkedBlockingQueue，那麼maximumPoolSize引數無效，同時指定的拒絕策略AbortPolicy也將無效。而且該執行緒池也不會拒絕提交的任務，如果客戶端提交任務的速度快於任務的執行，那麼keepAliveTime也是一個無效引數。
SingleThreadExecutor：只有一個執行緒的固定執行緒池
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
為單一worker執行緒的執行緒池，SingleThreadExecutor把corePool和maximumPoolSize均被設定為1，和FixedThreadPool一樣使用的是×××佇列LinkedBlockingQueue,所以帶來的影響和FixedThreadPool一樣。
CachedThreadPool：根據需要建立新執行緒的執行緒池
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
CachedThreadPool的corePool為0，maximumPoolSize為Integer.MAX_VALUE，這就意味著所有的任務一提交就會加入到阻塞佇列中。
keepAliveTime這是為60L，unit設定為TimeUnit.SECONDS，意味著空閒執行緒等待新任務的最長時間為60秒，空閒執行緒超過60秒後將會被終止。
阻塞佇列採用的SynchronousQueue，SynchronousQueue是一個沒有元素的阻塞佇列，加上corePool = 0 ，maximumPoolSize = Integer.MAX_VALUE，這樣就會存在一個問題，如果主執行緒提交任務的速度遠遠大於CachedThreadPool的處理速度，則CachedThreadPool會不斷地建立新執行緒來執行任務，這樣有可能會導致系統耗盡CPU和記憶體資源，所以在使用該執行緒池是，一定要注意控制併發的任務數，否則建立大量的執行緒可能導致嚴重的效能問題。
五、執行過程
提交任務：
執行緒池根據業務不同的需求提供了兩種方式提交任務：Executor.execute()、ExecutorService.submit()。其中ExecutorService.submit()可以獲取該任務執行的Future。
execute()
執行流程如下：
（1）如果執行緒池當前執行緒數小於corePoolSize，則呼叫addWorker建立新執行緒執行任務，成功返回true，失敗執行步驟2。
（2）如果執行緒池處於RUNNING狀態，則嘗試加入阻塞佇列，如果加入阻塞佇列成功，則嘗試進行Double Check，如果加入失敗，則執行步驟3。
如果加入阻塞佇列成功了，則會進行一個Double Check的過程。Double Check過程的主要目的是判斷加入到阻塞隊裡中的執行緒是否可以被執行。如果執行緒池不是RUNNING狀態，則呼叫remove()方法從阻塞佇列中刪除該任務，然後呼叫reject()方法處理任務。否則需要確保還有執行緒執行。
（3）如果執行緒池不是RUNNING狀態或者加入阻塞佇列失敗，則嘗試建立新執行緒直到maxPoolSize，如果失敗，則呼叫reject()方法執行相應的拒絕策略。
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);
}
addWorker
在這裡需要好好理論addWorker中的引數，在execute()方法中，有三處呼叫了該方法：
第一次：workerCountOf(c) < corePoolSize ==> addWorker(command, true)，這個很好理解，當然執行緒池的執行緒數量小於 corePoolSize ，則新建執行緒執行任務即可，在執行過程core == true，內部與corePoolSize比較即可。
第二次：加入阻塞佇列進行Double Check時，else if (workerCountOf(recheck) == 0) ==>addWorker(null, false)。如果執行緒池中的執行緒==0，按照道理應該該任務應該新建執行緒執行任務，但是由於已經該任務已經新增到了阻塞佇列，那麼就線上程池中新建一個空執行緒，然後從阻塞佇列中取執行緒即可。
第三次：執行緒池不是RUNNING狀態或者加入阻塞佇列失敗：else if (!addWorker(command, false))，這裡core == fase，則意味著是與maximumPoolSize比較。
執行流程：
（1）判斷當前執行緒是否可以新增任務，如果可以則進行下一步，否則return false；
rs >= SHUTDOWN ，表示當前執行緒處於SHUTDOWN ，STOP、TIDYING、TERMINATED狀態
rs == SHUTDOWN , firstTask != null時不允許新增執行緒，因為執行緒處於SHUTDOWN 狀態，不允許新增任務
rs == SHUTDOWN , firstTask == null，但workQueue.isEmpty() == true，不允許新增執行緒，因為firstTask == null是為了新增一個沒有任務的執行緒然後再從workQueue中獲取任務的，如果workQueue == null，則說明新增的任務沒有任何意義。
（2）內嵌迴圈，通過CAS worker + 1
（3）獲取主鎖mailLock，如果執行緒池處於RUNNING狀態獲取處於SHUTDOWN狀態且 firstTask == null，則將任務新增到workers Queue中，然後釋放主鎖mainLock，然後啟動執行緒，然後return true，如果中途失敗導致workerStarted= false，則呼叫addWorkerFailed()方法進行處理。
// 第一個引數是準備提交給這個執行緒執行的任務，之前說了，可以為 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();
}