Java并發(fā)編程源碼分析系列：
上一篇已經(jīng)對線程池的創(chuàng)建進行了分析，了解線程池既有預(yù)設(shè)的模板驼唱，也提供多種參數(shù)支撐靈活的定制。
本文將會圍繞線程池的生命周期铐尚，分析線程池執(zhí)行任務(wù)的過程。
線程池狀態(tài)
首先認(rèn)識兩個貫穿線程池代碼的參數(shù)：
- runState：線程池運行狀態(tài)
- workerCount：工作線程的數(shù)量
線程池用一個32位的int來同時保存runState和workerCount哆姻，其中高3位是runState宣增，其余29位是workerCount。代碼中會反復(fù)使用runStateOf和workerCountOf來獲取runState和workerCount矛缨。
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;
// 線程池狀態(tài)
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 & ~CAPACITY; }
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
- RUNNING：可接收新任務(wù)爹脾，可執(zhí)行等待隊列里的任務(wù)
- SHUTDOWN：不可接收新任務(wù)帖旨，可執(zhí)行等待隊列里的任務(wù)
- STOP：不可接收新任務(wù)，不可執(zhí)行等待隊列里的任務(wù)灵妨，并且嘗試終止所有在運行任務(wù)
- TIDYING：所有任務(wù)已經(jīng)終止解阅，執(zhí)行terminated()
- TERMINATED：terminated()執(zhí)行完成
線程池狀態(tài)默認(rèn)從RUNNING開始流轉(zhuǎn)，到狀態(tài)TERMINATED結(jié)束泌霍，中間不需要經(jīng)過每一種狀態(tài)货抄，但不能讓狀態(tài)回退。下面是狀態(tài)變化可能的路徑和變化條件：
Worker的創(chuàng)建
線程池是由Worker類負(fù)責(zé)執(zhí)行任務(wù)朱转，Worker繼承了AbstractQueuedSynchronizer蟹地，引出了Java并發(fā)框架的核心AQS。
AbstractQueuedSynchronizer藤为，簡稱AQS怪与，是Java并發(fā)包里一系列同步工具的基礎(chǔ)實現(xiàn)，原理是根據(jù)狀態(tài)位來控制線程的入隊阻塞缅疟、出隊喚醒來處理同步分别。
Worker利用AQS的功能實現(xiàn)對獨占線程變量的設(shè)置，這是一個需要同步的過程存淫。AQS不會在這里展開討論茎杂，有興趣的同學(xué)可以去看分析CountDownLatch的實現(xiàn)原理，里面詳細(xì)介紹了AQS的實現(xiàn)原理纫雁。
調(diào)用execute將會根據(jù)線程池的情況創(chuàng)建Worker，可以歸納出下圖四種情況：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
//1
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//2
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
//3
reject(command);
else if (workerCountOf(recheck) == 0)
//4
addWorker(null, false);
}
//5
else if (!addWorker(command, false))
//6
reject(command);
}
標(biāo)記1對應(yīng)第一種情況倾哺，要留意addWorker傳入了core轧邪，core=true為corePoolSize，core=false為maximumPoolSize羞海，新增時需要檢查workerCount是否超過允許的最大值忌愚。
標(biāo)記2對應(yīng)第二種情況，檢查線程池是否在運行却邓，并且將任務(wù)加入等待隊列硕糊。標(biāo)記3再檢查一次線程池狀態(tài)，如果線程池忽然處于非運行狀態(tài)腊徙，那就將等待隊列剛加的任務(wù)刪掉简十，再交給RejectedExecutionHandler處理。標(biāo)記4發(fā)現(xiàn)沒有worker撬腾，就先補充一個空任務(wù)的worker螟蝙。
標(biāo)記5對應(yīng)第三種情況，等待隊列不能再添加任務(wù)了民傻，調(diào)用addWorker添加一個去處理胰默。
標(biāo)記6對應(yīng)第四種情況场斑，addWorker的core傳入false，返回調(diào)用失敗牵署，代表workerCount已經(jīng)超出maximumPoolSize漏隐，那就交給RejectedExecutionHandler處理。
private boolean addWorker(Runnable firstTask, boolean core) {
//1
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
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;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
//2
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
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 rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && 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;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
標(biāo)記1的第一段代碼奴迅，目的很簡單青责，是為workerCount加一。至于為什么代碼寫了這么長半沽，是因為線程池的狀態(tài)在不斷變化爽柒，并發(fā)環(huán)境下需要保證變量的同步性。外循環(huán)判斷線程池狀態(tài)者填、任務(wù)非空和隊列非空浩村，內(nèi)循環(huán)使用CAS機制保證workerCount正確地遞增。不了解CAS可以看認(rèn)識非阻塞的同步機制CAS占哟，后續(xù)增減workerCount都會使用CAS心墅。
標(biāo)記2的第二段代碼，就比較簡單榨乎。創(chuàng)建一個新Worker對象怎燥，將Worker添加進workers里（Set集合）。成功添加后蜜暑，啟動worker里的線程铐姚。在finally里判斷線程是否啟動成功，不成功直接調(diào)用addWorkerFailed肛捍。
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (w != null)
workers.remove(w);
decrementWorkerCount();
tryTerminate();
} finally {
mainLock.unlock();
}
}
addWorkerFailed將減少已經(jīng)遞增的workerCount隐绵，并且調(diào)用tryTerminate結(jié)束線程池。
Worker的執(zhí)行
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this);
}
Worker在構(gòu)造函數(shù)里采用ThreadFactory創(chuàng)建Thread拙毫，在run方法里調(diào)用了runWorker依许，看來是真正執(zhí)行任務(wù)的地方。
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//1
while (task != null || (task = getTask()) != null) {
w.lock();
//2
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//3
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;
//4
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false; //5
} finally {
//6
processWorkerExit(w, completedAbruptly);
}
}
標(biāo)記1進入循環(huán)缀蹄，從getTask獲取要執(zhí)行的任務(wù)峭跳，直到返回null。這里達到了線程復(fù)用的效果缺前，讓線程處理多個任務(wù)蛀醉。
標(biāo)記2是一個比較復(fù)雜的判斷，保證了線程池在STOP狀態(tài)下線程是中斷的衅码，非STOP狀態(tài)下線程沒有被中斷滞欠。如果你不了解Java的中斷機制，看如何正確結(jié)束Java線程這篇肆良。
標(biāo)記3調(diào)用了run方法筛璧，真正執(zhí)行了任務(wù)逸绎。執(zhí)行前后提供了beforeExecute和afterExecute兩個方法，由子類實現(xiàn)夭谤。
標(biāo)記4里的completedTasks統(tǒng)計worker執(zhí)行了多少任務(wù)棺牧，最后累加進completedTaskCount變量，可以調(diào)用相應(yīng)方法返回一些統(tǒng)計信息朗儒。
標(biāo)記5的變量completedAbruptly表示worker是否異常終止颊乘，執(zhí)行到這里代表執(zhí)行正常，后續(xù)的方法需要這個變量醉锄。
標(biāo)記6調(diào)用processWorkerExit結(jié)束乏悄，后面會分析。
接著來看worker從等待隊列獲取任務(wù)的getTask方法：
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//1
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
//2
// Are workers subject to culling?
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
//3
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
標(biāo)記1檢查線程池的狀態(tài)恳不，這里就體現(xiàn)出SHUTDOWN和STOP的區(qū)別檩小。如果線程池是SHUTDOWN狀態(tài)，還會先處理完等待隊列的任務(wù)烟勋；如果是STOP狀態(tài)规求，就不再處理等待隊列里的任務(wù)了。
標(biāo)記2先看allowCoreThreadTimeOut這個變量卵惦，false時worker空閑阻肿，也不會結(jié)束；true時沮尿，如果worker空閑超過keepAliveTime丛塌，就會結(jié)束。接著是一個很復(fù)雜的判斷畜疾，好難轉(zhuǎn)成文字描述姨伤，自己看吧。注意一下wc>maximumPoolSize庸疾，出現(xiàn)這種可能是在運行中調(diào)用setMaximumPoolSize，還有wc>1当编，在等待隊列非空時届慈，至少保留一個worker。
標(biāo)記3是從等待隊列取任務(wù)的邏輯忿偷，根據(jù)timed分為等待keepAliveTime或者阻塞直到有任務(wù)金顿。
最后來看結(jié)束worker需要執(zhí)行的操作：
private void processWorkerExit(Worker w, boolean completedAbruptly) {
//1
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
//2
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
//3
tryTerminate();
int c = ctl.get();
//4
if (runStateLessThan(c, STOP)) {
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);
}
}
正常情況下，在getTask里就會將workerCount減一鲤桥。標(biāo)記1處用變量completedAbruptly判斷worker是否異常退出揍拆，如果是，需要補充對workerCount的減一茶凳。
標(biāo)記2將worker處理任務(wù)的數(shù)量累加到總數(shù)嫂拴，并且在集合workers中去除播揪。
標(biāo)記3嘗試終止線程池，后續(xù)會研究猪狈。
標(biāo)記4處理線程池還是RUNNING或SHUTDOWN狀態(tài)時，如果worker是異常結(jié)束辩恼，那么會直接addWorker雇庙。如果allowCoreThreadTimeOut=true，并且等待隊列有任務(wù)灶伊，至少保留一個worker疆前；如果allowCoreThreadTimeOut=false，workerCount不少于corePoolSize聘萨。
總結(jié)一下worker：線程池啟動后竹椒，worker在池內(nèi)創(chuàng)建，包裝了提交的Runnable任務(wù)并執(zhí)行匈挖，執(zhí)行完就等待下一個任務(wù)碾牌，不再需要時就結(jié)束。
線程池的關(guān)閉
線程池的關(guān)閉不是一關(guān)了事儡循，worker在池里處于不同狀態(tài)舶吗，必須安排好worker的"后事"，才能真正釋放線程池择膝。ThreadPoolExecutor提供兩種方法關(guān)閉線程池：
- shutdown：不能再提交任務(wù)誓琼，已經(jīng)提交的任務(wù)可繼續(xù)運行；
- shutdownNow：不能再提交任務(wù)肴捉，已經(jīng)提交但未執(zhí)行的任務(wù)不能運行腹侣，在運行的任務(wù)可繼續(xù)運行，但會被中斷齿穗，返回已經(jīng)提交但未執(zhí)行的任務(wù)傲隶。
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess(); //1 安全策略機制
advanceRunState(SHUTDOWN); //2
interruptIdleWorkers(); //3
onShutdown(); //4 空方法，子類實現(xiàn)
} finally {
mainLock.unlock();
}
tryTerminate(); //5
}
shutdown將線程池切換到SHUTDOWN狀態(tài)窃页，并調(diào)用interruptIdleWorkers請求中斷所有空閑的worker跺株，最后調(diào)用tryTerminate嘗試結(jié)束線程池。
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);
interruptWorkers();
tasks = drainQueue(); //1
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
shutdownNow和shutdown類似脖卖，將線程池切換為STOP狀態(tài)乒省，中斷目標(biāo)是所有worker。drainQueue會將等待隊列里未執(zhí)行的任務(wù)返回畦木。
interruptIdleWorkers和interruptWorkers實現(xiàn)原理都是遍歷workers集合袖扛，中斷條件符合的worker。
上面的代碼多次出現(xiàn)調(diào)用tryTerminate十籍，這是一個嘗試將線程池切換到TERMINATED狀態(tài)的方法蛆封。
final void tryTerminate() {
for (;;) {
int c = ctl.get();
//1
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
//2
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
//3
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
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
}
}
標(biāo)記1檢查線程池狀態(tài)唇礁，下面幾種情況，后續(xù)操作都沒有必要娶吞，直接return垒迂。
- RUNNING（還在運行，不能停）
- TIDYING或TERMINATED（已經(jīng)沒有在運行的worker）
- SHUTDOWN并且等待隊列非空（執(zhí)行完才能停）
標(biāo)記2在worker非空的情況下又調(diào)用了interruptIdleWorkers妒蛇，你可能疑惑在shutdown時已經(jīng)調(diào)用過了机断，為什么又調(diào)用，而且每次只中斷一個空閑worker绣夺？你需要知道吏奸，shutdown時worker可能在執(zhí)行中，執(zhí)行完阻塞在隊列的take陶耍，不知道要結(jié)束奋蔚，所有要補充調(diào)用interruptIdleWorkers。每次只中斷一個是因為processWorkerExit時烈钞，還會執(zhí)行tryTerminate泊碑，自動中斷下一個空閑的worker。
標(biāo)記3是最終的狀態(tài)切換毯欣。線程池會先進入TIDYING狀態(tài)馒过，再進入TERMINATED狀態(tài)，中間提供了terminated這個空方法供子類實現(xiàn)酗钞。
調(diào)用關(guān)閉線程池方法后腹忽，需要等待線程池切換到TERMINATED狀態(tài)。awaitTermination檢查限定時間內(nèi)線程池是否進入TERMINATED狀態(tài)砚作，代碼如下：
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;
nanos = termination.awaitNanos(nanos);
}
} finally {
mainLock.unlock();
}
}
后言
以上過了一遍線程池主要的邏輯窘奏，總體來看線程池的設(shè)計是很清晰的。如有錯誤或不足葫录，歡迎指出着裹，也歡迎留言交流。今次介紹了線程池運行的生命周期米同，下篇會研究更細(xì)粒度地控制任務(wù)的生命周期骇扇，也就是submit和Future。
