并發編程之：深入決議執行緒池

大家好，我是小黑，一個在互聯網茍且偷生的農民工，

本期帶來執行緒池的第二期內容，如果對執行緒池的基本概念還不是很清楚，可以先看我上一篇文章，

面試官：談談你對執行緒池的理解

本期內容會從以下幾個方面決議執行緒池的具體實作：

• 執行緒池狀態
• 執行緒池初始化
• 如何執行任務
• 鉤子方法
• 等待佇列和排隊策略
• 自定義拒絕策略
• 執行緒池關閉
• 動態調整容量
• 合理配置容量

執行緒池狀態

ThreadPoolExecutor中定義了如下幾種執行緒池狀態：

• RUNNING :運行狀態，該執行緒池可以接受新任務和處理排隊任務
• SHUTDOWN：關閉狀態，不接受新任務，但處理排隊任務
• STOP：停止狀態，不接受新任務、不處理排隊任務和中斷進行中任務
• TIDYING：整理狀態，所有的執行緒任務已終止，workerCount為零，轉換到整理狀態的執行緒將運行terminated()鉤子方法
• TERMINATED：終止狀態，terminated()執行完畢，執行緒池將會被設定為TERNINATED狀態，

我們在上面代碼中看到了對于執行緒池狀態的定義，但是并沒有發現有定義一個int型別的變數表示當前執行緒池的狀態，那是怎么做的呢？

我們看到在最上面有定義一個AtomicInteger ctl這樣一個原子型別的Integer，這個ctl不光可以表示執行緒池的運行狀態，同時能夠表示執行緒池的有效執行緒數workerCount，那么是怎么做到的呢？我們都知道Integer型別的記憶體大小是4個位元組，對應32個bit，ctl將32位的高三位用來表示執行緒池的運行狀態，低29位表示有效執行緒數，

這里可以想一下為什么是用高三位表示runState，不是兩位，也不是4位呢？

因為執行緒池的狀態定義了5種，而二進制要能夠表示5種值最少要用3位，比如1位只能表示0和1，兩位能表示00/01/10/11四種，那3位能表示的值則是2^3也就是8種，所以要標識5種狀態則最少需要3位，

因為這一點，也限制了一個執行緒池理論上能設定的最大執行緒數是2^29-1個，

那如果想要從這一個欄位里取出runState或者workCount的值應該怎么做的？

可以看到是通過位運算來實作的，這里先給大家插播一下位運算的邏輯，

• 按位與&：兩數同為1則為1，其他情況為0，如0101& 0100 的結果是0100
• 按位或|：兩數只要有1個是1，則為1，如 0101|0100的結果是0101
• 按位取反~：0變1,1變0，如0101按位取反則等于1010

要獲取高位的runState則使用ctl的值c和容量（也就是2^29-1）取反做與運算；

取低位的workerCount則不用取反，

通過一個欄位來表示兩個概念，并且使用Atomic可以保證操作的原子性，不得不說Doug Lea，YYDS!!!

執行緒池初始化

通過ThreadPoolExecutor的構造方法我們來看一下執行緒池在創建的時候都做了些什么，

可以發現，在構造方法中只是對7個引數進行賦值，并沒有去做執行緒的創建，所以在默認情況下，執行緒池創建后是沒有執行緒的，需要在任務提交時才會創建執行緒，

如果需要在執行緒池創建之后立即創建執行緒，ThreadPoolExecutor提供了兩個方法可以實作：

如何執行任務

通過ThreadPoolExecutor執行任務時可以通過呼叫execute()方法和submit()方法來完成，

在這之前我們需要先知道ThreadPoolExecutor中一些比較重要的變數，可點開下圖查看，

接下來我們來看execute()是如何執行任務的，

以上execute()方法的執行步驟可以總結為3步：

1. 如果有效執行緒數workerCount小于核心執行緒數，則嘗試增加一個執行緒執行當前任務，如果成功，則會在新執行緒中執行任務，如果失敗，則執行下一步；
2. 如果執行緒池狀態是running，則嘗試加入到等待佇列，如果入隊成功，則需要重新檢查執行緒池狀態是否是running，如果已經不是running則要將任務從佇列中remove并按照拒絕策略處理；如果重新檢查執行緒池狀態是running，則要判斷workCount是不是等于0，如果等于0則需要創建一個新的Worker用于執行剛入隊的任務；
3. 如果在第二步入隊失敗，會再次嘗試增加一個Worker執行該任務，如果這里還是不行，則表示執行緒池確實shutdown或者等待佇列滿了，就執行拒絕策略，

