基於ThreadPoolTaskExecutor的使用說明

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具體如下

<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的一行

任務:

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();
        }
    }
}

測試案例:

@Test
public 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大小,網絡請求原因的話,可以設置超時時間;業務邏輯的話,另辟蹊徑。。。

以上為個人經驗,希望能給大傢一個參考,也希望大傢多多支持WalkonNet。

推薦閱讀: