五一假期大雄看了一本《java併發編程藝術》，瞭解了線程池的基本工作流程，竟然發現線程池工作原理和互聯網公司運作模式十分相似。 線程池處理流程 原理解析 互聯網公司與線程池的關係 這裡用一個比喻來描述一下線程池，中間有一些名詞你可能不是太清楚，後邊源碼解析的部分會講到。 你可以把 線程池 看作是一個 ...
五一假期大雄看了一本《java併發編程藝術》，瞭解了線程池的基本工作流程，竟然發現線程池工作原理和互聯網公司運作模式十分相似。
線程池處理流程
原理解析
互聯網公司與線程池的關係
這裡用一個比喻來描述一下線程池，中間有一些名詞你可能不是太清楚，後邊源碼解析的部分會講到。
你可以把線程池看作是一個研發部門，研發部門有很多程式員（Worker), 他們在一個大辦公室里（HashSet workers)。程式員乾不完的需求（Runnable/Callable)放在需求池（workQueue）里排隊。每個研發部都配置有骨幹程式員數量（corePoolSize）和最大能容納的程式員數量（maximumPoolSize）。具體要做的任務就是產品的需求。
new 一個線程池相當於創建了一個研發部，創建研發部時需要指定骨幹程式員數量，最大能容納的程式員數量，需求池用哪種（BlockingQueue）,如果忙不過來的需求怎麼給產品回覆（拒絕策略）等等內容。剛開始這個研發部一個程式員也沒有。
當產品給這個研發部提一個需求時（當然肯定不會只提一個，他們會不斷的提需求。這裡以提一個需求為例）
首先會看骨幹程式員招聘滿了沒。
如果沒滿，會招聘一個骨幹程式員，招聘進來就讓他不停的工作（很殘酷啊），幹完剛派過來的任務他會主動在需求池找下一個需求來做（好員工），如果需求池沒有需求了，他就停止工作了，然後研發部會把他裁掉，如果裁掉後發現骨幹程式員數量不夠了，就會再招聘一個程式員。裁掉後，要是骨幹程式員數量還夠就不招聘了。
如果骨幹程式員數量滿了，就看需求池滿沒滿，如果需求池沒滿，就把需求扔進需求池裡；如果需求池滿了，就看程式員數量有沒有達到上限，如果達到了，就對產品說，這個需求我們做不了，沒資源；如果沒達到，就招聘一個程式員，招聘進來就讓他不停的工作，幹完剛派過來的需求他會主動到需求池找下一個任務來做，如果需求池沒有任務了，他就停止工作了，然後研發部會把他裁掉，如果裁掉後發現骨幹程式員數量不夠了，就會再招聘一個程式員。裁掉後，要是骨幹程式員數量還夠就不招聘了。
源碼解析
首先是worker（程式員）
Worker被裝在一個HashSet（workers）裡邊， 他是用來執行任務的，他們的職責就是不斷的從workQueue裡邊取任務，然後執行。當workQueue（需求池）裡邊拿不到任務，或者線程池達到特定狀態，worker就會從workers裡邊移走（被裁）。
下邊是Worker源碼，移除了非關鍵的東西
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
// 標識這個任務是在哪個線程運行
final Thread thread;
Runnable firstTask;
// 完成了幾個任務
volatile long completedTasks;
Worker(Runnable firstTask) {
// 阻止中斷，知道runWorker執行
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 直接用你提供的線程工廠搞個線程出來
this.thread = getThreadFactory().newThread(this);
}
// 調用ThreadPoolExecutor裡邊的runWorker方法
public void run() {
runWorker(this);
}
// 以下這些是AQS相關的東西
// 0代表沒有加鎖
// 1代表加鎖了
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實現了Runnable介面，所以他是個任務，有run方法；同時有繼承了AQS，所以他也是一把鎖。
下邊是提交任務的過程
提交任務有submit和execute, submit就是首先將Callable或者Runnable包裝成FutureTask，然後調用execute, 所以核心是分析execute
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
// 這個c裡邊有兩個信息，一個是現在有多少worker, 另一個是現線上程池的狀態是啥
// workerCountOf方法就是從裡邊提取 worker的數量的
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) { // 當前worker的數量比需要的核心線程數少
// 加worker去執行,加成功就完事了，也就是說只要worker比核心線程數少，就會創建worker
// 不管現在核心線程是否在工作，也不管workQueue是不是滿的
// addWorker的第二個參數表示是不是要加核心線程（或者叫核心worker)
if (addWorker(command, true))
return;
c = ctl.get();
}
// 當前worker達到或超過了核心線程數，或者加worker失敗了，才會走下邊的流程
// worker已經比核心線程數多了
// 如果 線程池沒有shutdown的話
// 就嘗試將任務加到workQueue裡邊，工作隊列入隊成功的話再往裡邊走
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
// 再次檢查狀態如果線程池要停了，那麼就拒絕任務，並且把worker從工作隊列扔掉
reject(command);
else if (workerCountOf(recheck) == 0)
// 如果沒有worker的話(說明沒加進去，這種場景我沒想到是什麼情況)，加一個worker
addWorker(null, false);
// 其他情況，丟到工作隊列就不用管了，等著worker去處理
}
// 如果隊列滿了加失敗了，或者線程池狀態不滿足了，就嘗試加普通worker(非核心線程)
else if (!addWorker(command, false))
// 加失敗了就拒絕任務
// 失敗一方面可能是worker數量已經達到你的給的maximumPoolSize
// 另一方面，可能是檢查到線程池的狀態不對了
reject(command);
}
可以發現execute方法就是完成了上邊說的“線程池處理流程”這個圖裡描述的過程。 大雄看到這裡還有幾個疑問，一個是Woker是如何創建並加入workers的，一個是worker是如何啟動的，再就是worker是如何運行的
生活還要繼續
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 做一些校驗，線程池的狀態要滿足一定條件
// 而且得提交任務過來，再就是workQueue不能是空的
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
// 看你是要創建核心worker還是普通worker
// 核心看超沒超過corePoolSize, 普通看超沒超過maximumPoolSize
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))
// 增加worker數量失敗就在來
break retry;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
// 中途線程池狀態發生變化了
continue retry;
}
}
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;
// 加worker是要加全局鎖的
mainLock.lock();
try {
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) {
// worker是在這裡啟動的
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
這段代碼解決了 Woker是如何創建並加入workers的以及worker是如何啟動的的問題。
addWorker做的核心工作就是，創建worker, 啟動worker, 在創建之前還會做一些校驗。調用了worker裡邊線程的start後就要等待cpu調度執行worker的run方法了。
public void run() {
runWorker(this);
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// task是創建worker帶進去的任務，會先執行他，然後從workQueue裡邊取
// 如果沒有的話跳出去
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 {
// 這個跟我們平時理解的Runnable還不一樣，可以體會下，他這個run就是一個普通的方法
// 他直接調run是要執行任務，線程的start只是把worker裡邊的那個run跑起來了
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 {
// 從while跳出來表明沒有任務可以執行了
processWorkerExit(w, completedAbruptly);
}
}
這個也比較容易，就是不斷的從workQueue取任務，執行，直到沒任務了跳出來。接下來就是worker如何被銷毀的問題了
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
// 移除掉worker（裁員）
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate();
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
// 比核心線程數多的話，執行完的Worker直接移除就好
if (workerCountOf(c) >= min)
return; // replacement not needed
}
// 小於核心線程數就會再加個Worker, 讓他繼續等待接收任務（招人）
addWorker(null, false);
}
}
直接從workers裡邊移除worker, 移除後如果worker數量比核心線程數還少，就再加個worker, 否則不加。
一些體會
看源碼一定不要過分糾結細節，就像這個線程池，我看網上很多文章去算那幾個位運算的十進位數，感覺是在浪費時間，沒有抓住重點。
當然這也不是絕對的（似乎說的矛盾了），一些細節的設計還是非常精妙值得學習的。還是這個位運算，為什麼只用一個int表示線程池狀態和worker的數量呢。
要多多聯想，還是這個位運算，他是不是和讀寫鎖用一個int既表示寫狀態又表示讀狀態十分相似。Worker繼承AQS，是否能讓你想起AQS的種種。
總之，個人覺得第一遍看是一定不能沉溺於細節的，他會讓你迷惘和喪失信心；第二遍、第三遍可以關註一下細節，感受大師級的設計的美妙之處。當然筆者僅僅粗略看了一遍（逃~）
最後
大雄五一假期閱讀了《java併發編程藝術》這本書，整理了一本gitbook筆記（還沒寫完），需要的同學可以掃描文末二維碼關註“大雄和你一起學編程”公眾號，後臺回覆我愛java領取。這本gitbook還沒徹底完成，所以可能還有些小錯誤。未來會大約每兩天推送其中的一篇文章。
如下是這本gitbook的目錄截圖