有了這個整體之后，我們再進一步看看addWorker()方法，在這之前需要先了解Worker類的結構，

可以發現Worker類繼承了AQS，并且實作Runnable介面，也就是說Worker物件可以交給一個Thread創建執行緒后執行，從Worker的構造方法里我們也能看出，thread是通過執行緒工廠創建一個執行緒，將this作為引數傳遞的，

然后我們再來看addWorker()方法：

簡單總結一下addWorker()方法分以下4步：

1. 如果執行緒池狀態并且作業佇列為空，則直接回傳false，如果作業執行緒數workerCount小于核心執行緒數或者最大執行緒數（這里取決于傳入的引數），則對workerCount自旋加1；
2. 先加鎖，然后將Worker加入到workerset中，解鎖；
3. 如果worker加入workerset成功，將執行緒啟動；
4. 如果執行緒啟動失敗，則將worke移除，workerCount原子減1

既然Worker類實作了Runnable方法，那對應run()方法中的邏輯就必須要看一下了，

在Worker的run方法中直接呼叫外部類ThreadPoolExecutor的runWorker(Worker)方法，

對runWorker()方法總結為以下幾個步驟：

1. 從傳入worker物件的初始任務開始執行，如果初始任務為空則會呼叫getTask()方法獲取任務，如果回傳空著該Worker執行緒則會退出；如果因為外部任務代碼導致的例外拋出，則也會終止回圈，但是不會將Worker執行緒退出；
2. 在運行任務之前都會獲取鎖防止其他任務執行時發生其他的池中斷，并確保在池沒有停止的情況下保證該執行緒不會設定其他中斷；
3. 每個運行任務都會呼叫beforeExecute()方法，這個方法可能會拋出例外，這種情況下會導致任務不處理，并且執行緒會終止；
4. 如果beforeExecute()正常執行，則會運行任務執行run()，在運行任務出拋出的例外和Error等會收集在thrown變數上，傳給afterExecute()；
5. run()執行完之后會執行afterExecute()如果該方法拋出例外同樣會讓執行緒終止，

那么getTask()是去哪里獲取任務呢？當然是從等待佇列中獲取，

getTask()的執行可以總結如下：

1. 會根據當前執行緒池設定的核心執行緒數，最大執行緒數，超時時間等，從任務佇列獲取任務，可能會超時等待，也可能會阻塞知道任務到達；
2. 如果執行緒池STOP，或者有超過最大執行緒數的作業執行緒，或者執行緒是SHUTDOWN并且佇列為空，或者在超時等待任務時超時這4種情況下會回傳空，

鉤子方法

在執行緒池執行任務的runWorker(Worker)方法中我們發現，會在任務執行前和執行后有兩個方法，

從方法簽名上可以看到這兩個方法都是protected的，當時在ThreadPoolExecutor中都沒有具體實作，所以這兩個方法主要用于在自定義執行緒池時覆寫，可以在任務執行前和執行后做一些事情，比如初始化threadLocals，收集統計資訊，列印日志等，需要注意的是這兩個方法中如果拋出例外都會使執行緒終止，

另外，ThreadPoolExecutor中的terminated()也可以被覆寫，可以用于在執行緒完全終止后執行一些特殊處理，

等待佇列和排隊策略

在上面的內容中很多次的提到了等待佇列，也就是ThreadPoolExecutor中的workQueue，用來存放等待執行的任務，

workQueue的型別定義為BlockingQueue<Runnable>通過可以使用以下的三種型別，

有界佇列

有界佇列顧名思義就是有邊界的佇列，需要指定佇列的大小，主要有ArrayBlockingQueue和PriorityBlockingQueue，PriorityBlockingQueue的優點是等待佇列中的任務可以按照任務的優先級處理，

有界佇列的大小設定需要和執行緒池大小相互配合，執行緒池較小佇列較大時，可以減少記憶體消耗，降低執行緒切換次數和CPU的使用率，但是可能會限制系統的吞吐量，所以要結合實際場景考慮如何設定，

無界佇列

