詳解Java並發包中線程池ThreadPoolExecutor
一、線程池簡介
線程池的使用主要是解決兩個問題：①當執行大量異步任務的時候線程池能夠提供更好的性能，在不使用線程池時候，每當需要執行異步任務的時候直接new一個線程來運行的話，線程的創建和銷毀都是需要開銷的。而線程池中的線程是可復用的，不需要每次執行異步任務的時候重新創建和銷毀線程；②線程池提供一種資源限制和管理的手段，比如可以限制線程的個數，動態的新增線程等等。
在下面的分析中，我們可以看到，線程池使用一個Integer的原子類型變量來記錄線程池狀態和線程池中的線程數量，通過線程池狀態來控制任務的執行，每個工作線程Worker線程可以處理多個任務。
二、ThreadPoolExecutor類
2.1、ThreadPoolExecutor成員變量以含義
ThreadPoolExecutor繼承瞭AbstractExecutorService，其中的成員變量ctl是一個Integer類型的原子變量，用來記錄線程池的狀態和線程池中的線程的個數。這裡（Integer看做32位）ctl高三位表示線程池的狀態，後面的29位表示線程池中的線程個數。如下所示是ThreadPoolExecutor源碼中的成員變量
//（高3位）表示線程池狀態，（低29位）表示線程池中線程的個數；
// 默認狀態是RUNNING，線程池中線程個數為0
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//表示具體平臺下Integer的二進制位數-3後的剩餘位數表示的數才是線程的個數;
//其中Integer.SIZE=32,-3之後的低29位表示的就是線程的個數瞭
private static final int COUNT_BITS = Integer.SIZE - 3;
//線程最大個數（低29位）00011111111111111111111111111111（1<<29-1）
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
//線程池狀態（高3位表示線程池狀態）
//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;
//獲取高3位（運行狀態）==> c & 11100000000000000000000000000000
private static int runStateOf(int c) { return c & ~CAPACITY; }
//獲取低29位（線程個數）==> c & 00011111111111111111111111111111
private static int workerCountOf(int c) { return c & CAPACITY; }
//計算原子變量ctl新值（運行狀態和線程個數）
private static int ctlOf(int rs, int wc) { return rs | wc; }
下面我們簡單解釋一下上面的線程狀態的含義：
①RUNNING：接受新任務並處理阻塞隊列中的任務
②SHUTDOWN：拒絕新任務但是處理阻塞隊列中的任務
③STOP：拒絕新任務並拋棄阻塞隊列中的任務，同時會中斷當前正在執行的任務
④TIDYING：所有任務執行完之後（包含阻塞隊列中的任務）當前線程池中活躍的線程數量為0，將要調用terminated方法
⑥TERMINATED：終止狀態。terminated方法調用之後的狀態
2.2、ThreadPoolExecutor的參數以及實現原理
①corePoolSize：線程池核心現車個數
②workQueue：用於保存等待任務執行的任務的阻塞隊列（比如基於數組的有界阻塞隊列ArrayBlockingQueue、基於鏈表的無界阻塞隊列LinkedBlockingQueue等等）
③maximumPoolSize：線程池最大線程數量
④ThreadFactory：創建線程的工廠
⑤RejectedExecutionHandler：拒絕策略，表示當隊列已滿並且線程數量達到線程池最大線程數量的時候對新提交的任務所采取的策略，主要有四種策略：AbortPolicy（拋出異常）、CallerRunsPolicy（隻用調用者所在線程來運行該任務）、DiscardOldestPolicy（丟掉阻塞隊列中最近的一個任務來處理當前提交的任務）、DiscardPolicy（不做處理，直接丟棄掉）
⑥keepAliveTime：存活時間，如果當前線程池中的數量比核心線程數量多，並且當前線程是閑置狀態，該變量就是這些線程的最大生存時間
⑦TimeUnit：存活時間的時間單位。
根據上面的參數介紹，簡單瞭解一下線程池的實現原理，以提交一個新任務為開始點，分析線程池的主要處理流程
2.3、關於一些線程池的使用類型
①newFixedThreadPool：創建一個核心線程個數和最大線程個數均為nThreads的線程池，並且阻塞隊列長度為Integer.MAX_VALUE,keepAliveTime=0說明隻要線程個數比核心線程個數多並且當前空閑即回收。
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
②newSingleThreadExecutor：創建一個核心線程個數和最大線程個數都為1 的線程池，並且阻塞隊列長度為Integer.MAX_VALUE，keepAliveTime=0說明隻要線程個數比核心線程個數多並且當前線程空閑即回收該線程。
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
③newCachedThreadPoolExecutor：創建一個按需創建線程的線程池，初始線程個數為0，最多線程個數為Integer.MAX_VALUE，並且阻塞隊列為同步隊列（最多隻有一個元素），keepAliveTime=60說明隻要當前線程在60s內空閑則回收。這個類型的線程池的特點就是：加入同步隊列的任務會被馬上執行，同步隊列中最多隻有一個任務
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
2.4、ThreadPoolExecutor中的其他成員
//獨占鎖，用來控制新增工作線程Worker操作的原子性 private final ReentrantLock mainLock = new ReentrantLock(); //工作線程集合，Worker繼承瞭AQS接口和Runnable接口，是具體處理任務的線程對象 //Worker實現AQS，並自己實現瞭簡單不可重入獨占鎖，其中state=0表示當前鎖未被獲取狀態，state=1表示鎖被獲取， //state=-1表示Work創建時候的默認狀態，創建時候設置state=-1是為瞭防止runWorker方法運行前被中斷 private final HashSet<Worker> workers = new HashSet<Worker>(); //termination是該鎖對應的條件隊列，在線程調用awaitTermination時候用來存放阻塞的線程 private final Condition termination = mainLock.newCondition();
三、execute(Runnable command)方法實現
executor方法的作用是提交任務command到線程池執行，可以簡單的按照下面的圖進行理解，ThreadPoolExecutor的實現類似於一個生產者消費者模型，當用戶添加任務到線程池中相當於生產者生產元素，workers工作線程則直接執行任務或者從任務隊列中獲取任務，相當於消費之消費元素。
public void execute(Runnable command) {
//（1）首先檢查任務是否為null，為null拋出異常，否則進行下面的步驟
if (command == null)
throw new NullPointerException();
//（2）ctl值中包含瞭當前線程池的狀態和線程池中的線程數量
int c = ctl.get();
//（3）workerCountOf方法是獲取低29位，即獲取當前線程池中的線程個數，如果小於corePoolSize，就開啟新的線程運行
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//（4）如果線程池處理RUNNING狀態，就添加任務到阻塞隊列中
if (isRunning(c) && workQueue.offer(command)) {
//（4-1）二次檢查，獲取ctl值
int recheck = ctl.get();
//（4-2）如果當前線程池不是出於RUNNING狀態，就從隊列中刪除任務，並執行拒絕策略
if (! isRunning(recheck) && remove(command))
reject(command);
//（4-3）否則，如果線程池為空，就添加一個線程
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//（5）如果隊列滿，則新增線程，如果新增線程失敗，就執行拒絕策略
else if (!addWorker(command, false))
reject(command);
}
我們在看一下上面代碼的執行流程，按照標記的數字進行分析：
①步驟（3）判斷當前線程池中的線程個數是否小於corePoolSize，如果小於核心線程數，會向workers裡面新增一個核心線程執行任務。
②如果當前線程池中的線程數量大於核心線程數，就執行（4）。（4）首先判斷當前線程池是否處於RUNNING狀態，如果處於該狀態，就添加任務到任務隊列中，這裡需要判斷線程池的狀態是因為線程池可能已經處於非RUNNING狀態，而在非RUNNING狀態下是需要拋棄新任務的。
③如果想任務隊列中添加任務成功，需要進行二次校驗，因為在添加任務到任務隊列後，可能線程池的狀態發生瞭變化，所以這裡需要進行二次校驗，如果當前線程池已經不是RUNNING狀態瞭，需要將任務從任務隊列中移除，然後執行拒絕策略；如果二次校驗通過，則執行4-3代碼重新判斷當前線程池是否為空，如果線程池為空沒有線程，那麼就需要新創建一個線程。
④如果上面的步驟（4）創建添加任務失敗，說明隊列已滿，那麼（5）會嘗試再開啟新的線程執行任務（類比上圖中的thread3和thread4，即不是核心線程的那些線程），如果當前線程池中的線程個數已經大於最大線程數maximumPoolSize，表示不能開啟新的線程。這就屬於線程池滿並且任務隊列滿，就需要執行拒絕策略瞭。
下面我們在看看addWorker方法的實現
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//(6)檢查隊列是否隻在必要時候為空
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
//(7)使用CAS增加線程個數
for (;;) {
//根據ctl值獲得當前線程池中的線程數量
int wc = workerCountOf(c);
//(7-1)如果線程數量超出限制，返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//(7-2)CAS增加線程數量，同時隻有一個線程可以成功
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // 重新讀取ctl值
//(7-3)CAS失敗瞭，需要查看當前線程池狀態是否發生變化，如果發生變化需要跳轉到外層循環嘗試重新獲取線程池狀態，否則內層循環重新進行CAS增加線程數量
if (runStateOf(c) != rs)
continue retry;
}
}
//(8)執行到這裡說明CAS增加新線程個數成功瞭，我們需要開始創建新的工作線程Worker
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//(8-1)創建新的worker
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
//(8-2)加獨占鎖，保證workers的同步，可能線程池中的多個線程調用瞭線程池的execute方法
mainLock.lock();
try {
// (8-3)重新檢查線程池狀態，以免在獲取鎖之前調用shutdown方法改變線程池狀態
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//(8-4)添加新任務
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
//(8-6)添加新任務成功之後，啟動任務
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
簡單再分析說明一下上面的代碼，addWorker方法主要分為兩部分，第一部分是使用CAS線程安全的添加線程數量，第二部分則是創建新的線程並且將任務並發安全的添加到新的workers之中，然後啟動線程執行。
①代碼（6）中檢查隊列是否隻在必要時候為空，隻有線程池狀態符合條件才能夠進行下面的步驟，從（6）中的判斷條件來看，下面的集中情況addWorker會直接返回false
( I )當前線程池狀態為STOP，TIDYING或者TERMINATED ； (I I）當前線程池狀態為SHUTDOWN並且已經有瞭第一個任務; (I I I)當前線程池狀態為SHUTDOWN並且任務隊列為空
②外層循環中判斷條件通過之後，在內層循環中使用CAS增加線程數，當CAS成功就退出雙重循環進行（8）步驟代碼的執行，如果失敗需要查看當前線程池的狀態是否發生變化，如果發生變化需要進行外層循環重新判斷線程池狀態然後在進入內層循環重新進行CAS增加線程數，如果線程池狀態沒有發生變化但是上一次CAS失敗就繼續進行CAS嘗試。
③執行到（8）代碼處，表明當前已經成功增加 瞭線程數，但是還沒有線程執行任務。ThreadPoolExecutor中使用全局獨占鎖mainLock來控制將新增的工作線程Worker線程安全的添加到工作者線程集合workers中。
④（8-2）獲取瞭獨占鎖，但是在獲取到鎖之後，還需要進行重新檢查線程池的狀態，這是為瞭避免在獲取全局獨占鎖之前其他線程調用瞭shutDown方法關閉瞭線程池。如果線程池已經關閉需要釋放鎖。否則將新增的線程添加到工作集合中，釋放鎖啟動線程執行任務。
上面的addWorker方法最後幾行中，會判斷添加工作線程是否成功，如果失敗，會執行addWorkerFailed方法，將任務從workers中移除，並且workerCount做-1操作。
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
//獲取鎖
mainLock.lock();
try {
//如果worker不為null
if (w != null)
//workers移除worker
workers.remove(w);
//通過CAS操作，workerCount-1
decrementWorkerCount();
tryTerminate();
} finally {
//釋放鎖
mainLock.unlock();
}
}
四、工作線程Worker的執行
4.1、工作線程Worker類源碼分析
上面查看addWorker方法在CAS更新線程數成功之後，下面就是創建新的Worker線程執行任務，所以我們這裡先查看Worker類，下面是Worker類的源碼，我們可以看出，Worker類繼承瞭AQS並實現瞭Runnable接口，所以他既是一個自定義的同步組件，也是一個執行任務的線程類。下面我們分析Worker類的執行
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/** 使用線程工廠創建的線程，執行任務 */
final Thread thread;
/** 初始化執行任務 */
Runnable firstTask;
/** 計數 */
volatile long completedTasks;
/**
* 給出初始firstTask，線程創建工廠創建新的線程
*/
Worker(Runnable firstTask) {
setState(-1); // 防止在調用runWorker之前被中斷
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this); //使用threadFactory創建線程
}
/** run方法實際上執行的是runWorker方法 */
public void run() {
runWorker(this);
}
// 關於同步狀態（鎖）
//
// 同步狀態state=0表示鎖未被獲取
// 同步狀態state=1表示鎖被獲取
protected boolean isHeldExclusively() {
return getState() != 0;
}
//下面都是重寫AQS的方法，Worker為自定義的同步組件
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
在構造函數中我們可以看出，首先將同步狀態state置為-1，而Worker這個同步組件的state有三個值，其中state=-1表示Work創建時候的默認狀態，創建時候設置state=-1是為瞭防止runWorker方法運行前被中斷前面說到過這個結論，這裡置為-1是為瞭避免當前Worker在調用runWorker方法之前被中斷（當其他線程調用線程池的shutDownNow時候，如果Worker的state>=0則會中斷線程），設置為-1就不會被中斷瞭。而Worker實現Runnable接口，那麼需要重寫run方法，在run方法中，我們可以看到，實際上執行的是runWorker方法，在runWorker方法中，會首先調用unlock方法，該方法會將state置為0，所以這個時候調用shutDownNow方法就會中斷當前線程，而這個時候已經進入瞭runWork方法瞭，就不會在還沒有執行runWorker方法的時候就中斷線程。
4.2、runWorker方法的源碼分析
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // 這個時候調用unlock方法，將state置為0，就可以被中斷瞭
boolean completedAbruptly = true;
try {
//(10)如果當前任務為null，或者從任務隊列中獲取到的任務為null，就跳轉到(11)處執行清理工作
while (task != null || (task = getTask()) != null) {
//task不為null，就需要線程執行任務，這個時候，需要獲取工作線程內部持有的獨占鎖
w.lock();
/**如果線程池已被停止(STOP)(至少大於STOP狀態)，要確保線程都被中斷
* 如果狀態不對，檢查當前線程是否中斷並清除中斷狀態，並且再次檢查線程池狀態是否大於STOP
* 如果上述滿足，檢查該對象是否處於中斷狀態，不清除中斷標記
*/
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) {
thrown = x; throw new Error(x);
} finally {
//執行任務之後的方法
afterExecute(task, thrown);
}
} finally {
task = null;
//更新當前已完成任務數量
w.completedTasks++;
//釋放鎖
w.unlock();
}
}
completedAbruptly = false;
} finally {
//執行清理工作：處理並退出當前worker
processWorkerExit(w, completedAbruptly);
}
}
我們梳理一下runWorker方法的執行流程
①首先先執行unlock方法，將Worker的state置為0，這樣工作線程就可以被中斷瞭（後續的操作如果線程池關閉就需要線程被中斷）
②首先判斷判斷當前的任務（當前工作線程中的task，或者從任務隊列中取出的task）是否為null，如果不為null就往下執行，為null就執行processWorkerExit方法。
③獲取工作線程內部持有的獨占鎖（避免在執行任務期間，其他線程調用shutdown後正在執行的任務被中斷，shutdown隻會中斷當前被阻塞掛起的沒有執行任務的線程）
④然後執行beforeExecute()方法，該方法為擴展接口代碼，表示在具體執行任務之前所做的一些事情，然後執行task.run()方法執行具體任務，執行完之後會調用afterExecute()方法，用以處理任務執行完畢之後的工作，也是一個擴展接口代碼。
⑤更新當前線程池完成的任務數，並釋放鎖
4.3、執行清理工作的方法processWorkerExit
下面是方法processWorkerExit的源碼，在下面的代碼中
①首先(1-1)處統計線程池完成的任務個數，並且在此之前獲取全局鎖，然後更新當前的全局計數器，然後從工作線程集合中移除當前工作線程，完成清理工作。
②代碼(1-2)調用瞭tryTerminate方法，在該方法中，判斷瞭當前線程池狀態是SHUTDOWN並且隊列不為空或者當前線程池狀態為STOP並且當前線程池中沒有活動線程，則置線程池狀態為TERMINATED。如果設置稱為瞭TERMINATED狀態，還需要調用全局條件變量termination的signalAll方法喚醒所有因為調用線程池的awaitTermination方法而被阻塞住的線程，使得線程池中的所有線程都停止，從而使得線程池為TERMINATED狀態。
③代碼(1-3)處判斷當前線程池中的線程個數是否小於核心線程數，如果是，需要新增一個線程保證有足夠的線程可以執行任務隊列中的任務或者提交的任務。
private void processWorkerExit(Worker w, boolean completedAbruptly) {
/*
*completedAbruptly:是由runWorker傳過來的參數，表示是否突然完成的意思
*當在就是在執行任務過程當中出現異常，就會突然完成，傳true
*
*如果是突然完成，需要通過CAS操作，更新workerCount(-1操作)
*不是突然完成，則不需要-1，因為getTask方法當中已經-1(getTask方法中執行瞭decrementWorkerCount()方法)
*/
if (completedAbruptly)
decrementWorkerCount();
//(1-1)在統計完成任務個數之前加上全局鎖，然後統計線程池中完成的任務個數並更新全局計數器，並從工作集中刪除當前worker
final ReentrantLock mainLock = this.mainLock;
mainLock.lock(); //獲得全局鎖
try {
completedTaskCount += w.completedTasks; //更新已完成的任務數量
workers.remove(w); //將完成該任務的線程worker從工作線程集合中移除
} finally {
mainLock.unlock(); //釋放鎖
}
/**(1-2)
* 這一個方法調用完成瞭下面的事情：
* 判斷如果當前線程池狀態是SHUTDOWN並且工作隊列為空，
* 或者當前線程池狀態是STOP並且當前線程池裡面沒有活動線程，
* 則設置當前線程池狀態為TERMINATED，如果設置成瞭TERMINATED狀態，
* 還需要調用條件變量termination的signAll方法激活所有因為調用線程池的awaitTermination方法而被阻塞的線程
*/
tryTerminate();
//（1-3）如果當前線程池中線程數小於核心線程，則增加核心線程數
int c = ctl.get();
//判斷當前線程池的狀態是否小於STOP（RUNNING或者SHUTDOWN）
if (runStateLessThan(c, STOP)) {
//如果任務忽然完成，執行後續的代碼
if (!completedAbruptly) {
//allowCoreThreadTimeOut表示是否允許核心線程超時，默認為false
//min這裡當默認為allowCoreThreadTimeOut默認為false的時候，min置為coorPoolSize
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
//這裡說明：如果允許核心線程超時，那麼allowCoreThreadTimeOut可為true，那麼min值為0，不需要維護核心線程瞭
//如果min為0並且任務隊列不為空
if (min == 0 && ! workQueue.isEmpty())
min = 1; //這裡表示如果min為0，且隊列不為空，那麼至少需要一個核心線程存活來保證任務的執行
//如果工作線程數大於min,表示當前線程數滿足，直接返回
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}
在tryTerminate方法中，我們簡單說明瞭該方法的作用，下面是該方法的源碼，可以看出源碼實現上和上面所總結的功能是差不多的
final void tryTerminate() {
for (;;) {
//獲取線程池狀態
int c = ctl.get();
//如果線程池狀態為RUNNING
//或者狀態大於TIDYING
//或者狀態==SHUTDOWN並未任務隊列不為空
//直接返回，不能調用terminated方法
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
//如果線程池中工作線程數不為0，需要中斷線程
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
//獲得線程池的全局鎖
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//通過CAS操作，將線程池狀態設置為TIDYING
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { //private static int ctlOf(int rs, int wc) { return rs | wc; }
try {
//調用terminated方法
terminated();
} finally {
//最終將線程狀態設置為TERMINATED
ctl.set(ctlOf(TERMINATED, 0));
//調用條件變量termination的signaAll方法喚醒所有因為
//調用線程池的awaitTermination方法而被阻塞的線程
//private final Condition termination = mainLock.newCondition();
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
五、補充（shutdown、shutdownNow、awaitTermination方法）
5.1、shutdown操作
我們在使用線程池的時候知道，調用shutdown方法之後線程池就不會再接受新的任務瞭，但是任務隊列中的任務還是需要執行完的。調用該方法會立刻返回，並不是等到線程池的任務隊列中的所有任務執行完畢在返回的。
public void shutdown() {
//獲得線程池的全局鎖
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//進行權限檢查
checkShutdownAccess();
//設置當前線程池的狀態的SHUTDOWN，如果線程池狀態已經是該狀態就會直接返回，下面我們會分析這個方法的源碼
advanceRunState(SHUTDOWN);
//設置中斷 標志
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
//嘗試將狀態變為TERMINATED，上面已經分析過該方法的源碼
tryTerminate();
}
該方法的源碼比較簡短，首先檢查瞭安全管理器，是查看當前調用shutdown命令的線程是否有關閉線程的權限，如果有權限還需要看調用線程是否有中斷工作線程的權限，如果沒有權限將會拋出SecurityException異常或者空指針異常。下面我們查看一下advanceRunState 方法的源碼。
private void advanceRunState(int targetState) {
for (;;) {
//下面的方法執行的就是：
//首先獲取線程的ctl值，然後判斷當前線程池的狀態如果已經是SHUTDOWN，那麼if條件第一個為真就直接返回
//如果不是SHUTDOWN狀態，就需要CAS的設置當前狀態為SHUTDOWN
int c = ctl.get();
if (runStateAtLeast(c, targetState) ||
//private static int ctlOf(int rs, int wc) { return rs | wc; }
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
break;
}
}
我們可以看出advanceRunState 方法實際上就是判斷當前線程池的狀態是否為SHUTDWON，如果是那麼就返回，否則就需要設置當前狀態為SHUTDOWN。
我們再來看看shutdown方法中調用線程中斷的方法interruptIdleWorkers源碼
private void interruptIdleWorkers() {
interruptIdleWorkers(false);
}
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
//如果工作線程沒有被中斷，並且沒有正在運行設置中斷標志
if (!t.isInterrupted() && w.tryLock()) {
try {
//需要中斷當前線程
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
上面的代碼中，需要設置所有空閑線程的中斷標志。首先獲取線程池的全局鎖，同時隻有一個線程可以調用shutdown方法設置中斷標志。然後嘗試獲取工作線程Worker自己的鎖，獲取成功則可以設置中斷標志（這是由於正在執行任務的線程需要獲取自己的鎖，並且不可重入，所以正在執行的任務沒有被中斷），這裡要中斷的那些線程是阻塞到getTask()方法並嘗試從任務隊列中獲取任務的線程即空閑線程。
5.2、shutdownNow操作
在使用線程池的時候，如果我們調用瞭shutdownNow方法，線程池不僅不會再接受新的任務，還會將任務隊列中的任務丟棄，正在執行的任務也會被中斷，然後立刻返回該方法，不會等待激活的任務完成，返回值為當前任務隊列中被丟棄的任務列表
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess(); //還是進行權限檢查
advanceRunState(STOP); //設置線程池狀態臺STOP
interruptWorkers(); //中斷所有線程
tasks = drainQueue(); //將任務隊列中的任務移動到task中
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks; //返回tasks
}
從上面的代碼中，我們可以可以發現，shutdownNow方法也是首先需要檢查調用該方法的線程的權限，之後不同於shutdown方法之處在於需要即刻設置當前線程池狀態為STOP，然後中斷所有線程（空閑線程+正在執行任務的線程），移除任務隊列中的任務
private void interruptWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) //不需要判斷當前線程是否在執行任務（即不需要調用w.tryLock方法），中斷所有線程
w.interruptIfStarted();
} finally {
mainLock.unlock();
}
}
5.3、awaitTermination操作
當線程調用該方法之後，會阻塞調用者線程，直到線程池狀態為TERMINATED狀態才會返回，或者等到超時時間到之後會返回，下面是該方法的源碼。
//調用該方法之後，會阻塞調用者線程，直到線程池狀態為TERMINATED狀態才會返回，或者等到超時時間到之後會返回
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//阻塞當前線程，（獲取瞭Worker自己的鎖），那麼當前線程就不會再執行任務（因為獲取不到鎖）
for (;;) {
//當前線程池狀態為TERMINATED狀態，會返回true
if (runStateAtLeast(ctl.get(), TERMINATED))
return true;
//超時時間到返回false
if (nanos <= 0)
return false;
nanos = termination.awaitNanos(nanos);
}
} finally {
mainLock.unlock();
}
}
在上面的代碼中，調用者線程需要首先獲取線程Worker自己的獨占鎖，然後在循環判斷當前線程池是否已經是TERMINATED狀態，如果是則直接返回，否則說明當前線程池中還有線程正在執行任務，這時候需要查看當前設置的超時時間是否小於0，小於0說明不需要等待就直接返回，如果大於0就需要調用條件變量termination的awaitNanos方法等待設置的時間，並在這段時間之內等待線程池的狀態變為TERMINATED。
我們在前面說到清理線程池的方法processWorkerExit的時候，需要調用tryTerminated方法，在該方法中會查看當前線程池狀態是否為TERMINATED，如果是該狀態也會調用termination.signalAll()方法喚醒所有線程池中因調用awaitTermination而被阻塞住的線程。
如果是設置瞭超時時間，那麼termination的awaitNanos方法也會返回，這時候需要重新檢查線程池狀態是否為TERMINATED，如果是則返回，不是就繼續阻塞自己。
以上就是Java並發包中線程池ThreadPoolExecutor原理探究的詳細內容，更多關於Java 並發包 線程池 ThreadPoolExecutor的資料請關註WalkonNet其它相關文章！
推薦閱讀：
- java線程池中Worker線程執行流程原理解析
- Java線程池ThreadPoolExecutor源碼深入分析
- Android開發中線程池源碼解析
- Android線程池源碼閱讀記錄介紹
- 深入理解Java線程池從設計思想到源碼解讀