五一假期大雄看了一本《java併發程式設計藝術》，瞭解了執行緒池的基本工作流程，竟然發現執行緒池工作原理和網際網路公司運作模式十分相似。 ## 執行緒池處理流程  ## 原理解析 ### 網際網路公司與執行緒池的關係 這裡用一個比喻來描述一下執行緒池，中間有一些名詞你可能不是太清楚，後邊原始碼解析的部分會講到。 你可以把**執行緒池**看作是一個**研發部門**，研發部門有很多**程式設計師（Worker)**, 他們在一個**大辦公室裡（HashSet workers)**。程式設計師幹不完的**需求**（Runnable/Callable)放在**需求池（workQueue）**裡排隊。每個研發部都配置有**骨幹程式設計師數量（corePoolSize）**和**最大能容納的程式設計師數量（maximumPoolSize）**。具體要做的任務就是**產品的需求**。 new 一個執行緒池相當於**建立**了一個研發部，建立研發部時需要指定骨幹程式設計師數量，最大能容納的程式設計師數量，需求池用哪種（BlockingQueue）,如果忙不過來的需求怎麼給產品回覆（拒絕策略）等等內容。剛開始這個研發部一個程式設計師也沒有。 當產品給這個研發部提一個需求時（當然肯定不會只提一個，他們會不斷的提需求。這裡以提一個需求為例） 首先會看骨幹程式設計師招聘滿了沒。 如果沒滿，會招聘一個**骨幹程式設計師**，招聘進來就讓他不停的工作（很殘酷啊），幹完剛派過來的任務他會**主動在需求池找**下一個需求來做（好員工），如果需求池**沒有需求**了，他就停止工作了，然後研發部會把他**裁掉**，如果裁掉後發現骨幹程式設計師數量**不夠**了，就會**再招聘**一個程式設計師。裁掉後，要是骨幹程式設計師數量還夠就不招聘了。 如果骨幹程式設計師數量滿了，就看需求池滿沒滿，如果**需求池沒滿**，就把需求扔進需求池裡；如果**需求池滿了**，就看程式設計師數量**有沒有達到上限**，如果達到了，就對產品說，這個需求我們**做不了**，沒資源；如果沒達到，就招聘一個程式設計師，招聘進來就讓他不停的工作，幹完剛派過來的需求他會主動到需求池找下一個任務來做，如果需求池沒有任務了，他就停止工作了，然後研發部會把他裁掉，如果裁掉後發現骨幹程式設計師數量不夠了，就會再招聘一個程式設計師。裁掉後，要是骨幹程式設計師數量還夠就不招聘了。 ### 原始碼解析 首先是worker（程式設計師） Worker被裝在一個HashSet（workers）裡邊， 他是用來執行任務的，他們的職責就是不斷的從workQueue裡邊取任務，然後執行。當workQueue（需求池）裡邊拿不到任務，或者執行緒池達到特定狀態，worker就會從workers裡邊移走（被裁）。 下邊是Worker原始碼，移除了非關鍵的東西 ```java 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 ```java 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是如何執行的 生活還要繼續 ```java 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方法了。 ```java 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如何被銷燬的問題了 ```java 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的目錄截圖  ![歡迎關注“大雄和你一起學程式設計”公眾號](https://img2020.cnblogs.com/blog/1128201/202005/1128201-20200508232133337-14489848