ThreadPoolTaskExecutor的使用
當我們需要實現并發、異步等操作時，通常都會使用到ThreadPoolTaskExecutor，現對其使用稍作總結。
springboot 配置
提交任務
- 無返回值的任務使用execute(Runnable)
- 有返回值的任務使用submit(Runnable)
處理流程
當一個任務被提交到線程池時，首先查看線程池的核心線程是否都在執行任務，否就選擇一條線程執行任務，是就執行第二步。
查看核心線程池是否已滿，不滿就創建一條線程執行任務，否則執行第三步。
查看任務隊列是否已滿，不滿就將任務存儲在任務隊列中，否則執行第四步。
查看線程池是否已滿，不滿就創建一條線程執行任務，否則就按照策略處理無法執行的任務。
在ThreadPoolExecutor中表現為：
如果當前運行的線程數小于corePoolSize，那么就創建線程來執行任務（執行時需要獲取全局鎖）。
如果運行的線程大于或等于corePoolSize，那么就把task加入BlockQueue。
如果創建的線程數量大于BlockQueue的最大容量，那么創建新線程來執行該任務。
如果創建線程導致當前運行的線程數超過maximumPoolSize，就根據飽和策略來拒絕該任務。
關閉線程池
調用shutdown或者shutdownNow，兩者都不會接受新的任務，而且通過調用要停止線程的interrupt方法來中斷線程，有可能線程永遠不會被中斷，不同之處在于shutdownNow會首先將線程池的狀態設置為STOP，然后嘗試停止所有線程（有可能導致部分任務沒有執行完）然后返回未執行任務的列表。而shutdown則只是將線程池的狀態設置為shutdown，然后中斷所有沒有執行任務的線程，并將剩余的任務執行完。
配置線程個數
如果是CPU密集型任務，那么線程池的線程個數應該盡量少一些，一般為CPU的個數+1條線程。
如果是IO密集型任務，那么線程池的線程可以放的很大，如2*CPU的個數。
對于混合型任務，如果可以拆分的話，通過拆分成CPU密集型和IO密集型兩種來提高執行效率；如果不能拆分的的話就可以根據實際情況來調整線程池中線程的個數。
監控線程池狀態
常用狀態
taskCount：線程需要執行的任務個數。
completedTaskCount：線程池在運行過程中已完成的任務數。
largestPoolSize：線程池曾經創建過的最大線程數量。
getPoolSize：獲取當前線程池的線程數量。
getActiveCount：獲取活動的線程的數量
通過繼承線程池，重寫beforeExecute，afterExecute和terminated方法來在線程執行任務前，線程執行任務結束，和線程終結前獲取線程的運行情況，根據具體情況調整線程池的線程數量。
ThreadPoolTaskExecutor配置問題
最近線上出現一個奇葩問題，使用的是ThreadPoolTaskExecutor來處理后續服務調用，剛開始運行ThreadPoolTaskExecutor處理后續服務調用是沒有問題的，但是一段時間之后，發現后續服務一直沒有被調用，導致了極其嚴重的后果
有關spring中ThreadPoolTaskExecutor具體如下
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
<bean id="threadPoolTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"> <!-- 核心線程數，默認為1 --> <property name="corePoolSize" value="5" /> <!-- 最大線程數，默認為Integer.MAX_VALUE --> <property name="maxPoolSize" value="16" /> <!-- 隊列最大長度，一般需要設置值>=notifyScheduledMainExecutor.maxNum；默認為Integer.MAX_VALUE --> <!--<property name="queueCapacity" value="10" />--> <!-- 線程池維護線程所允許的空閑時間，默認為60s --> <property name="keepAliveSeconds" value="300" /> <!-- 線程池對拒絕任務（無線程可用）的處理策略， 目前只支持AbortPolicy、CallerRunsPolicy；默認為后者 --> <property name="rejectedExecutionHandler"> <!-- AbortPolicy:直接拋出java.util.concurrent.RejectedExecutionException異常 --> <!-- CallerRunsPolicy: 主線程直接執行該任務，執行完之后嘗試添加下一個任務到線程池中， --> <!-- DiscardOldestPolicy: 拋棄舊的任務、暫不支持；會導致被丟棄的任務無法再次被執行 --> <!-- DiscardPolicy: 拋棄當前任務、暫不支持；會導致被丟棄的任務無法再次被執行 --> <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" /> </property></bean> |
那就不得不了解一下java.util.concurrent包下Executor構架了
回憶一下線程池工作原理
如果當前運行的線程少于corePoolSize，則創建新線程來執行任務（需要獲得全局鎖）
如果運行的線程等于或多于corePoolSize ,則將任務加入BlockingQueue
如果無法將任務加入BlockingQueue(隊列已滿)，則創建新的線程來處理任務（需要獲得全局鎖）
如果創建新線程將使當前運行的線程超出maxiumPoolSize，任務將被拒絕，并調用
RejectedExecutionHandler.rejectedExecution()方法
測試場景1
首先，注釋queueCapacity的一行
任務：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
public class CustomRunnable implements Runnable { private int id; public CustomRunnable(int id) { this.id = id; } @Override public void run() { try { System.out.println("begin execute "+ Thread.currentThread().getName() + "-- task id: "+ id); String rs = ClientUtil.get("http://www.****.com"); System.out.println("end execute task: "+ id); } catch (Exception e) { e.printStackTrace(); } }} |
測試案例：
|
1
2
3
4
5
6
7
8
|
@Testpublic void threadTest() throws InterruptedException { for (int i=0; i< 35; i++){ Thread t= new Thread(new CustomRunnable(i)); executor.execute(t); } Thread.sleep(1800000);} |
測試結果：
七月 09, 2018 5:46:47 下午 org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor initialize
信息: Initializing ExecutorService 'threadPoolTaskExecutor'
begin execute threadPoolTaskExecutor-1-- task id: 0
begin execute threadPoolTaskExecutor-2-- task id: 1
begin execute threadPoolTaskExecutor-3-- task id: 2
begin execute threadPoolTaskExecutor-4-- task id: 3
begin execute threadPoolTaskExecutor-5-- task id: 4
end execute task: 4
begin execute threadPoolTaskExecutor-5-- task id: 5
end execute task: 1
begin execute threadPoolTaskExecutor-2-- task id: 6
end execute task: 0
begin execute threadPoolTaskExecutor-1-- task id: 7
end execute task: 2
begin execute threadPoolTaskExecutor-3-- task id: 8
end execute task: 3
begin execute threadPoolTaskExecutor-4-- task id: 9
...
可以發現，一開始線程池就創建了corePoolSize大小的線程，對于之后的新加進的任務，就放到BlockingQueue中，默認是使用LinkedBlockingQueue,大小是Integer.MAX_VALUE，因為隊列大小太大，所以就不會創建maxPoolSize大小的線程數量，因此，只有線程處理完當前任務，才會去處理下一個任務，所以，剛加進去的任務得不到立即處理
測試場景2
只需要打開queueCapacity的一行，其他不變
測試結果：
七月 09, 2018 6:07:13 下午 org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor initialize
信息: Initializing ExecutorService 'threadPoolTaskExecutor'
begin execute threadPoolTaskExecutor-1-- task id: 0
begin execute threadPoolTaskExecutor-2-- task id: 1
begin execute threadPoolTaskExecutor-3-- task id: 2
begin execute threadPoolTaskExecutor-4-- task id: 3
begin execute threadPoolTaskExecutor-5-- task id: 4
begin execute threadPoolTaskExecutor-6-- task id: 15
begin execute threadPoolTaskExecutor-7-- task id: 16
begin execute threadPoolTaskExecutor-8-- task id: 17
begin execute threadPoolTaskExecutor-9-- task id: 18
begin execute threadPoolTaskExecutor-10-- task id: 19
begin execute threadPoolTaskExecutor-11-- task id: 20
begin execute threadPoolTaskExecutor-12-- task id: 21
begin execute threadPoolTaskExecutor-14-- task id: 23
begin execute threadPoolTaskExecutor-15-- task id: 24
begin execute main-- task id: 26
begin execute threadPoolTaskExecutor-13-- task id: 22
begin execute threadPoolTaskExecutor-16-- task id: 25
begin execute threadPoolTaskExecutor-11-- task id: 5
end execute task: 15
begin execute threadPoolTaskExecutor-6-- task id: 6
end execute task: 23
begin execute threadPoolTaskExecutor-14-- task id: 7
end execute task: 4
begin execute threadPoolTaskExecutor-5-- task id: 8
end execute task: 17
begin execute threadPoolTaskExecutor-8-- task id: 9
....
可以發現，因為初始任務數量大于corePoolSize大小，所以線程池初始化就創建了maxPoolSize大小數量的純種，對于后續新加進的任務會入到BlockingQueue隊列中去，之后等待線程處理完一個任務之后再處理隊列中的任務
猜想
線上出現這種原因可能就是因為queueCapacity被設置成了默認(Integer.MAX_VALUE)，而且初始化純種的corePoolSize數量過少，并且線程處理速度較慢(業務邏輯，網絡請求等等原因),導致后續任務會一直填加到隊列中去，遲遲得不到立即處理。
解決方案
手動設置queueCapacity大小，網絡請求原因的話，可以設置超時時間;業務邏輯的話，另辟蹊徑。。。
以上為個人經驗，希望能給大家一個參考，也希望大家多多支持服務器之家。
原文鏈接：https://www.cnblogs.com/lcxdevelop/p/10487857.html
