一、前言

在開發(fā)過程中，好多場(chǎng)景要用到線程池。每次都是自己根據(jù)業(yè)務(wù)場(chǎng)景來設(shè)置線程池中的各個(gè)參數(shù)。

這兩天又有需求碰到了，索性總結(jié)一下方便以后再遇到可以直接看著用。

雖說根據(jù)業(yè)務(wù)場(chǎng)景來設(shè)置各個(gè)參數(shù)的值，但有些萬變不離其宗，掌握它的原理對(duì)如何用好線程池起了至關(guān)重要的作用。

那我們接下來就來進(jìn)行線程池的分析。

二、ThreadPoolExecutor的重要參數(shù)

我們先來看下ThreadPoolExecutor的帶的那些重要參數(shù)的構(gòu)造器。

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    ...
}

1、corePoolSize: 核心線程數(shù)

這個(gè)應(yīng)該是最重要的參數(shù)了，所以如何合理的設(shè)置它十分重要。

• 核心線程會(huì)一直存活，及時(shí)沒有任務(wù)需要執(zhí)行。
• 當(dāng)線程數(shù)小于核心線程數(shù)時(shí)，即使有線程空閑，線程池也會(huì)優(yōu)先創(chuàng)建新線程處理。
• 設(shè)置allowCoreThreadTimeout=true（默認(rèn)false）時(shí)，核心線程會(huì)超時(shí)關(guān)閉。

如何設(shè)置好的前提我們要很清楚的知道CPU密集型和IO密集型的區(qū)別。

(1)、CPU密集型

CPU密集型也叫計(jì)算密集型，指的是系統(tǒng)的硬盤、內(nèi)存性能相對(duì)CPU要好很多，此時(shí)，系統(tǒng)運(yùn)作大部分的狀況是CPU Loading 100%，CPU要讀/寫I/O(硬盤/內(nèi)存)，I/O在很短的時(shí)間就可以完成，而CPU還有許多運(yùn)算要處理，CPU Loading 很高。

在多重程序系統(tǒng)中，大部分時(shí)間用來做計(jì)算、邏輯判斷等CPU動(dòng)作的程序稱之CPU bound。例如一個(gè)計(jì)算圓周率至小數(shù)點(diǎn)一千位以下的程序，在執(zhí)行的過程當(dāng)中絕大部分時(shí)間用在三角函數(shù)和開根號(hào)的計(jì)算，便是屬于CPU bound的程序。

CPU bound的程序一般而言CPU占用率相當(dāng)高。這可能是因?yàn)槿蝿?wù)本身不太需要訪問I/O設(shè)備，也可能是因?yàn)槌绦蚴嵌嗑€程實(shí)現(xiàn)因此屏蔽掉了等待I/O的時(shí)間。

(2)、IO密集型

IO密集型指的是系統(tǒng)的CPU性能相對(duì)硬盤、內(nèi)存要好很多，此時(shí)，系統(tǒng)運(yùn)作，大部分的狀況是CPU在等I/O (硬盤/內(nèi)存) 的讀/寫操作，此時(shí)CPU Loading并不高。

I/O bound的程序一般在達(dá)到性能極限時(shí)，CPU占用率仍然較低。這可能是因?yàn)槿蝿?wù)本身需要大量I/O操作，而pipeline做得不是很好，沒有充分利用處理器能力。

好了，了解完了以后我們就開搞了。

(3)、先看下機(jī)器的CPU核數(shù)，然后在設(shè)定具體參數(shù)：

自己測(cè)一下自己機(jī)器的核數(shù)

System.out.println(Runtime.getRuntime().availableProcessors());

即CPU核數(shù) = Runtime.getRuntime().availableProcessors()

(4)、分析下線程池處理的程序是CPU密集型還是IO密集型

CPU密集型：corePoolSize = CPU核數(shù) + 1

IO密集型：corePoolSize = CPU核數(shù) * 2

2、maximumPoolSize：最大線程數(shù)

• 當(dāng)線程數(shù)>=corePoolSize，且任務(wù)隊(duì)列已滿時(shí)。線程池會(huì)創(chuàng)建新線程來處理任務(wù)。
• 當(dāng)線程數(shù)=maxPoolSize，且任務(wù)隊(duì)列已滿時(shí)，線程池會(huì)拒絕處理任務(wù)而拋出異常。

3、keepAliveTime：線程空閑時(shí)間

• 當(dāng)線程空閑時(shí)間達(dá)到keepAliveTime時(shí)，線程會(huì)退出，直到線程數(shù)量=corePoolSize。
• 如果allowCoreThreadTimeout=true，則會(huì)直到線程數(shù)量=0。