佇列的大小沒有限制，常用的有LinkedBlockingQueue，使用該佇列是要謹慎，當任務比較耗時時，可能會導致大量任務堆積在佇列中導致記憶體溢位，使用Executors.newFixedThreadPool創建的執行緒池就是使用的LinkedBlockingQueue，

同步移交佇列

如果不希望佇列等待，而是直接交給作業執行緒執行，則可以使用同步移交佇列SynchronousQueue，該佇列實際不會存放元素，要放入時必須有另一個現在在等待接收元素才能成功，在這之前會一直阻塞，

自定義拒絕策略

上一期我們有講到過執行緒池的4種拒絕策略，

• AbortPolicy：拒絕處理，拋出例外
• CallerRunsPolicy：由創建該執行緒的執行緒（main）執行
• DiscardPolicy： 丟棄，不拋出例外
• DiscardOldestPolicy：和最早創建的執行緒進行競爭，不拋出例外

當然我們也可以自定義拒絕策略，比如我們在拒絕策略中做一些日志記錄等自定義的需求，

執行緒池關閉

ThreadPoolExecutor中有兩個方法可以讓執行緒池關閉，如下：

• shutdown()：不會立即終止執行緒池，而是要等所有任務快取佇列中的任務都執行完后才終止，但再也不會接受新的任務
• shutdownNow()：立即終止執行緒池，并嘗試打斷正在執行的任務，并且清空任務快取佇列，回傳尚未執行的任務

動態調整容量

ThreadPoolExecutor提供了動態調整執行緒池容量大小的方法：setCorePoolSize()和setMaximumPoolSize()，

• setCorePoolSize：設定核心池大小，如果設定的值小于核心執行緒數，則多余執行緒會在下一次空閑時終止，如果設定的值較大，并且等待佇列中有任務，則會立即創建執行緒執行等待佇列中的任務，
• setMaximumPoolSize：設定執行緒池最大能創建的執行緒數目大小，如果新值小于當前的最大執行緒數，則多余的執行緒會在下次空閑時終止，

配置執行緒池大小

那么最后我們來說一下應該如何來配置執行緒池的大小呢？或許大多數程式員都聽過這樣一種說法：

• CPU 密集型應用，執行緒池大小設定為 CPU核數 + 1
• IO 密集型應用，執行緒池大小設定為 2*CPU核數

到底對不對呢？

我認為是不對的，因為在實際場景中，一臺服務器可能都不止一個應用，而這兩個公式都只和CPU核數相關，所以肯定是不正確的，只有在一臺服務器只部署一個應用時才能勉強說的通；還有一個原因就是在一個應用中可能不僅僅是CPU密集型或者IO密集型，可能二者都有，那又該如何選擇呢？以及一個應用中可能會按照功能劃分多個執行緒池，所以最終結論我覺得這兩種說法不對，

那么我們到底應該如何設定執行緒池的大小呢？有沒有什么可以實踐的方法，這里需要給大家介紹一個理論知識，

利特爾法則 (Little's Law)

一個系統請求數等于請求的到達率與平均每個單獨請求花費的時間之乘積

別看這個名字感覺很高大上，其實概念很簡單，

結合到我們的場景中，我們設定單位時間為1秒鐘來計算，λ=每秒收到的請求數，W=每個任務執行的時間，L=λW=每秒平均在系統中運行的執行緒，

假設我們的應用是單核的，則可以直接將L設定為執行緒池大小，但是真實情況并不是，那么多個CPU對我們這個公式中的哪部分資料會有影響呢？

主要是對于W的值有影響，需要知道一個請求中的執行緒IO時間和執行緒CPU時間，帶入公式后則是：

λ=((IO時間+CPU時間)/CPU時間)*CPU個數

那么需要獲得IO時間和CPU時間，則需要通過在代碼中進行埋點才能準確獲得，比如通過AOP切面編程在請求前后獲取時間得到結果，

當然，僅僅依靠這個公式還是不夠的，還需要通過壓力測驗進行調整和檢驗，才能更準確的配置，

以上就是本期的全部內容，我們下期見，關注我的公眾號【小黑說Java】,更多干貨內容，

標籤：Java

上一篇：再見了，我的散裝研發管理平臺；再見了，4臺ECS！

下一篇：作業系統IO之零拷貝技術