Java執行緒池ThreadPoolExecutor實現原理剖析
摘要：
【編者的話】在Java中，使用執行緒池來非同步執行一些耗時任務是非常常見的操作。最初我們一般都是直接使用new Thread().start的方式，但我們知道，執行緒的建立和銷燬都會耗費大量的資源，關於執行緒可以參考之前的一篇部落格《 Java執行緒那點事兒 》，因此我們需要重用執行緒資源。...
【編者的話】在Java中，使用執行緒池來非同步執行一些耗時任務是非常常見的操作。最初我們一般都是直接使用new Thread().start的方式，但我們知道，執行緒的建立和銷燬都會耗費大量的資源，關於執行緒可以參考之前的一篇部落格《 ofollow,noindex" target="_blank">Java執行緒那點事兒 》，因此我們需要重用執行緒資源。
當然也有其他待解決方案，比如說coroutine，目前Kotlin已經支援了，JDK也已經有了相關的提案： Project Loom ，目前的實現方式和Kotlin有點類似，都是基於ForkJoinPool，當然目前還有很多限制以及問題沒解決，比如synchronized還是鎖住當前執行緒等。
繼承結構
繼承結構看起來很清晰，最頂層的Executor只提供了一個最簡單的void execute(Runnable command)方法，然後是ExecutorService，ExecutorService提供了一些管理相關的方法，例如關閉、判斷當前執行緒池的狀態等，另外不同於Executor#execute，ExecutorService提供了一系列方法，可以將任務包裝成一個Future，從而使得任務提交方可以跟蹤任務的狀態。而父類AbstractExecutorService則提供了一些預設的實現。
構造器
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.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
- corePoolSize、 maximumPoolSize。執行緒池會自動根據corePoolSize和maximumPoolSize去調整當前執行緒池的大小。當你通過submit或者execute方法提交任務的時候，如果當前執行緒池的執行緒數小於corePoolSize,那麼執行緒池就會建立一個新的執行緒處理任務, 即使其他的core執行緒是空閒的。如果當前執行緒數大於corePoolSize並且小於maximumPoolSize，那麼只有在佇列"滿"的時候才會建立新的執行緒。因此這裡會有很多的坑，比如你的core和max執行緒數設定的不一樣，希望請求積壓在佇列的時候能夠實時的擴容，但如果制定了一個無界佇列，那麼就不會擴容了，因為佇列不存在滿的概念。
- keepAliveTime。如果當前執行緒池中的執行緒數超過了corePoolSize，那麼如果在keepAliveTime時間內都沒有新的任務需要處理，那麼超過corePoolSize的這部分執行緒就會被銷燬。預設情況下是不會回收core執行緒的，可以通過設定allowCoreThreadTimeOut改變這一行為。
- workQueue。即實際用於儲存任務的佇列，這個可以說是最核心的一個引數了，直接決定了執行緒池的行為，比如說傳入一個有界佇列，那麼佇列滿的時候，執行緒池就會根據core和max引數的設定情況決定是否需要擴容，如果傳入了一個SynchronousQueue，這個佇列只有在另一個執行緒在同步remove的時候才可以put成功，對應到執行緒池中，簡單來說就是如果有執行緒池任務處理完了，呼叫poll或者take方法獲取新的任務的時候，新提交的任務才會put成功，否則如果當前的執行緒都在忙著處理任務，那麼就會put失敗，也就會走擴容的邏輯，如果傳入了一個DelayedWorkQueue，顧名思義，任務就會根據過期時間來決定什麼時候彈出，即為ScheduledThreadPoolExecutor的機制。
- threadFactory。建立執行緒都是通過ThreadFactory來實現的，如果沒指定的話，預設會使用Executors.defaultThreadFactory()，一般來說，我們會在這裡對執行緒設定名稱、異常處理器等。
- handler。即當任務提交失敗的時候，會呼叫這個處理器，ThreadPoolExecutor內建了多個實現，比如拋異常、直接拋棄等。這裡也需要根據業務場景進行設定，比如說當佇列積壓的時候，針對性的對執行緒池擴容或者傳送告警等策略。
看完這幾個引數的含義，我們看一下Executors提供的一些工具方法，只要是為了方便使用，但是我建議最好少用這個類，而是直接用ThreadPoolExecutor的建構函式，多瞭解一下這幾個引數到底是什麼意思，自己的業務場景是什麼樣的，比如執行緒池需不需要擴容、用不用回收空閒的執行緒等。
public class Executors {
/*
* 提供一個固定大小的執行緒池，並且執行緒不會回收，由於傳入的是一個無界佇列，相當於佇列永遠不會滿
* 也就不會擴容，因此需要特別注意任務積壓在佇列中導致記憶體爆掉的問題
*/
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
/*
*這個執行緒池會一直擴容，由於SynchronousQueue的特性，如果當前所有的執行緒都在處理任務，那麼
*新的請求過來，就會導致建立一個新的執行緒處理任務。如果執行緒一分鐘沒有新任務處理，就會被回
*收掉。特別注意，如果每一個任務都比較耗時，併發又比較高，那麼可能每次任務過來都會建立一個線
*程
*/
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
}
原始碼分析
既然是個執行緒池，那就必然有其生命週期：執行中、關閉、停止等。ThreadPoolExecutor是用一個AtomicInteger去的前三位表示這個狀態的，另外又重用了低29位用於表示執行緒數，可以支援最大大概5億多，絕逼夠用了，如果以後硬體真的發展到能夠啟動這麼多執行緒，改成AtomicLong就可以了。
狀態這裡主要分為下面幾種：
- RUNNING：表示當前執行緒池正在執行中，可以接受新任務以及處理佇列中的任務
- SHUTDOWN：不再接受新的任務，但會繼續處理佇列中的任務
- STOP：不再接受新的任務，也不處理佇列中的任務了，並且會中斷正在進行中的任務
- TIDYING：所有任務都已經處理完畢，執行緒數為0，轉為為TIDYING狀態之後，會呼叫terminated()回撥
- TERMINATED：terminated()已經執行完畢
同時我們可以看到所有的狀態都是用二進位制位表示的，並且依次遞增，從而方便進行比較，比如想獲取當前狀態是否至少為SHUTDOWN等，同時狀態之前有幾種轉換：
- RUNNING -> SHUTDOWN。呼叫了shutdown()之後，或者執行了finalize()
- (RUNNING 或者 SHUTDOWN) -> STOP。呼叫了shutdownNow()之後會轉換這個狀態
- SHUTDOWN -> TIDYING。當執行緒池和佇列都為空的時候
- STOP -> TIDYING。當執行緒池為空的時候
- IDYING -> TERMINATED。執行完terminated()回撥之後會轉換為這個狀態
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static final int COUNT_BITS = Integer.SIZE - 3; private static final int CAPACITY= (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; //由於前三位表示狀態，因此將CAPACITY取反，和進行與操作即可 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; } 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修改worker的數目 private boolean compareAndIncrementWorkerCount(int expect) { return ctl.compareAndSet(expect, expect + 1); } //只嘗試一次，失敗了則返回，是否重試由呼叫方決定 private boolean compareAndDecrementWorkerCount(int expect) { return ctl.compareAndSet(expect, expect - 1); } //跟上一個不一樣，會一直重試 private void decrementWorkerCount() { do {} while (! compareAndDecrementWorkerCount(ctl.get())); }
下面是比較核心的欄位，這裡workers採用的是非執行緒安全的HashSet，而不是執行緒安全的版本，主要是因為這裡有些複合的操作，比如說將worker新增到workers後，我們還需要判斷是否需要更新largestPoolSize等，workers只在獲取到mainLock的情況下才會進行讀寫，另外這裡的mainLock也用於在中斷執行緒的時候序列執行，否則如果不加鎖的話，可能會造成併發去中斷執行緒，引起不必要的中斷風暴。
private final ReentrantLock mainLock = new ReentrantLock(); private final HashSet<Worker> workers = new HashSet<Worker>(); private final Condition termination = mainLock.newCondition(); private int largestPoolSize; private long completedTaskCount;
核心方法
拿到一個執行緒池之後，我們就可以開始提交任務，讓它去執行了，那麼我們看一下submit方法是如何實現的。
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
這兩個方法都很簡單，首先將提交過來的任務（有兩種形式：Callable、Runnable）都包裝成統一的RunnableFuture，然後呼叫execute方法，execute可以說是執行緒池最核心的一個方法。
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
/*
獲取當前worker的數目，如果小於corePoolSize那麼就擴容，
這裡不會判斷是否已經有core執行緒，而是隻要小於corePoolSize就會直接增加worker
*/
if (workerCountOf(c) < corePoolSize) {
/*
呼叫addWorker(Runnable firstTask, boolean core)方法擴容
firstTask表示為該worker啟動之後要執行的第一個任務，core表示要增加的為core執行緒
*/
if (addWorker(command, true))
return;
//如果增加失敗了那麼重新獲取ctl的快照,比如可能執行緒池在這期間關閉了
c = ctl.get();
}
/*
如果當前執行緒池正在執行中，並且將任務丟到佇列中成功了，
那麼就會進行一次double check,看下在這期間執行緒池是否關閉了，
如果關閉了，比如處於SHUTDOWN狀態，如上文所講的，SHUTDOWN狀態的時候，
不再接受新任務，remove成功後呼叫拒絕處理器。而如果仍然處於執行中的狀態，
那麼這裡就double check下當前的worker數，如果為0，有可能在上述邏輯的執行
過程中，有worker銷燬了，比如說任務丟擲了未捕獲異常等，那麼就會進行一次擴容，
但不同於擴容core執行緒，這裡由於任務已經丟到佇列中去了，因此就不需要再傳遞firstTask了，
同時要注意，這裡擴容的是非core執行緒
*/
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
/*
如果在上一步中，將任務丟到佇列中失敗了，那麼就進行一次擴容，
這裡會將任務傳遞到firstTask引數中，並且擴容的是非core執行緒，
如果擴容失敗了，那麼就執行拒絕策略。
*/
reject(command);
}
這裡要特別注意下防止佇列失敗的邏輯，不同的佇列丟任務的邏輯也不一樣，例如說無界佇列，那麼就永遠不會put失敗，也就是說擴容也永遠不會執行，如果是有界佇列，那麼當佇列滿的時候，會擴容非core執行緒，如果是SynchronousQueue，這個佇列比較特殊，當有另外一個執行緒正在同步獲取任務的時候，你才能put成功，因此如果當前執行緒池中所有的worker都忙著處理任務的時候，那麼後續的每次新任務都會導致擴容，當然如果worker沒有任務處理了，阻塞在獲取任務這一步的時候，新任務的提交就會直接丟到佇列中去，而不會擴容。
上文中多次提到了擴容，那麼我們下面看一下執行緒池具體是如何進行擴容的：
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
//獲取當前執行緒池的狀態
int rs = runStateOf(c);
/*
如果狀態為大於SHUTDOWN, 比如說STOP，STOP上文說過佇列中的任務不處理了，也不接受新任務，
因此可以直接返回false不擴容了，如果狀態為SHUTDOWN並且firstTask為null，同時佇列非空，
那麼就可以擴容
*/
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
/*
若worker的數目大於CAPACITY則直接返回，
然後根據要擴容的是core執行緒還是非core執行緒，進行判斷worker數目
是否超過設定的值，超過則返回
*/
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
/*
通過CAS的方式自增worker的數目，成功了則直接跳出迴圈
*/
if (compareAndIncrementWorkerCount(c))
break retry;
//重新讀取狀態變數，如果狀態改變了，比如執行緒池關閉了，那麼就跳到最外層的for迴圈，
//注意這裡跳出的是retry。
c = ctl.get();// Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//建立Worker
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
/*
獲取鎖，並判斷執行緒池是否已經關閉
*/
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // 若執行緒已經啟動了，比如說已經呼叫了start()方法，那麼就拋異常，
throw new IllegalThreadStateException();
//新增到workers中
workers.add(w);
int s = workers.size();
if (s > largestPoolSize) //更新largestPoolSize
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
//若Worker建立成功，則啟動執行緒，這麼時候worker就會開始執行任務了
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
//新增失敗
addWorkerFailed(w);
}
return workerStarted;
}
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (w != null)
workers.remove(w);
decrementWorkerCount();
//每次減少worker或者從佇列中移除任務的時候都需要呼叫這個方法
tryTerminate();
} finally {
mainLock.unlock();
}
}
這裡有個貌似不太起眼的方法tryTerminate，這個方法會在所有可能導致執行緒池終結的地方呼叫，比如說減少worker的數目等，如果滿足條件的話，那麼將執行緒池轉換為TERMINATED狀態。另外這個方法沒有用private修飾，因為ScheduledThreadPoolExecutor繼承自ThreadPoolExecutor，而ScheduledThreadPoolExecutor也會呼叫這個方法。
final void tryTerminate() {
for (;;) {
int c = ctl.get();
/*
如果當前執行緒處於執行中、TIDYING、TERMINATED狀態則直接返回，執行中的沒
什麼好說的，後面兩種狀態可以說執行緒池已經正在終結了，另外如果處於SHUTDOWN狀態，
並且workQueue非空，表明還有任務需要處理，也直接返回
*/
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
//可以退出，但是執行緒數非0，那麼就中斷一個執行緒，從而使得關閉的訊號能夠傳遞下去,
//中斷worker後，worker捕獲異常後，會嘗試退出，並在這裡繼續執行tryTerminate()方法，
//從而使得訊號傳遞下去
if (workerCountOf(c) != 0) {
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//嘗試轉換成TIDYING狀態，執行完terminated回撥之後
//會轉換為TERMINATED狀態，這個時候執行緒池已經完整關閉了，
//通過signalAll方法，喚醒所有阻塞在awaitTermination上的執行緒
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated();
} finally {
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
/**
* 中斷空閒的執行緒
* @param onlyOne
*/
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
//遍歷所有worker，若之前沒有被中斷過，
//並且獲取鎖成功，那麼就嘗試中斷。
//鎖能夠獲取成功，那麼表明當前worker沒有在執行任務，而是在
//獲取任務，因此也就達到了只中斷空閒執行緒的目的。
Thread t = w.thread;
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
Worker
下面看一下Worker類，也就是這個類實際負責執行任務，Worker類繼承自AbstractQueuedSynchronizer，AQS可以理解為一個同步框架，提供了一些通用的機制，利用模板方法模式，讓你能夠原子的管理同步狀態、blocking和unblocking執行緒、以及佇列，具體的內容之後有時間會再寫，還是比較複雜的。這裡Worker對AQS的使用相對比較簡單，使用了狀態變數state表示是否獲得鎖，0表示解鎖、1表示已獲得鎖，同時通過exclusiveOwnerThread儲存當前持有鎖的執行緒。另外再簡單提一下，比如說CountDownLatch, 也是基於AQS框架實現的，countdown方法遞減state，await阻塞等待state為0。
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/** Thread this worker is running in.Null if factory fails. */
final Thread thread;
/** Initial task to run.Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker*/
public void run() {
runWorker(this);
}
protected boolean isHeldExclusively() {
return getState() != 0;
}
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) {
}
}
}
}
注意這裡Worker初始化的時候，會通過setState(-1)將state設定為-1，並在runWorker()方法中置為0，上文說過Worker是利用state這個變數來表示鎖的狀態，那麼加鎖的操作就是通過CAS將state從0改成1，那麼初始化的時候改成-1，也就是表示在Worker啟動之前，都不允許加鎖操作，我們再看interruptIfStarted()以及interruptIdleWorkers()方法，這兩個方法在嘗試中斷Worker之前，都會先加鎖或者判斷state是否大於0，因此這裡的將state設定為-1，就是為了禁止中斷操作，並在runWorker中置為0，也就是說只能在Worker啟動之後才能夠中斷Worker。
另外執行緒啟動之後，其實就是呼叫了runWorker方法，下面我們看一下具體是如何實現的。
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // 呼叫unlock()方法，將state置為0，表示其他操作可以獲得鎖或者中斷worker
boolean completedAbruptly = true;
try {
/*
首先嚐試執行firstTask，若沒有的話，則呼叫getTask()從佇列中獲取任務
*/
while (task != null || (task = getTask()) != null) {
w.lock();
/*
如果執行緒池正在關閉，那麼中斷執行緒。
*/
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//執行beforeExecute回撥
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回撥
afterExecute(task, thrown);
}
} finally {
task = null;
//這裡加了鎖，因此沒有執行緒安全的問題，volatile修飾保證其他執行緒的可見性
w.completedTasks++;
w.unlock();//解鎖
}
}
completedAbruptly = false;
} finally {
//拋異常了，或者當前佇列中已沒有任務需要處理等
processWorkerExit(w, completedAbruptly);
}
}
private void processWorkerExit(Worker w, boolean completedAbruptly) {
//如果是異常終止的，那麼減少worker的數目
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//將當前worker中workers中刪除掉，並累加當前worker已執行的任務到completedTaskCount中
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
//上文說過，減少worker的操作都需要呼叫這個方法
tryTerminate();
/*
如果當前執行緒池仍然是執行中的狀態，那麼就看一下是否需要新增另外一個worker替換此worker
*/
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
/*
如果是異常結束的則直接擴容，否則的話則為正常退出，比如當前佇列中已經沒有任務需要處理，
如果允許core執行緒超時的話，那麼看一下當前佇列是否為空，空的話則不用擴容。否則話看一下
是否少於corePoolSize個worker在執行。
*/
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}
private Runnable getTask() {
boolean timedOut = false; // 上一次poll()是否超時了
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 若執行緒池關閉了(狀態大於STOP)
// 或者執行緒池處於SHUTDOWN狀態，但是佇列為空，那麼返回null
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
/*
如果允許core執行緒超時 或者 不允許core執行緒超時但當前worker的數目大於core執行緒數，
那麼下面的poll()則超時呼叫
*/
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
/*
獲取任務超時了並且(當前執行緒池中還有不止一個worker 或者 佇列中已經沒有任務了)，那麼就嘗試
減少worker的數目，若失敗了則重試
*/
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
//從佇列中抓取任務
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
//走到這裡表明，poll呼叫超時了
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
關閉執行緒池
關閉執行緒池一般有兩種形式，shutdown()和shutdownNow()。
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
//通過CAS將狀態更改為SHUTDOWN，這個時候執行緒池不接受新任務，但會繼續處理佇列中的任務
advanceRunState(SHUTDOWN);
//中斷所有空閒的worker,也就是說除了正在處理任務的worker，其他阻塞在getTask()上的worker
//都會被中斷
interruptIdleWorkers();
//執行回撥
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
//這個方法不會等待所有的任務處理完成才返回
}
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
/*
不同於shutdown()，會轉換為STOP狀態，不再處理新任務，佇列中的任務也不處理，
而且會中斷所有的worker，而不只是空閒的worker
*/
advanceRunState(STOP);
interruptWorkers();
tasks = drainQueue();//將所有的任務從佇列中彈出
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
private List<Runnable> drainQueue() {
BlockingQueue<Runnable> q = workQueue;
ArrayList<Runnable> taskList = new ArrayList<Runnable>();
/*
將佇列中所有的任務remove掉，並新增到taskList中，
但是有些佇列比較特殊，比如說DelayQueue，如果第一個任務還沒到過期時間，則不會彈出，
因此這裡通過呼叫toArray方法，然後再一個一個的remove掉
*/
q.drainTo(taskList);
if (!q.isEmpty()) {
for (Runnable r : q.toArray(new Runnable[0])) {
if (q.remove(r))
taskList.add(r);
}
}
return taskList;
}
從上文中可以看到，呼叫了shutdown()方法後，不會等待所有的任務處理完畢才返回，因此需要呼叫awaitTermination()來實現。
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (;;) {
//執行緒池若已經終結了，那麼就返回
if (runStateAtLeast(ctl.get(), TERMINATED))
return true;
//若超時了，也返回掉
if (nanos <= 0)
return false;
//阻塞在訊號量上，等待執行緒池終結,但是要注意這個方法可能會因為一些未知原因隨時喚醒當前執行緒，
//因此需要重試，在tryTerminate()方法中，執行完terminated()回撥後，表明執行緒池已經終結了,
//然後會通過termination.signalAll()喚醒當前執行緒
nanos = termination.awaitNanos(nanos);
}
} finally {
mainLock.unlock();
}
}
一些統計相關的方法
public int getPoolSize() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//若執行緒已終結則直接返回0，否則計算works中的數目
//想一下為什麼不用workerCount呢?
return runStateAtLeast(ctl.get(), TIDYING) ? 0
: workers.size();
} finally {
mainLock.unlock();
}
}
public int getActiveCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int n = 0;
for (Worker w : workers)
if (w.isLocked())//上鎖的表明worker當前正在處理任務，也就是活躍的worker
++n;
return n;
} finally {
mainLock.unlock();
}
}
public int getLargestPoolSize() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
return largestPoolSize;
} finally {
mainLock.unlock();
}
}
//獲取任務的總數，這個方法慎用，若是個無解佇列，或者佇列擠壓比較嚴重，會很蛋疼
public long getTaskCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
long n = completedTaskCount;//比如有些worker被銷燬後，其處理完成的任務就會疊加到這裡
for (Worker w : workers) {
n += w.completedTasks;//疊加歷史處理完成的任務
if (w.isLocked())//上鎖表明正在處理任務,也算一個
++n;
}
return n + workQueue.size();//獲取佇列中的數目
} finally {
mainLock.unlock();
}
}
public long getCompletedTaskCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
long n = completedTaskCount;
for (Worker w : workers)
n += w.completedTasks;
return n;
} finally {
mainLock.unlock();
}
}
總結
這篇部落格基本上覆蓋了執行緒池的方方面面，但仍然有非常多的細節可以深究，比如說異常的處理，可以參照之前的一篇部落格：《 深度解析Java執行緒池的異常處理機制 》，另外還有AQS、unsafe等可以之後再單獨總結。