4、queueCapacity：任務(wù)隊(duì)列容量（阻塞隊(duì)列）

• 當(dāng)核心線程數(shù)達(dá)到最大時(shí)，新任務(wù)會(huì)放在隊(duì)列中排隊(duì)等待執(zhí)行

5、allowCoreThreadTimeout：允許核心線程超時(shí)

6、rejectedExecutionHandler：任務(wù)拒絕處理器

兩種情況會(huì)拒絕處理任務(wù)：

• 當(dāng)線程數(shù)已經(jīng)達(dá)到maxPoolSize，且隊(duì)列已滿，會(huì)拒絕新任務(wù)。
• 當(dāng)線程池被調(diào)用shutdown()后，會(huì)等待線程池里的任務(wù)執(zhí)行完畢再shutdown。如果在調(diào)用shutdown()和線程池真正shutdown之間提交任務(wù)，會(huì)拒絕新任務(wù)。

線程池會(huì)調(diào)用rejectedExecutionHandler來處理這個(gè)任務(wù)。如果沒有設(shè)置默認(rèn)是AbortPolicy，會(huì)拋出異常。

ThreadPoolExecutor 采用了策略的設(shè)計(jì)模式來處理拒絕任務(wù)的幾種場(chǎng)景。

這幾種策略模式都實(shí)現(xiàn)了RejectedExecutionHandler 接口。

• AbortPolicy 丟棄任務(wù)，拋運(yùn)行時(shí)異常。
• CallerRunsPolicy 執(zhí)行任務(wù)。
• DiscardPolicy 忽視，什么都不會(huì)發(fā)生。
• DiscardOldestPolicy 從隊(duì)列中踢出最先進(jìn)入隊(duì)列（最后一個(gè)執(zhí)行）的任務(wù)。

三、如何設(shè)置參數(shù)

默認(rèn)值：

corePoolSize = 1
maxPoolSize = Integer.MAX_VALUE
queueCapacity = Integer.MAX_VALUE
keepAliveTime = 60s
allowCoreThreadTimeout = false
rejectedExecutionHandler = AbortPolicy()

如何來設(shè)置呢？

需要根據(jù)幾個(gè)值來決定

tasks ：每秒的任務(wù)數(shù)，假設(shè)為500~1000

taskcost：每個(gè)任務(wù)花費(fèi)時(shí)間，假設(shè)為0.1s

responsetime：系統(tǒng)允許容忍的最大響應(yīng)時(shí)間，假設(shè)為1s

做幾個(gè)計(jì)算

corePoolSize = 每秒需要多少個(gè)線程處理？

threadcount = tasks/(1/taskcost) = tasks*taskcout = (500 ~ 1000)*0.1 = 50~100 個(gè)線程。

corePoolSize設(shè)置應(yīng)該大于50。

根據(jù)8020原則，如果80%的每秒任務(wù)數(shù)小于800，那么corePoolSize設(shè)置為80即可。

queueCapacity = (coreSizePool/taskcost)*responsetime

計(jì)算可得 queueCapacity = 80/0.1*1 = 800。意思是隊(duì)列里的線程可以等待1s，超過了的需要新開線程來執(zhí)行。

切記不能設(shè)置為Integer.MAX_VALUE，這樣隊(duì)列會(huì)很大，線程數(shù)只會(huì)保持在corePoolSize大小，當(dāng)任務(wù)陡增時(shí)，不能新開線程來執(zhí)行，響應(yīng)時(shí)間會(huì)隨之陡增。

maxPoolSize 最大線程數(shù)在生產(chǎn)環(huán)境上我們往往設(shè)置成corePoolSize一樣，這樣可以減少在處理過程中創(chuàng)建線程的開銷。

rejectedExecutionHandler：根據(jù)具體情況來決定，任務(wù)不重要可丟棄，任務(wù)重要?jiǎng)t要利用一些緩沖機(jī)制來處理。

keepAliveTime和allowCoreThreadTimeout采用默認(rèn)通常能滿足。

以上都是理想值，實(shí)際情況下要根據(jù)機(jī)器性能來決定。如果在未達(dá)到最大線程數(shù)的情況機(jī)器cpu load已經(jīng)滿了，則需要通過升級(jí)硬件和優(yōu)化代碼，降低taskcost來處理。

以下是我自己的的線程池配置：

