詳解 JUC 線程池中的 ThreadPoolExecutor
但願美好的東西可以美好地終結，是一種卑微的人之常情。java
前提
很早以前就打算看一次JUC線程池ThreadPoolExecutor的源碼實現，因爲近段時間比較忙，一直沒有時間整理出源碼分析的文章。以前在分析擴展線程池實現可回調的Future時候曾經提到併發大師Doug Lea在設計線程池ThreadPoolExecutor的提交任務的頂層接口Executor只有一個無狀態的執行方法：安全
public interface Executor {
void execute(Runnable command);
}
而ExecutorService提供了不少擴展方法底層基本上是基於Executor#execute()方法進行擴展。本文着重分析ThreadPoolExecutor#execute()的實現，筆者會從實現原理、源碼實現等角度結合簡化例子進行詳細的分析。ThreadPoolExecutor的源碼從JDK8到JDK11基本沒有變化，本文編寫的時候使用的是JDK11。併發
ThreadPoolExecutor的原理
ThreadPoolExecutor裏面使用到JUC同步器框架AbstractQueuedSynchronizer（俗稱AQS）、大量的位操做、CAS操做。ThreadPoolExecutor提供了固定活躍線程（核心線程）、額外的線程（線程池容量 - 核心線程數這部分額外建立的線程，下面稱爲非核心線程）、任務隊列以及拒絕策略這幾個重要的功能。框架
JUC同步器框架
ThreadPoolExecutor裏面使用到JUC同步器框架，主要用於四個方面：異步
- 全局鎖
mainLock成員屬性，是可重入鎖ReentrantLock類型，主要是用於訪問工做線程Worker集合和進行數據統計記錄時候的加鎖操做。 - 條件變量
termination，Condition類型，主要用於線程進行等待終結awaitTermination()方法時的帶期限阻塞。 - 任務隊列
workQueue，BlockingQueue類型，任務隊列，用於存放待執行的任務。 - 工做線程，內部類
Worker類型，是線程池中真正的工做線程對象。
關於AQS筆者以前寫過一篇相關源碼分析的文章：JUC同步器框架AbstractQueuedSynchronizer源碼圖文分析。ide
核心線程
這裏先參考ThreadPoolExecutor的實現而且進行簡化，實現一個只有核心線程的線程池，要求以下：函數
- 暫時不考慮任務執行異常狀況下的處理。
- 任務隊列爲無界隊列。
- 線程池容量固定爲核心線程數量。
- 暫時不考慮拒絕策略。
public class CoreThreadPool implements Executor {
private BlockingQueue<Runnable> workQueue;
private static final AtomicInteger COUNTER = new AtomicInteger();
private int coreSize;
private int threadCount = 0;
public CoreThreadPool(int coreSize) {
this.coreSize = coreSize;
this.workQueue = new LinkedBlockingQueue<>();
}
@Override
public void execute(Runnable command) {
if (++threadCount <= coreSize) {
new Worker(command).start();
} else {
try {
workQueue.put(command);
} catch (InterruptedException e) {
throw new IllegalStateException(e);
}
}
}
private class Worker extends Thread {
private Runnable firstTask;
public Worker(Runnable runnable) {
super(String.format("Worker-%d", COUNTER.getAndIncrement()));
this.firstTask = runnable;
}
@Override
public void run() {
Runnable task = this.firstTask;
while (null != task || null != (task = getTask())) {
try {
task.run();
} finally {
task = null;
}
}
}
}
private Runnable getTask() {
try {
return workQueue.take();
} catch (InterruptedException e) {
throw new IllegalStateException(e);
}
}
public static void main(String[] args) throws Exception {
CoreThreadPool pool = new CoreThreadPool(5);
IntStream.range(0, 10)
.forEach(i -> pool.execute(() ->
System.out.println(String.format("Thread:%s,value:%d", Thread.currentThread().getName(), i))));
Thread.sleep(Integer.MAX_VALUE);
}
}
某次運行結果以下：oop
Thread:Worker-0,value:0 Thread:Worker-3,value:3 Thread:Worker-2,value:2 Thread:Worker-1,value:1 Thread:Worker-4,value:4 Thread:Worker-1,value:5 Thread:Worker-2,value:8 Thread:Worker-4,value:7 Thread:Worker-0,value:6 Thread:Worker-3,value:9
設計此線程池的時候，核心線程是懶建立的，若是線程空閒的時候則阻塞在任務隊列的take()方法，其實對於ThreadPoolExecutor也是相似這樣實現，只是若是使用了keepAliveTime而且容許核心線程超時（allowCoreThreadTimeOut設置爲true）則會使用BlockingQueue#poll(keepAliveTime)進行輪詢代替永久阻塞。源碼分析
其餘附加功能
構建ThreadPoolExecutor實例的時候，須要定義maximumPoolSize（線程池最大線程數）和corePoolSize（核心線程數）。當任務隊列是有界的阻塞隊列，核心線程滿負載，任務隊列已經滿的狀況下，會嘗試建立額外的maximumPoolSize - corePoolSize個線程去執行新提交的任務。當ThreadPoolExecutor這裏實現的兩個主要附加功能是：ui
- 必定條件下會建立非核心線程去執行任務，非核心線程的回收週期（線程生命週期終結時刻）是
keepAliveTime，線程生命週期終結的條件是：下一次經過任務隊列獲取任務的時候而且存活時間超過keepAliveTime。 - 提供拒絕策略，也就是在覈心線程滿負載、任務隊列已滿、非核心線程滿負載的條件下會觸發拒絕策略。
源碼分析
先分析線程池的關鍵屬性，接着分析其狀態控制，最後重點分析ThreadPoolExecutor#execute()方法。
關鍵屬性
public class ThreadPoolExecutor extends AbstractExecutorService {
// 控制變量-存放狀態和線程數
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 任務隊列，必須是阻塞隊列
private final BlockingQueue<Runnable> workQueue;
// 工做線程集合，存放線程池中全部的（活躍的）工做線程，只有在持有全局鎖mainLock的前提下才能訪問此集合
private final HashSet<Worker> workers = new HashSet<>();
// 全局鎖
private final ReentrantLock mainLock = new ReentrantLock();
// awaitTermination方法使用的等待條件變量
private final Condition termination = mainLock.newCondition();
// 記錄峯值線程數
private int largestPoolSize;
// 記錄已經成功執行完畢的任務數
private long completedTaskCount;
// 線程工廠，用於建立新的線程實例
private volatile ThreadFactory threadFactory;
// 拒絕執行處理器，對應不一樣的拒絕策略
private volatile RejectedExecutionHandler handler;
// 空閒線程等待任務的時間週期，單位是納秒
private volatile long keepAliveTime;
// 是否容許核心線程超時，若是爲true則keepAliveTime對核心線程也生效
private volatile boolean allowCoreThreadTimeOut;
// 核心線程數
private volatile int corePoolSize;
// 線程池容量
private volatile int maximumPoolSize;
// 省略其餘代碼
}
下面看參數列表最長的構造函數：
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：int類型，核心線程數量。maximumPoolSize：int類型，最大線程數量，也就是線程池的容量。keepAliveTime：long類型，線程空閒等待時間，也和工做線程的生命週期有關，下文會分析。unit：TimeUnit類型，keepAliveTime參數的時間單位，實際上keepAliveTime最終會轉化爲納秒。workQueue：BlockingQueue類型，等待隊列或者叫任務隊列。threadFactory：ThreadFactory類型，線程工廠，用於建立工做線程（包括核心線程和非核心線程），默認使用Executors.defaultThreadFactory()做爲內建線程工廠實例，通常自定義線程工廠才能更好地跟蹤工做線程。handler：RejectedExecutionHandler
類型，線程池的拒絕執行處理器，更多時候稱爲拒絕策略，拒絕策略執行的時機是當阻塞隊列已滿、沒有空閒的線程（包括核心線程和非核心線程）而且繼續提交任務。提供了4種內建的拒絕策略實現：
AbortPolicy：直接拒絕策略，也就是不會執行任務，直接拋出RejectedExecutionException，這是默認的拒絕策略。DiscardPolicy：拋棄策略，也就是直接忽略提交的任務（通俗來講就是空實現）。DiscardOldestPolicy：拋棄最老任務策略，也就是經過poll()方法取出任務隊列隊頭的任務拋棄，而後執行當前提交的任務。CallerRunsPolicy：調用者執行策略，也就是當前調用Executor#execute()的線程直接調用任務Runnable#run()，通常不但願任務丟失會選用這種策略，但從實際角度來看，原來的異步調用意圖會退化爲同步調用。
狀態控制
狀態控制主要圍繞原子整型成員變量ctl：
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;
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;
// 經過ctl值獲取運行狀態
private static int runStateOf(int c) { return c & ~COUNT_MASK; }
// 經過ctl值獲取工做線程數
private static int workerCountOf(int c) { return c & COUNT_MASK; }
// 經過運行狀態和工做線程數計算ctl的值，或運算
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;
}
// CAS操做線程數增長1
private boolean compareAndIncrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect + 1);
}
// CAS操做線程數減小1
private boolean compareAndDecrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect - 1);
}
// 線程數直接減小1
private void decrementWorkerCount() {
ctl.addAndGet(-1);
}
接下來分析一下線程池的狀態變量，工做線程上限數量位的長度是COUNT_BITS，它的值是Integer.SIZE - 3，也就是正整數29：
咱們知道，整型包裝類型Integer實例的大小是4 byte，一共32 bit，也就是一共有32個位用於存放0或者1。在ThreadPoolExecutor實現中，使用32位的整型包裝類型存放工做線程數和線程池狀態。其中，低29位用於存放工做線程數，而高3位用於存放線程池狀態，因此線程池的狀態最多隻能有2^3種。工做線程上限數量爲2^29 - 1，超過5億，這個數量在短期內不用考慮會超限。
接着看工做線程上限數量掩碼COUNT_MASK，它的值是(1 < COUNT_BITS) - l，也就是1左移29位，再減去1，若是補全32位，它的位視圖以下：
而後就是線程池的狀態常量，這裏只詳細分析其中一個，其餘類同，這裏看RUNNING狀態：
// -1的補碼爲：111-11111111111111111111111111111 // 左移29位後：111-00000000000000000000000000000 // 10進制值爲：-536870912 // 高3位111的值就是表示線程池正在處於運行狀態 private static final int RUNNING = -1 << COUNT_BITS;
控制變量ctl的組成就是經過線程池運行狀態rs和工做線程數wc經過或運算獲得的：
// rs=RUNNING值爲：111-00000000000000000000000000000
// wc的值爲0：000-00000000000000000000000000000
// rs | wc的結果爲：111-00000000000000000000000000000
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static int ctlOf(int rs, int wc) {
return rs | wc;
}
那麼咱們怎麼從ctl中取出高3位的線程池狀態？上面源碼中提供的runStateOf()方法就是提取運行狀態：
// 先把COUNT_MASK取反(~COUNT_MASK)，
獲得：111-00000000000000000000000000000
// ctl位圖特色是：xxx-yyyyyyyyyyyyyyyyyyyyyyyyyyyyyy
// 二者作一次與運算便可獲得高3位xxx
private static int runStateOf(int c){
return c & ~COUNT_MASK;
}
同理，取出低29位的工做線程數量只須要把ctl和COUNT_MASK(000-11111111111111111111111111111)作一次與運算便可。
工做線程數爲0的前提下，小結一下線程池的運行狀態常量：
這裏有一個比較特殊的技巧，因爲運行狀態值存放在高3位，因此能夠直接經過十進制值（甚至能夠忽略低29位，直接用ctl進行比較，或者使用ctl和線程池狀態常量進行比較）來比較和判斷線程池的狀態：
工做線程數爲0的前提下：RUNNING(-536870912) < SHUTDOWN(0) < STOP(536870912) < TIDYING(1073741824) < TERMINATED(1610612736)
下面這三個方法就是使用這種技巧：
// ctl和狀態常量比較，判斷是否小於
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
// ctl和狀態常量比較，判斷是否小於或等於
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
// ctl和狀態常量SHUTDOWN比較，判斷是否處於RUNNING狀態
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
最後是線程池狀態的躍遷圖：
PS：線程池源碼中有不少中間變量用了簡單的單字母表示，例如c就是表示ctl、wc就是表示worker count、rs就是表示running status。
execute方法源碼分析
線程池異步執行任務的方法實現是ThreadPoolExecutor#execute()，源碼以下：
// 執行命令，其中命令（下面稱任務）對象是Runnable的實例
public void execute(Runnable command) {
// 判斷命令（任務）對象非空
if (command == null)
throw new NullPointerException();
// 獲取ctl的值
int c = ctl.get();
// 判斷若是當前工做線程數小於核心線程數，則建立新的核心線程而且執行傳入的任務
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
// 若是建立新的核心線程成功則直接返回
return;
// 這裏說明建立核心線程失敗，須要更新ctl的臨時變量c
c = ctl.get();
}
// 走到這裏說明建立新的核心線程失敗，也就是當前工做線程數大於等於corePoolSize
// 判斷線程池是否處於運行中狀態，同時嘗試用非阻塞方法向任務隊列放入任務（放入任務失敗返回false）
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 這裏是向任務隊列投聽任務成功，對線程池的運行中狀態作二次檢查
// 若是線程池二次檢查狀態是非運行中狀態，則從任務隊列移除當前的任務調用拒絕策略處理之（也就是移除前面成功入隊的任務實例）
if (! isRunning(recheck) && remove(command))
// 調用拒絕策略處理任務 - 返回
reject(command);
// 走到下面的else if分支，說明有如下的前提：
// 0、待執行的任務已經成功加入任務隊列
// 一、線程池多是RUNNING狀態
// 二、傳入的任務可能從任務隊列中移除失敗（移除失敗的惟一可能就是任務已經被執行了）
// 若是當前工做線程數量爲0，則建立一個非核心線程而且傳入的任務對象爲null - 返回
// 也就是建立的非核心線程不會立刻運行，而是等待獲取任務隊列的任務去執行
// 若是前工做線程數量不爲0，原來應該是最後的else分支，可是能夠什麼也不作，由於任務已經成功入隊列，總會有合適的時機分配其餘空閒線程去執行它
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 走到這裏說明有如下的前提：
// 0、線程池中的工做線程總數已經大於等於corePoolSize（簡單來講就是核心線程已經所有懶建立完畢）
// 一、線程池可能不是RUNNING狀態
// 二、線程池多是RUNNING狀態同時任務隊列已經滿了
// 若是向任務隊列投聽任務失敗，則會嘗試建立非核心線程傳入任務執行
// 建立非核心線程失敗，此時須要拒絕執行任務
else if (!addWorker(command, false))
// 調用拒絕策略處理任務 - 返回
reject(command);
}
這裏簡單分析一下整個流程：
- 若是當前工做線程總數小於
corePoolSize，則直接建立核心線程執行任務（任務實例會傳入直接用於構造工做線程實例）。 - 若是當前工做線程總數大於等於
corePoolSize，判斷線程池是否處於運行中狀態，同時嘗試用非阻塞方法向任務隊列放入任務，這裏會二次檢查線程池運行狀態，若是當前工做線程數量爲0，則建立一個非核心線程而且傳入的任務對象爲null。 - 若是向任務隊列投聽任務失敗（任務隊列已經滿了），則會嘗試建立非核心線程傳入任務實例執行。
- 若是建立非核心線程失敗，此時須要拒絕執行任務，調用拒絕策略處理任務。
這裏是一個疑惑點：爲何須要二次檢查線程池的運行狀態，當前工做線程數量爲0，嘗試建立一個非核心線程而且傳入的任務對象爲null？這個能夠看API註釋：
若是一個任務成功加入任務隊列，咱們依然須要二次檢查是否須要添加一個工做線程（由於全部存活的工做線程有可能在最後一次檢查以後已經終結）或者執行當前方法的時候線程池是否已經shutdown了。因此咱們須要二次檢查線程池的狀態，必須時把任務從任務隊列中移除或者在沒有可用的工做線程的前提下新建一個工做線程。
任務提交流程從調用者的角度來看以下：
addWorker方法源碼分析
boolean addWorker(Runnable firstTask, boolean core)方法的第一的參數能夠用於直接傳入任務實例，第二個參數用於標識將要建立的工做線程是否核心線程。方法源碼以下：
// 添加工做線程，若是返回false說明沒有新建立工做線程，若是返回true說明建立和啓動工做線程成功
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
// 注意這是一個死循環 - 最外層循環
for (int c = ctl.get();;) {
// 這個是十分複雜的條件，這裏先拆分多個與（&&）條件：
// 1. 線程池狀態至少爲SHUTDOWN狀態，也就是rs >= SHUTDOWN(0)
// 2. 線程池狀態至少爲STOP狀態，也就是rs >= STOP(1)，或者傳入的任務實例firstTask不爲null，或者任務隊列爲空
// 其實這個判斷的邊界是線程池狀態爲shutdown狀態下，不會再接受新的任務，在此前提下若是狀態已經到了STOP、或者傳入任務不爲空、或者任務隊列爲空（已經沒有積壓任務）都不須要添加新的線程
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP)
|| firstTask != null
|| workQueue.isEmpty()))
return false;
// 注意這也是一個死循環 - 二層循環
for (;;) {
// 這裏每一輪循環都會從新獲取工做線程數wc
// 1. 若是傳入的core爲true，表示將要建立核心線程，經過wc和corePoolSize判斷，若是wc >= corePoolSize，則返回false表示建立核心線程失敗
// 1. 若是傳入的core爲false，表示將要創非建核心線程，經過wc和maximumPoolSize判斷，若是wc >= maximumPoolSize，則返回false表示建立非核心線程失敗
if (workerCountOf(c)
>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
return false;
// 成功經過CAS更新工做線程數wc，則break到最外層的循環
if (compareAndIncrementWorkerCount(c))
break retry;
// 走到這裏說明了經過CAS更新工做線程數wc失敗，這個時候須要從新判斷線程池的狀態是否由RUNNING已經變爲SHUTDOWN
c = ctl.get(); // Re-read ctl
// 若是線程池狀態已經由RUNNING已經變爲SHUTDOWN，則從新跳出到外層循環繼續執行
if (runStateAtLeast(c, SHUTDOWN))
continue retry;
// 若是線程池狀態依然是RUNNING，CAS更新工做線程數wc失敗說明有多是併發更新致使的失敗，則在內層循環重試便可
// else CAS failed due to workerCount change; retry inner loop
}
}
// 標記工做線程是否啓動成功
boolean workerStarted = false;
// 標記工做線程是否建立成功
boolean workerAdded = false;
Worker w = null;
try {
// 傳入任務實例firstTask建立Worker實例，Worker構造裏面會經過線程工廠建立新的Thread對象，因此下面能夠直接操做Thread t = w.thread
// 這一步Worker實例已經建立，可是沒有加入工做線程集合或者啓動它持有的線程Thread實例
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
// 這裏須要全局加鎖，由於會改變一些指標值和非線程安全的集合
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 c = ctl.get();
// 這裏主要在加鎖的前提下判斷ThreadFactory建立的線程是否存活或者判斷獲取鎖成功以後線程池狀態是否已經更變爲SHUTDOWN
// 1. 若是線程池狀態依然爲RUNNING，則只須要判斷線程實例是否存活，須要添加到工做線程集合和啓動新的Worker
// 2. 若是線程池狀態小於STOP，也就是RUNNING或者SHUTDOWN狀態下，同時傳入的任務實例firstTask爲null，則須要添加到工做線程集合和啓動新的Worker
// 對於2，換言之，若是線程池處於SHUTDOWN狀態下，同時傳入的任務實例firstTask不爲null，則不會添加到工做線程集合和啓動新的Worker
// 這一步其實有可能建立了新的Worker實例可是並不啓動（臨時對象，沒有任何強引用），這種Worker有可能成功下一輪GC被收集的垃圾對象
if (isRunning(c) ||
(runStateLessThan(c, STOP) && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
// 把建立的工做線程實例添加到工做線程集合
workers.add(w);
int s = workers.size();
// 嘗試更新歷史峯值工做線程數，也就是線程池峯值容量
if (s > largestPoolSize)
largestPoolSize = s;
// 這裏更新工做線程是否啓動成功標識爲true，後面纔會調用Thread#start()方法啓動真實的線程實例
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 若是成功添加工做線程，則調用Worker內部的線程實例t的Thread#start()方法啓動真實的線程實例
if (workerAdded) {
t.start();
// 標記線程啓動成功
workerStarted = true;
}
}
} finally {
// 線程啓動失敗，須要從工做線程集合移除對應的Worker
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
// 添加Worker失敗
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 從工做線程集合移除之
if (w != null)
workers.remove(w);
// wc數量減1
decrementWorkerCount();
// 基於狀態判斷嘗試終結線程池
tryTerminate();
} finally {
mainLock.unlock();
}
}
筆者發現了Doug Lea大神十分喜歡複雜的條件判斷，並且單行復雜判斷不喜歡加花括號，像下面這種代碼在他編寫的不少類庫中都比較常見：
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP)
|| firstTask != null
|| workQueue.isEmpty()))
return false;
// ....
// 代碼拆分一下以下
boolean atLeastShutdown = runStateAtLeast(c, SHUTDOWN); # rs >= SHUTDOWN(0)
boolean atLeastStop = runStateAtLeast(c, STOP) || firstTask != null || workQueue.isEmpty();
if (atLeastShutdown && atLeastStop){
return false;
}
上面的分析邏輯中須要注意一點，Worker實例建立的同時，在其構造函數中會經過ThreadFactory建立一個Java線程Thread實例，後面會加鎖後二次檢查是否須要把Worker實例添加到工做線程集合workers中和是否須要啓動Worker中持有的Thread實例，只有啓動了Thread實例實例，Worker才真正開始運做，不然只是一個無用的臨時對象。Worker自己也實現了Runnable接口，它能夠當作是一個Runnable的適配器。
工做線程內部類Worker源碼分析
線程池中的每個具體的工做線程被包裝爲內部類Worker實例，Worker繼承於AbstractQueuedSynchronizer(AQS)，實現了Runnable接口：
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;
// 保存ThreadFactory建立的線程實例，若是ThreadFactory建立線程失敗則爲null
final Thread thread;
// 保存傳入的Runnable任務實例
Runnable firstTask;
// 記錄每一個線程完成的任務總數
volatile long completedTasks;
// 惟一的構造函數，傳入任務實例firstTask，注意能夠爲null
Worker(Runnable firstTask) {
// 禁止線程中斷，直到runWorker()方法執行
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 經過ThreadFactory建立線程實例，注意一下Worker實例自身做爲Runnable用於建立新的線程實例
this.thread = getThreadFactory().newThread(this);
}
// 委託到外部的runWorker()方法，注意runWorker()方法是線程池的方法，而不是Worker的方法
public void run() {
runWorker(this);
}
// Lock methods
//
// The value 0 represents the unlocked state.
// The value 1 represents the locked state.
// 是否持有獨佔鎖，state值爲1的時候表示持有鎖，state值爲0的時候表示已經釋放鎖
protected boolean isHeldExclusively() {
return getState() != 0;
}
// 獨佔模式下嘗試獲取資源，這裏沒有判斷傳入的變量，直接CAS判斷0更新爲1是否成功，成功則設置獨佔線程爲當前線程
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
// 獨佔模式下嘗試是否資源，這裏沒有判斷傳入的變量，直接把state設置爲0
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(); }
// 啓動後進行線程中斷，注意這裏會判斷線程實例的中斷標誌位是否爲false，只有中斷標誌位爲false纔會中斷
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
Worker的構造函數裏面的邏輯十分重要，經過ThreadFactory建立的Thread實例同時傳入Worker實例，由於Worker自己實現了Runnable，因此能夠做爲任務提交到線程中執行。只要Worker持有的線程實例w調用Thread#start()方法就能在合適時機執行Worker#run()。簡化一下邏輯以下：
// addWorker()方法中構造 Worker worker = createWorker(); // 經過線程池構造時候傳入 ThreadFactory threadFactory = getThreadFactory(); // Worker構造函數中 Thread thread = threadFactory.newThread(worker); // addWorker()方法中啓動 thread.start();
Worker繼承自AQS，這裏使用了AQS的獨佔模式，有個技巧是構造Worker的時候，把AQS的資源（狀態）經過setState(-1)設置爲-1，這是由於Worker實例剛建立時AQS中state的默認值爲0，此時線程還沒有啓動，不能在這個時候進行線程中斷，見Worker#interruptIfStarted()方法。Worker中兩個覆蓋AQS的方法tryAcquire()和tryRelease()都沒有判斷外部傳入的變量，前者直接CAS(0,1)，後者直接setState(0)。接着看核心方法ThreadPoolExecutor#runWorker()：
final void runWorker(Worker w) {
// 獲取當前線程，實際上和Worker持有的線程實例是相同的
Thread wt = Thread.currentThread();
// 獲取Worker中持有的初始化時傳入的任務對象，這裏注意存放在臨時變量task中
Runnable task = w.firstTask;
// 設置Worker中持有的初始化時傳入的任務對象爲null
w.firstTask = null;
// 因爲Worker初始化時AQS中state設置爲-1，這裏要先作一次解鎖把state更新爲0，容許線程中斷
w.unlock(); // allow interrupts
// 記錄線程是否由於用戶異常終結，默認是true
boolean completedAbruptly = true;
try {
// 初始化任務對象不爲null，或者從任務隊列獲取任務不爲空（從任務隊列獲取到的任務會更新到臨時變量task中）
// getTask()因爲使用了阻塞隊列，這個while循環若是命中後半段會處於阻塞或者超時阻塞狀態，getTask()返回爲null會致使線程跳出死循環使線程終結
while (task != null || (task = getTask()) != null) {
// Worker加鎖，本質是AQS獲取資源而且嘗試CAS更新state由0更變爲1
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
// 若是線程池正在中止（也就是由RUNNING或者SHUTDOWN狀態向STOP狀態變動），那麼要確保當前工做線程是中斷狀態
// 不然，要保證當前線程不是中斷狀態
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
// 鉤子方法，任務執行前
beforeExecute(wt, task);
try {
task.run();
// 鉤子方法，任務執行後 - 正常狀況
afterExecute(task, null);
} catch (Throwable ex) {
// 鉤子方法，任務執行後 - 異常狀況
afterExecute(task, ex);
throw ex;
}
} finally {
// 清空task臨時變量，這個很重要，不然while會死循環執行同一個task
task = null;
// 累加Worker完成的任務數
w.completedTasks++;
// Worker解鎖，本質是AQS釋放資源，設置state爲0
w.unlock();
}
}
// 走到這裏說明某一次getTask()返回爲null，線程正常退出
completedAbruptly = false;
} finally {
// 處理線程退出，completedAbruptly爲true說明因爲用戶異常致使線程非正常退出
processWorkerExit(w, completedAbruptly);
}
}
這裏重點拆解分析一下判斷當前工做線程中斷狀態的代碼：
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
// 先簡化一下判斷邏輯，以下
// 判斷線程池狀態是否至少爲STOP，rs >= STOP(1)
boolean atLeastStop = runStateAtLeast(ctl.get(), STOP);
// 判斷線程池狀態是否至少爲STOP，同時判斷當前線程的中斷狀態而且清空當前線程的中斷狀態
boolean interruptedAndAtLeastStop = Thread.interrupted() && runStateAtLeast(ctl.get(), STOP);
if (atLeastStop || interruptedAndAtLeastStop && !wt.isInterrupted()){
wt.interrupt();
}
Thread.interrupted()方法獲取線程的中斷狀態同時會清空該中斷狀態，這裏之因此會調用這個方法是由於在執行上面這個if邏輯同時外部有可能調用shutdownNow()方法，shutdownNow()方法中也存在中斷全部Worker線程的邏輯，可是因爲shutdownNow()方法中會遍歷全部Worker作線程中斷，有可能沒法及時在任務提交到Worker執行以前進行中斷，因此這個中斷邏輯會在Worker內部執行，就是if代碼塊的邏輯。這裏還要注意的是：STOP狀態下會拒絕全部新提交的任務，不會再執行任務隊列中的任務，同時會中斷全部Worker線程。也就是，即便任務Runnable已經runWorker()中前半段邏輯取出，只要還沒走到調用其Runnable#run()，都有可能被中斷。假設恰好發生了進入if代碼塊的邏輯同時外部調用了shutdownNow()方法，那麼if邏輯內會判斷線程中斷狀態而且重置，那麼shutdownNow()方法中調用的interruptWorkers()就不會由於中斷狀態判斷出現問題致使二次中斷線程（會致使異常）。
小結一下上面runWorker()方法的核心流程：
Worker先執行一次解鎖操做，用於解除不可中斷狀態。- 經過
while循環調用getTask()方法從任務隊列中獲取任務（固然，首輪循環也有多是外部傳入的firstTask任務實例）。 - 若是線程池更變爲
STOP狀態，則須要確保工做線程是中斷狀態而且進行中斷處理，不然要保證工做線程必須不是中斷狀態。 - 執行任務實例
Runnale#run()方法，任務實例執行以前和以後（包括正常執行完畢和異常執行狀況）分別會調用鉤子方法beforeExecute()和afterExecute()。 while循環跳出意味着runWorker()方法結束和工做線程生命週期結束（Worker#run()生命週期完結），會調用processWorkerExit()處理工做線程退出的後續工做。
相關文章
- 1. Java線程池ThreadPoolExecutor詳解
- 2. Java：多線程，線程池，ThreadPoolExecutor詳解
- 3. JUC中線程池的使用
- 4. [JUC-10] 線程池
- 5. JUC線程池和spring線程池
- 6. 線程池 ThreadPoolExecutor
- 7. ThreadPoolExecutor 線程池的講解
- 8. JUC--自定義線程池
- 9. ThreadPoolExecutor線程池
- 10. 線程池ThreadPoolExecutor
- 更多相關文章...
- • C# 多線程 - C#教程
- • 免費ARP詳解 - TCP/IP教程
- • Flink 數據傳輸及反壓詳解
- • Scala 中文亂碼解決
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
歡迎關注本站公眾號,獲取更多信息
