# 理論知識 週末上海下起了雨也降溫了，無事開啟電腦看看原始碼，就想到了執行緒池。執行緒池的技術網路上已經有很多文章都已經寫過了，而且理論都是一樣的。 但是理論歸理論，面試的時候也許你剛好看了一篇能應付過去，但是如果深究細節可能就會懵逼。所以我很建議任何理論我們都需要自己去探究一下才好，自己實踐過的才有自己的理解而不是死記硬背，這樣才會經久不忘。 執行緒池屬於開發中常見的一種**池化技術**，這類的池化技術的目的都是為了提高資源的利用率和提高效率，類似的HttpClient連線池，資料庫連線池等。 在沒有執行緒池的時候，我們要建立多執行緒的併發，一般都是通過繼承 Thread 類或實現 Runnable 介面或者實現 Callable 介面，我們知道執行緒資源是很寶貴的，而且執行緒之間切換執行時需要記住上下文資訊，所以過多的建立執行緒去執行任務會造成資源的浪費而且對CPU影響較大。 為了方便， JDK 1.5 之後為我們提供了幾種建立執行緒池的方法： - **Executors.newFixedThreadPool(nThreads)**：建立一個定長執行緒池，可控制執行緒最大併發數，超出的執行緒會在佇列中等待。 - **Executors.newCachedThreadPool()**：建立一個可快取執行緒池，如果執行緒池長度超過處理需要，可靈活回收空閒執行緒，若無可回收，則新建執行緒。 - **Executors.newSingleThreadExecutor()**：建立一個單執行緒化的執行緒池，它只會用唯一的工作執行緒來執行任務， 保證所有任務按照指定順序(FIFO, LIFO, 優先順序)執行。 - **Executors.newScheduledThreadPool(nThreads)**：建立一個定長執行緒池，支援定時及週期性任務執行。 雖然這些都是 JDK 預設提供的，但是還是要說它們的定製性太差了而且有點雞肋，很多時候不能滿足我們的需求。例如通過 newFixedThreadPool 方式建立的固定執行緒池，它內部使用的佇列是 LinkedBlockingQueue，但是它的佇列大小預設是 Integer.MAX_VALUE，這會有什麼問題？ 當核心執行緒滿了的時候，任務會進入佇列中等待，直到佇列滿了為止。但是也許任務還未達到 `Integer.MAX_VALUE` 這個值的時候，記憶體就已經 OOM 了，因為記憶體放不下這麼多的任務，畢竟記憶體大小有限。 所以更多的時候我們都是自定義執行緒池，也就是使用 new ThreadPoolExecutor 的方式，其實你看原始碼你可以發現以上的4個執行緒池技術底層都是通過 ThreadPoolExecutor 來建立的，只不過它們自己為我們填充了這些引數的固定值而已。 ThreadPoolExecutor 的建構函式如下所示： ```java ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler); ``` 我們來看下這幾個核心引數的涵義和作用： - **corePoolSize**： 為執行緒池的核心執行緒基本大小。 - **maximumPoolSize**： 為執行緒池最大執行緒大小。 - **keepAliveTime** 和 **unit** 則是執行緒空閒後的存活時間。 - **workQueue**： 用於存放任務的阻塞佇列。 - **handler**： 當佇列和最大執行緒池都滿了之後的飽和策略。 通過這些引數的配置使得整個執行緒池的工作流程如下：  前幾年一般普通的技術面試瞭解了以上的知識內容也差不多就夠了，但是目前的大環境的影響或者面試更高階的開發上面的知識點是經不起深度考問的。例如以下幾個問題你是否瞭解：執行緒池的內部有哪些狀態？是如何判斷核心執行緒數是否已滿的？最大執行緒數是否包含核心執行緒數？當執行緒池中的執行緒數剛好達到 maximumPoolSize 這個值的時候，這個任務能否正常被執行？......，想要了解這些問題的答案我們只能在執行緒池的原始碼中尋找了。 # 實戰模擬測試 我們自定義一個執行緒池，然後通過 for 迴圈連續建立10個任務並列印執行緒執行資訊，整體程式碼如下所示： ```java public static void main(String[] args) { ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(3, 6, 5L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(4)); for (int i = 0; i < 10; i++) { threadPoolExecutor.execute(() -> { System.out.println("測試執行緒池：" + Thread.currentThread().getName() + "," + threadPoolExecutor.toString()); }); } } ``` 當 corePoolSize = 3，maximumPoolSize = 6，workQueue 大小為4的時候，我們的列印資訊為：  可以發現總的建立了6個執行緒來執行完成了10個任務，其實很好理解，c=3個核心執行緒執行了3個任務，然後4個任務在佇列中等待核心執行緒執行，最後額外建立了e=3個執行緒執行了剩下的3個任務，總建立的執行緒數就是 c + e = 6 <= 6（最大執行緒數）。 如果我們調整物件建立的時候的建構函式引數，例如 ```java ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(3, 5, 5L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(2)); ``` 我們再次執行上述的程式碼，則會報錯，丟擲如下 RejectedExecutionException 異常資訊，可以看到是因為拒絕策略攔截的異常資訊。  還是按照上面的邏輯分析，這時核心執行緒數是 c = 3，而阻塞佇列的大小是 2，因此核心執行緒會處理掉其中5個任務，而剩下的5個任務會額外建立 e=5個執行緒去執行，那麼匯流排程數就是 c + e = 8，但是這時的最大執行緒數 maximumPoolSize = 5，因此超過了最大執行緒數的限制，這時就執行了預設的拒絕策略丟擲異常。其實它在準備建立第6個執行緒的時候就已經報錯了，從這裡也可以得知**只要建立的匯流排程數 >= maximumPoolSize 的時候，執行緒池就不會繼續執行任務了而會去執行拒絕策略的邏輯**。 # 技術來源於生活 人們常常在生活中遇到一些困難的時候會進行頭腦風暴從而產生一些意想不到的解決方案，這些都是思想和智慧的結晶。我們很多技術的解決方案也都來源於生活。 我經常想如果以後不做程式設計師應該做什麼？餐飲似乎是最大眾的了，畢竟民以食為天。 開餐館前期肯定不能做太大，一是本金的問題，還有就是需要市場試水。在市場需求不明確的情況下租個小店面還是靠譜的，就算虧也不會太多。 店面租個幾十平的，就做香辣烤魚，餐桌大概15桌的樣子。然後就是員工了，除了廚師主要是服務員了，但是我不能招15個服務員啊，每桌分配一個太浪費了，需要提高資源利用率控制成本，所以員工不能招太多，我只需要招5個固定服務員負責在大廳招呼顧客和傳菜就可以了，每個人負責3個餐桌。  但是我沒想到我們餐館做的烤魚很合大眾口味，很受歡迎又加上營銷效果好，成了一家網紅餐館。生意更是蒸蒸日上，每天座無虛席。但是空間有限啊，所以我們只能讓後來無座的顧客稍微等候了，於是我們安排了一個取號排隊等候區，顧客等待叫號有序就餐。  這時候餐館的人員不變，仍然是5個服務員負責處理大廳的主要服務工作，同時排隊等候區面積也不能過大，有個範圍限制，不能影響我們的正常人員活動，同時也不能超過餐館的範圍排到餐館外，如果顧客排隊站到門外馬路上了，這是就很危險的。隨著口碑的發酵，一傳十，十傳百，我們的顧客絡繹不絕，同時我們為了提高消費率又做起了外賣的服務，可以打包外帶。 為了避免發生上述這種危險的情況和提高訂單處理率，我們只能額外請一些臨時工了，讓他們來幫忙處理我們的外賣訂單從而提高業務處理能力。 但是也不是請的越多越好，我們有成本控制，因為請的臨時工我們也需要付工資。那怎麼辦呢？最終只能忍痛了啊，對於超出我們處理能力的訂單，我們就採取一定的拒絕策略，例如告知顧客當天的份額已經售罄，請改天再來。 以上就是我們執行緒池執行的一個現實生活中的例子，核心執行緒就是我們的5個固定服務員，而排隊等候區就是我們的等待佇列，佇列不能設為無限大，因為會造成OOM，如果佇列滿了執行緒池會另起額外執行緒去處理任務，也就是上述例子中的臨時工，餐館有經營成本控制所以有員工上限，不能請過多的臨時工，這就是最大執行緒數。如果臨時工達到最大數且佇列也滿了，那麼我們只能通過拒絕策略暫時不接受額外的服務要求了。 # 一起看原始碼 口說無憑，理論都是這樣說的，那實際上原始碼是不是真是這樣寫的呢？我們一起來看下執行緒池的原始碼。通過 `threadPoolExecutor.execute(...)`的入口進入原始碼，刪除了註釋資訊之後的原始碼內容如下，由於封裝的好，所以只有短短几行。 ```java public void execute(Runnable command) { // #1 任務非空校驗 if (command == null) throw new NullPointerException(); // #2 新增核心執行緒執行任務 int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // #3 任務入佇列 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); } // #4 新增普通執行緒執行任務，如果失敗則執行拒絕策略 else if (!addWorker(command, false)) reject(command); } ``` 如果不關注細節只關注整體，從以上原始碼中我們可以發現其中主要分為了四個步驟來處理邏輯。排除第一步的非空校驗程式碼，我們可以看出剩下的三步其實就是我們執行緒池的執行邏輯，也就是上面的執行流程圖的邏輯內容。 - (1) 任務的非空校驗。 - (2) 獲取當前RUNNING的執行緒數，如果小於核心執行緒數，則建立核心執行緒去執行任務，否則走#3。 - (3) 如果當前執行緒池處於RUNNING狀態，那麼就將任務放入佇列中。這時還會再做個雙重校驗，因為可能存在有些執行緒在我們上次檢查後死了，或者從我們進入這個方法後pool被關閉了，所以我們需要再次檢查state。如果執行緒池停止了就需要回滾剛才的新增任務到佇列中的操作並通過拒絕策略拒絕該任務，或者如果池中沒有執行緒了，則新開啟一個執行緒執行任務。 - (4) 如果佇列滿了之後無法在將任務加入佇列，則建立新的執行緒去執行任務，如果也失敗了，那麼就可能是執行緒池關閉了或者執行緒池飽和了，這時執行拒絕策略不再接受任務。 >