@Configuration
public class ConcurrentThreadGlobalConfig {
    @Bean
    public ThreadPoolTaskExecutor defaultThreadPool() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //核心線程數(shù)目
        executor.setCorePoolSize(65);
        //指定最大線程數(shù)
        executor.setMaxPoolSize(65);
        //隊(duì)列中最大的數(shù)目
        executor.setQueueCapacity(650);
        //線程名稱前綴
        executor.setThreadNamePrefix("DefaultThreadPool_");
        //rejection-policy：當(dāng)pool已經(jīng)達(dá)到max size的時(shí)候，如何處理新任務(wù)
        //CALLER_RUNS：不在新線程中執(zhí)行任務(wù)，而是由調(diào)用者所在的線程來執(zhí)行
        //對(duì)拒絕task的處理策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //線程空閑后的最大存活時(shí)間
        executor.setKeepAliveSeconds(60);
        //加載
        executor.initialize();
        return executor;
    }
}

四、線程池隊(duì)列的選擇

workQueue - 當(dāng)線程數(shù)目超過核心線程數(shù)時(shí)用于保存任務(wù)的隊(duì)列。主要有3種類型的BlockingQueue可供選擇：無界隊(duì)列，有界隊(duì)列和同步移交。從參數(shù)中可以看到，此隊(duì)列僅保存實(shí)現(xiàn)Runnable接口的任務(wù)。

這里再重復(fù)一下新任務(wù)進(jìn)入時(shí)線程池的執(zhí)行策略：

• 當(dāng)正在運(yùn)行的線程小于corePoolSize，線程池會(huì)創(chuàng)建新的線程。
• 當(dāng)大于corePoolSize而任務(wù)隊(duì)列未滿時(shí)，就會(huì)將整個(gè)任務(wù)塞入隊(duì)列。
• 當(dāng)大于corePoolSize而且任務(wù)隊(duì)列滿時(shí)，并且小于maximumPoolSize時(shí)，就會(huì)創(chuàng)建新額線程執(zhí)行任務(wù)。
• 當(dāng)大于maximumPoolSize時(shí)，會(huì)根據(jù)handler策略處理線程。

1、無界隊(duì)列

隊(duì)列大小無限制，常用的為無界的LinkedBlockingQueue，使用該隊(duì)列作為阻塞隊(duì)列時(shí)要尤其當(dāng)心，當(dāng)任務(wù)耗時(shí)較長(zhǎng)時(shí)可能會(huì)導(dǎo)致大量新任務(wù)在隊(duì)列中堆積最終導(dǎo)致OOM。

閱讀代碼發(fā)現(xiàn)，Executors.newFixedThreadPool 采用就是 LinkedBlockingQueue，而博主踩到的就是這個(gè)坑，當(dāng)QPS很高，發(fā)送數(shù)據(jù)很大，大量的任務(wù)被添加到這個(gè)無界LinkedBlockingQueue 中，導(dǎo)致cpu和內(nèi)存飆升服務(wù)器掛掉。

當(dāng)然這種隊(duì)列，maximumPoolSize 的值也就無效了。

當(dāng)每個(gè)任務(wù)完全獨(dú)立于其他任務(wù)，即任務(wù)執(zhí)行互不影響時(shí)，適合于使用無界隊(duì)列；例如，在 Web 頁服務(wù)器中。

這種排隊(duì)可用于處理瞬態(tài)突發(fā)請(qǐng)求，當(dāng)命令以超過隊(duì)列所能處理的平均數(shù)連續(xù)到達(dá)時(shí)，此策略允許無界線程具有增長(zhǎng)的可能性。

2、有界隊(duì)列

當(dāng)使用有限的 maximumPoolSizes 時(shí)，有界隊(duì)列有助于防止資源耗盡，但是可能較難調(diào)整和控制。

常用的有兩類，一類是遵循FIFO原則的隊(duì)列如ArrayBlockingQueue，另一類是優(yōu)先級(jí)隊(duì)列如PriorityBlockingQueue。

PriorityBlockingQueue中的優(yōu)先級(jí)由任務(wù)的Comparator決定。

使用有界隊(duì)列時(shí)隊(duì)列大小需和線程池大小互相配合，線程池較小有界隊(duì)列較大時(shí)可減少內(nèi)存消耗，降低cpu使用率和上下文切換，但是可能會(huì)限制系統(tǒng)吞吐量。

3、同步移交隊(duì)列

如果不希望任務(wù)在隊(duì)列中等待而是希望將任務(wù)直接移交給工作線程，可使用SynchronousQueue作為等待隊(duì)列。

SynchronousQueue不是一個(gè)真正的隊(duì)列，而是一種線程之間移交的機(jī)制。要將一個(gè)元素放入SynchronousQueue中，必須有另一個(gè)線程正在等待接收這個(gè)元素。

只有在使用無界線程池或者有飽和策略時(shí)才建議使用該隊(duì)列。

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。