雙重校驗中有以下兩個點需要注意： > > **1. 為什麼需要 double check 執行緒池的狀態？** 在多執行緒環境下，執行緒池的狀態時刻在變化，而 ctl.get() 是非原子操作，很有可能剛獲取了執行緒池狀態後執行緒池狀態就改變了。判斷是否將 command 加入 workque 是執行緒池之前的狀態。倘若沒有 double check，萬一執行緒池處於非 running 狀態（在多執行緒環境下很有可能發生），那麼 command 永遠不會執行。 > > **2、為什麼 addWorker(null, false) 的任務為null？** addWorker(null, false)，這個方法執行時只是建立了一個新的執行緒，但是沒有傳入任務，這是因為前面已經將任務新增到佇列中了，這樣可以防止執行緒池處於 running 狀態，但是沒有執行緒去處理這個任務。 而根據以上程式碼的具體步驟我們可以畫出詳細的執行流程，如下圖所示  以上的原始碼其實只有10幾行，看起來很簡單，主要是它的封裝性比較好，其中主要有兩個點需要重點解釋，分別是：**執行緒池的狀態**和 `addWorker()`新增工作的方法，這兩個點弄明白了這段執行緒池的原始碼差不多也就理解了。 ## 執行緒池執行狀態-runState 執行緒有狀態，執行緒池也有它的執行狀態，這些狀態提供了主生命週期控制，伴隨著執行緒池的執行，由內部來維護，從原始碼中我們可以發現執行緒池共有5個狀態：`RUNNING`，`SHUTDOWN`，`STOP`，`TIDYING`，`TERMINATED`。 各狀態值所代表的的含義和該狀態值下可執行的操作，具體資訊如下： | 執行狀態 | 狀態描述 | | :----------: | :----------:| |**RUNNING** | 接收新任務，並且也能處理阻塞佇列中的任務。 | |**SHUTDOWN**|不接收新任務，但是卻可以繼續處理阻塞佇列中的任務。| |**STOP** |不接收新任務，同時也不處理佇列任務，並且中斷正在進行的任務。| |**TIDYING**|所有任務都已終止，workercount(有效執行緒數)為0，執行緒轉向 TIDYING 狀態將會執行 terminated() 鉤子方法。| |**TERMINATED** |terminated() 方法呼叫完成後變成此狀態。| 生命週期狀態流轉如下圖所示：  很多時候我們表示狀態都是通過簡單的 int 值來表示，例如資料庫資料的刪除標誌 delete_flag 其中0表示有效，1表示刪除。而線上程池的原始碼裡我們可以看到它是通過如下方式來進行表示的， ```java 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; // runState is stored in the high-order bits 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; ``` 執行緒池內部使用一個變數維護兩個值：執行狀態（runState）和執行緒數量 （workerCount）何做到的呢？將十進位制 int 值轉換為二進位制的值，共32位，其中高3位代表執行狀態（runState ），而低29位代表工作執行緒數（workerCount）。  關於內部封裝的獲取生命週期狀態、獲取執行緒池執行緒數量的計算方法如以下程式碼所示： ```java //獲取執行緒池狀態 private static int runStateOf(int c) { return c & ~CAPACITY; } //獲取執行緒數量 private static int workerCountOf(int c) { return c & CAPACITY; } // Packing and unpacking ctl private static int ctlOf(int rs, int wc) { return rs | wc; } ``` 通過巧妙的位運算可以分別獲取**高3位的執行狀態值**和**低29位的執行緒數量值**，如果感興趣的可以去看下具體的實現程式碼，這裡就不再贅述了。 ## 新增工作執行緒-addWorker 新增執行緒是通過 addWorker() 方法來實現的，這個方法有兩個入參，`Runnable firstTask` 和 `boolean core`。 ```java private boolean addWorker(Runnable firstTask, boolean core){...} ``` - Runnable firstTask 即是當前新增的執行緒需要執行的首個任務. - boolean core 用來標記當前執行的執行緒是否是核心執行緒還是普通執行緒. 返回前面的執行緒池的 execute() 方法的程式碼中，可以發現這個addWorker() 有三個地方在呼叫，分別在 #2，#3和#4。 - \#2：當工作執行緒數 < 核心執行緒數的時候，通過`addWorker(command, true)`新增核心執行緒執行command任務。 - \#3：double check的時候，如果發現執行緒池處於正常執行狀態但是裡面沒有工作執行緒，則添加個空任務和一個普通執行緒，這樣一個 task 為空的 worker 線上程執行的時候會去阻塞任務佇列裡拿任務，這樣就相當於建立了一個新的執行緒，只是沒有馬上分配任務。 - \#4：佇列已滿的情況下，通過新增普通執行緒（非核心執行緒）去執行當前任務，如果失敗了則執行拒絕策略。 addWorker() 方法呼叫的地方我們看完了，接下來我們一起來看下它裡面究竟做了些什麼，原始碼如下： ```java private boolean addWorker(Runnable firstTask, boolean core) { 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 } } 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; } ``` 這個方法稍微有點長，我們分段來看下，將上面的程式碼我們拆分成兩個部分來看，首先看第一部分： ```java 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; // 嘗試通過CAS方式增加workerCount 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 } } ``` 這部分程式碼有兩層巢狀的 for 死迴圈，在第一行有個`retry:`程式碼，這個也許有些同學沒怎麼見過，這個是相當於是一個位置標記，retry後面跟迴圈，標記這個迴圈的位置。 我們平時寫 for 迴圈的時候，是通過`continue;`或`break;`來跳出當前迴圈，但是如果我們有多重巢狀的 for 迴圈，如果我們想在裡層的某個迴圈體中當達到某個條件的時候直接跳出所有迴圈或跳出到某個指定的位置，則使用`retry:`來標記這個位置就可以了。 程式碼中共有4個位置有改變迴圈體繼續執行下去，分別是兩個`return false;`，一個`break retry;`和一個`continue retry;`。 **首先我們來看下第一個`return false;`**，這個 return 在最外層的一個 for 迴圈， ```java if (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; ``` 這是一個判斷執行緒池狀態和執行緒佇列情況的程式碼，這個邏輯判斷有點繞可以改成 ``` rs >= shutdown && (rs != shutdown || firstTask != null || workQueue.isEmpty()) ``` 這樣就好理解了，邏輯判斷成立可以分為以下幾種情況直接返回 false，表示新增工作執行緒失敗。 - rs > shutdown：執行緒池狀態處於 `STOP`，`TIDYING`，`TERMINATED`時，新增工作執行緒失敗，不接受新任務。 - rs >= shutdown && firstTask != null：執行緒池狀態處於 `SHUTDOWN`，`STOP`，`TIDYING`，`TERMINATED`狀態且worker的首個任務不為空時，新增工作執行緒失敗，不接受新任務。 - rs >= shutdown && workQueue.isEmppty：執行緒池狀態處於 `SHUTDOWN`，`STOP`，`TIDYING`，`TERMINATED`狀態且阻塞佇列為空時，新增工作執行緒失敗，不接受新任務。 這樣看來，最外層的 for 迴圈是不斷的校驗當前的執行緒池狀態是否能接受新任務，如果校驗通過了之後才能繼續往下執行。 **然後接下來看第二個`return false;`**，這個 return 是在內層的第二個 for 迴圈中，是判斷執行緒池中當前的工作執行緒數量的，不滿足條件的話直接返回 false，表示新增工作執行緒失敗。 - 工作執行緒數量是否超過可表示的最大容量（CAPACITY）. - 如果新增核心工作執行緒，是否超過最大核心執行緒容量（corePoolSize）. - 如果新增普通工作執行緒，是否超過執行緒池最大執行緒容量（maximumPoolSize）. **後面的`break retry;`** ，表示如果嘗試通過CAS方式增加工作執行緒數workerCount成功，則跳出這個雙迴圈，往下執行後面第二部分的程式碼，而`continue retry;`是再次校驗下執行緒池狀態是否發生變化，如果發生了變化則重新從最外層 for 開始繼續迴圈執行。 通過第一部分程式碼的解析，我們發現只有`break retry;`的時候才能執行到後面第二部分的程式碼，而後面第二部分程式碼做了些什麼呢？ ```java boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { //建立Worker物件例項 w = new Worker(firstTask); //獲取Worker物件裡的執行緒 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()); //滿足 rs < SHUTDOWN 判斷執行緒池是否是RUNNING，或者 //rs == SHUTDOWN && firstTask == null 執行緒池如果是SHUTDOWN， //且首個任務firstTask為空， if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); //將Worker例項加入執行緒池workers workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; //執行緒新增成功標誌位 -> true workerAdded = true; } } finally { //釋放鎖 mainLock.unlock(); } //如果worker例項加入執行緒池成功，則啟動執行緒，同時修改執行緒啟動成功標誌位 -> true if (workerAdded) { t.start(); workerStarted = true; } } } finally { if (! workerStarted) //新增執行緒失敗 addWorkerFailed(w); } return workerStarted; ``` 這部分程式碼主要的目的其實就是啟動一個執行緒，前面是一堆的條件判斷，看是否能夠啟動一個工作執行緒。它由兩個`try...catch...finally`內容組成，可以將他們拆開來看，這樣就很容易看懂。 我們先看裡面一層的`try...catch...finally`，當Worker例項中的 Thread 執行緒不為空的時候，開啟一個獨佔鎖`ReentrantLock mainLock`，防止其他執行緒也來修改操作。 ```java 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(); } ``` - 首先檢查執行緒池的狀態，當執行緒池處於 RUNNING 狀態或者執行緒池處於 SHUTDOWN 狀態但是當前執行緒的 firstTask 為空，滿足以上條件時才能將 worker 例項新增進執行緒池，即`workers.add(w);`。 - 同時修改 largestPoolSize，largestPoolSize變數用於記錄出現過的最大執行緒數。 - 將標誌位 workerAdded 設定為 true，表示新增工作執行緒成功。 - 無論成功與否，在 finally 中都必須執行 `mainLock.unlock()`來釋放鎖。 外面一層的`try...catch...finally`主要是為了判斷工作執行緒是否啟動成功，如果內層`try...catch...finally`程式碼執行成功，即 worker 新增進執行緒池成功，workerAdded 標誌位置為true，則啟動 worker 中的執行緒 `t.start()`，同時將標誌位 workerStarted 置為 true，表示執行緒啟動成功。 ```java if (workerAdded) { t.start(); workerStarted = true; } ``` 如果失敗了，即 workerStarted == false，則在 finally 裡面必須執行`addWorkerFailed(w)`方法，這個方法相當於是用來回滾操作的，前面增的這裡移除，前面加的這裡減去。 ```java private void addWorkerFailed(Worker w) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { if (w != null) //從執行緒池中移除worker例項 workers.remove(w); //通過CAS，將工作執行緒數量workerCount減1 decrementWorkerCount(); // tryTerminate(); } finally { mainLock.unlock(); } } ``` ## Worker類 上面我們分析了addWorker 方法的原始碼，並且看到了 `Thread t = w.thread`，`workers.add(w)`和`t.start()`等程式碼，知道了執行緒池的執行狀態和新增工作執行緒的流程，那麼我們還有一些疑問： - 這裡的 Worker 是什麼？和 Thread 有什麼區別？ - 執行緒啟動後是如何拿任務？在哪拿任務去執行的？ - 阻塞佇列滿後，額外新建立的執行緒是去佇列裡拿任務的嗎？如果不是那它是去哪拿的？ - 核心執行緒會一直存在於執行緒池中嗎？額外建立的普通執行緒執行完任務後會銷燬嗎？ Worker 是 `ThreadPoolExecutor`的一個內部類，主要是用來維護執行緒執行任務的中斷控制狀態，它實現了Runnable 介面同時繼承了AQS，實現 Runnable 介面意味著 Worker 就是一個執行緒，繼承 AQS 是為了實現獨佔鎖這個功能。 ```java 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; //建構函式，初始化AQS的state值為-1 Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } } ``` 至於為什麼沒有使用可重入鎖 ReentrantLock，而是使用AQS，為的就是實現不可重入的特性去反應執行緒現在的執行狀態。 1. lock方法一旦獲取了獨佔鎖，表示當前執行緒正在執行任務中。 2. 如果正在執行任務，則不應該中斷執行緒。 3. 如果該執行緒現在不是獨佔鎖的狀態，也就是空閒的狀態，說明它沒有在處理任務，這時可以對該執行緒進行中斷。 4. 執行緒池在執行 shutdown 方法或 tryTerminate 方法時會呼叫 interruptIdleWorkers 方法來中斷空閒的執行緒，interruptIdleWorkers 方法會使用 tryLock 方法來判斷執行緒池中的執行緒是否是空閒狀態；如果執行緒是空閒狀態則可以安全回收。 Worker 類有一個構造方法，構造引數為給定的首個任務 firstTask，並持有一個執行緒thread。thread是在呼叫構造方法時通過 ThreadFactory 來建立的執行緒，可以用來執行任務； **firstTask用它來初始化時傳入的第一個任務，這個任務可以有也可以為null。如果這個值是非空的，那麼執行緒就會在啟動初期立即執行這個任務；如果這個值是null，那麼就需要建立一個執行緒去執行任務列表（workQueue）中的任務，也就是非核心執行緒的建立。** # 任務執行-runWorker 上面我們一起看過執行緒的啟動`t.start()`，具體執行是在 Worker 的 run() 方法中 ```java public void run() { runWorker(this); } ``` run() 方法中又呼叫了runWorker() 方法，所有的實現都在這裡 ```java final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { while (task != null || (task = getTask()) != null) { 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 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { 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; w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } } ``` 很多人看到這樣的程式碼就感覺頭痛，其實你細看，這裡面我們可以看關鍵點，裡面有三塊`try...catch...finally`程式碼，我們將這三塊分別單獨拎出來看並且將拋異常的地方暫時刪掉或註釋掉，這樣它看起來就清爽了很多 ```java Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; //由於Worker初始化時AQS中state設定為-1，這裡要先做一次解鎖把state更新為0，允許執行緒中斷 w.unlock(); // allow interrupts boolean completedAbruptly = true; try { // 迴圈的判斷任務（firstTask或從佇列中獲取的task）是否為空 while (task != null || (task = getTask()) != null) { // Worker加鎖，本質是AQS獲取資源並且嘗試CAS更新state由0更變為1 w.lock(); // 如果執行緒池執行狀態是stopping, 確保執行緒是中斷狀態; // 如果不是stopping, 確保執行緒是非中斷狀態. if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); //此處省略了第二個try...catch...finally } // 走到這裡說明某一次getTask()返回為null，執行緒正常退出 completedAbruptly = false; } finally { //處理執行緒退出 processWorkerExit(w, completedAbruptly); } ``` 第二個`try...catch...finally` ```java try { beforeExecute(wt, task); Throwable thrown = null; //此處省略了第三個try...catch...finally } finally { task = null; w.completedTasks++; w.unlock(); } ``` 第三個`try...catch...finally` ```java 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); } ``` 上面的程式碼中可以看到有`beforeExecute`、`afterExecute`和`terminaerd`三個函式，它們都是鉤子函式，可以分別在子類中重寫它們用來擴充套件ThreadPoolExecutor，例如新增日誌、計時、監視或者統計資訊收集的功能。 - beforeExecute()：執行緒執行之前呼叫 - afterExecute()：執行緒執行之後呼叫 - terminaerd()：執行緒池退出時候呼叫 這樣拆分完之後發現，其實主要注意兩個點就行了，分別是`getTask()`和`task.run()`，`task.run()`就是執行任務，那我們繼續來看下`getTask()`是如何獲取任務的。 # 獲取任務-getTask ```java private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? for (;;) { int c = ctl.get(); int rs = runStateOf(c); //1.執行緒池狀態是STOP，TIDYING，TERMINATED //2.執行緒池shutdown並且佇列是空的. //滿足以上兩個條件之一則工作執行緒數wc減去1，然後直接返回null if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); //允許核心工作執行緒物件銷燬淘汰或者工作執行緒數 > 最大核心執行緒數corePoolSize boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; //1.工作執行緒數 > 最大執行緒數maximumPoolSize 或者timed == true && timedOut == true //2.工作執行緒數 > 1 或者佇列為空 //同時滿足以上兩個條件則通過CAS把執行緒數減去1，同時返回null。CAS把執行緒數減去1失敗會進入下一輪迴圈做重試 if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { /// 如果timed為true，通過poll()方法做超時拉取，keepAliveTime時間內沒有等待到有效的任務，則返回null // 如果timed為false，通過take()做阻塞拉取，會阻塞到有下一個有效的任務時候再返回（一般不會是null） Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } } ``` 裡面有個關鍵字`allowCoreThreadTimeOut`，它的預設值為false，在Java1.6開始你可以通過`threadPoolExecutor.allowCoreThreadTimeOut(true)`方式來設定為true，通過字面意思就可以明白這個欄位的作用是什麼了，即是否允許核心執行緒超時銷燬。 **預設的情況下核心執行緒數量會一直保持，即使這些執行緒是空閒的它也是會一直存在的，而當設定為 true 時，執行緒池中 corePoolSize 執行緒空閒時間達到 keepAliveTime 也將銷燬關閉。** # 結尾 通過整片分析下來，執行緒池裡面有很多細節處需要注意，閱讀完原始碼之後也理解了更多，解開了很多困惑，獲取到了更多的知識點，所以原始碼的閱讀是很重