動態(tài)調(diào)整線程池參數(shù)實踐
點擊上方老周聊架構(gòu)關(guān)注我
一、線程池遇到的挑戰(zhàn)
我們上一篇 《一文讀懂線程池的實現(xiàn)原理 》已經(jīng)從線程池如何維護自身狀態(tài)、線程池如何管理任務(wù)、線程池如何管理線程三個維度來深入剖析線程池的底層原理與源碼剖析，這讓我們對線程池的原理有了較為深入的理解。這對我們多線程編程有很大的幫助，但在使用線程池時還是會面臨幾個棘手的問題。
開發(fā)人員個人經(jīng)驗與水平參差不齊，配置線程池參數(shù)都是按照自己想法來，沒有統(tǒng)一的一個配置標(biāo)準(zhǔn)。
線程池執(zhí)行情況與任務(wù)類型相關(guān)性較大，IO 密集型和 CPU 密集型的任務(wù)運行起來的情況差異非常大。
當(dāng)你配置好線程池后，有的時間段流量高峰期，導(dǎo)致線程池忙不過來；有的時間段流量低峰期，線程池比較空閑。這就會導(dǎo)致資源調(diào)度失衡，降低了系統(tǒng)的穩(wěn)定性。
我們先來看下美團調(diào)研的業(yè)界一些線程池參數(shù)配置方案：
第一種方案是出自《Java并發(fā)編程實踐》，顯然和業(yè)務(wù)場景有所偏離。
第二種方案也不太合理，為什么呢？我們一個項目里一般來說不止一個自定義線程池吧？比如有專門處理數(shù)據(jù)異步持久化的線程池，有專門處理查詢請求的線程池，這樣去做一個簡單的線程隔離。但是如果都用這樣的參數(shù)配置的話，顯然是不合理的。
第三種方案雖然考慮到了業(yè)務(wù)場景，但這是理想狀態(tài)。流量是不可能這么均衡的，就拿美團來說，下午 3、4 點的流量，能和 12 點左右午飯時的流量比嗎？
基于上面線程池的幾個痛點，那有沒有好的解決方案呢？有的，那就是動態(tài)調(diào)整線程池參數(shù)。
盡管業(yè)界沒有一些成熟的經(jīng)驗配置策略，那么我們是不是可以從修改線程池參數(shù)的成本入手？畢竟每次線上線程池故障的話，都得修改代碼里的線程池相應(yīng)的參數(shù)，然后再部署上線，這個過程在對可用性要求極高的項目中那是極其慢的，可能給公司造成巨大的損失。
既然改代碼里的線程池相應(yīng)的參數(shù)并上線這個過程慢，那我們是不是可以把相應(yīng)的線程池參數(shù)配置到分布式配置中心上去？實現(xiàn)線程池參數(shù)可動態(tài)配置和即時生效，線程池參數(shù)動態(tài)化前后的參數(shù)修改流程對比如下：
二、動態(tài)化線程池
2.1 整體設(shè)計
動態(tài)化線程池的核心設(shè)計包括以下三個方面：
2.1.1 簡化線程池配置
線程池構(gòu)造參數(shù)有 7 個，但是最核心的是 3 個：corePoolSize、maximumPoolSize、workQueue，它們最大程度地決定了線程池的任務(wù)分配和線程分配策略。考慮到在實際應(yīng)用中我們獲取并發(fā)性的場景主要是兩種：
并行執(zhí)行子任務(wù)，提高響應(yīng)速度。這種情況下，應(yīng)該使用同步隊列，沒有什么任務(wù)應(yīng)該被緩存下來，而是應(yīng)該立即執(zhí)行。
并行執(zhí)行大批次任務(wù)，提升吞吐量。這種情況下，應(yīng)該使用有界隊列，使用隊列去緩沖大批量的任務(wù)，隊列容量必須聲明，防止任務(wù)無限制堆積。
所以線程池只需要提供這三個關(guān)鍵參數(shù)的配置，并且提供兩種隊列的選擇，就可以滿足絕大多數(shù)的業(yè)務(wù)需求。
2.1.2 參數(shù)可動態(tài)修改
為了解決參數(shù)不好配，修改參數(shù)成本高等問題。在 Java 線程池留有高擴展性的基礎(chǔ)上，封裝線程池，允許線程池監(jiān)聽同步外部的消息，根據(jù)消息進行修改配置。將線程池的配置放置在平臺側(cè)，允許開發(fā)同學(xué)簡單的查看、修改線程池配置。
2.1.3 增加線程池監(jiān)控
對某事物缺乏狀態(tài)的觀測，就對其改進無從下手。在線程池執(zhí)行任務(wù)的生命周期添加監(jiān)控能力，幫助開發(fā)同學(xué)了解線程池狀態(tài)。
2.2 功能架構(gòu)
動態(tài)化線程池提供如下功能：
動態(tài)調(diào)參：支持線程池參數(shù)動態(tài)調(diào)整、界面化操作；包括修改線程池核心大小、最大核心大小、隊列長度等；參數(shù)修改后及時生效。
任務(wù)監(jiān)控：支持應(yīng)用粒度、線程池粒度、任務(wù)粒度的 Transaction 監(jiān)控；可以看到線程池的任務(wù)執(zhí)行情況、最大任務(wù)執(zhí)行時間、平均任務(wù)執(zhí)行時間、P95/P99線等。
負載告警：支持告警規(guī)則配置，當(dāng)超過閾值時（線程池隊列任務(wù)積壓到一定值、線程池負載數(shù)達到一定閾值）會通知相關(guān)的開發(fā)負責(zé)人。
操作監(jiān)控：創(chuàng)建、修改和刪除線程池都會通知到應(yīng)用的開發(fā)負責(zé)人。
操作日志：可以查看線程池參數(shù)的修改記錄，誰在什么時候修改了線程池參數(shù)、修改前的參數(shù)值是什么。
權(quán)限校驗：只有應(yīng)用開發(fā)負責(zé)人才能夠修改應(yīng)用的線程池參數(shù)。
三、Apollo 分布式配置中心
了解完動態(tài)化線程池的整體設(shè)計與功能架構(gòu)后，我相信你也可以設(shè)計出一款動態(tài)線程池組件出來的。下面跟著老周來實踐一下動態(tài)調(diào)整線程池參數(shù)，可能不像上面設(shè)計的那樣那么全面，但會把動態(tài)調(diào)整線程池參數(shù)的核心給實現(xiàn)一下。
不難發(fā)現(xiàn)動態(tài)化線程池的核心是配置管理，那我們就得找一個分布式配置中心，這里老周用的 Apollo，還有其它的像 Spring Cloud Config、disconf、某些大型互聯(lián)網(wǎng)公司自研的分布式配置中心等，根據(jù)自己的項目情況以及使用場景來選擇就行。
3.1 Apollo 總體設(shè)計
Apollo（阿波羅）是攜程框架部門研發(fā)的分布式配置中心，能夠集中化管理應(yīng)用不同環(huán)境、不同集群的配置，配置修改后能夠?qū)崟r推送到應(yīng)用端，并且具備規(guī)范的權(quán)限、流程治理等特性，適用于微服務(wù)配置管理場景。
3.1.1 基礎(chǔ)模型
如下圖即是 Apollo 的基礎(chǔ)模型：
用戶在配置中心對配置進行修改并發(fā)布
配置中心通知 Apollo 客戶端有配置更新
Apollo 客戶端從配置中心拉取最新的配置、更新本地配置并通知到應(yīng)用
3.1.2 架構(gòu)模塊
上圖簡要描述了 Apollo 的總體設(shè)計，我們可以從下往上看：
Config Service 提供配置的讀取、推送等功能，服務(wù)對象是 Apollo 客戶端
Admin Service 提供配置的修改、發(fā)布等功能，服務(wù)對象是 Apollo Portal（管理界面）
Config Service 和 Admin Service 都是多實例、無狀態(tài)部署，所以需要將自己注冊到 Eureka 中并保持心跳
在 Eureka 之上我們架了一層 Meta Server 用于封裝 Eureka 的服務(wù)發(fā)現(xiàn)接口
Client 通過域名訪問 Meta Server 獲取 Config Service 服務(wù)列表（IP+Port），而后直接通過 IP+Port 訪問服務(wù)，同時在 Client 側(cè)會做 load balance、錯誤重試
Portal 通過域名訪問 Meta Server 獲取 Admin Service 服務(wù)列表（IP+Port），而后直接通過 IP+Port 訪問服務(wù)，同時在 Portal 側(cè)會做 load balance、錯誤重試
為了簡化部署，我們實際上會把 Config Service、Eureka 和 Meta Server 三個邏輯角色部署在同一個 JVM 進程中
3.2 服務(wù)端設(shè)計
3.2.1 配置發(fā)布后的實時推送設(shè)計
在配置中心中，一個重要的功能就是配置發(fā)布后實時推送到客戶端。本文重點分析配置更新推送方式，下面我們簡要看一下這塊是怎么設(shè)計實現(xiàn)的。
上圖簡要描述了配置發(fā)布的大致過程：
用戶在 Portal 操作配置發(fā)布
Portal 調(diào)用 Admin Service 的接口操作發(fā)布
Admin Service 發(fā)布配置后，發(fā)送 ReleaseMessage 給各個 Config Service
Config Service 收到 ReleaseMessage 后，通知對應(yīng)的客戶端
上圖的發(fā)送 ReleaseMessage 的實現(xiàn)方式詳情請往下繼續(xù)看：
Admin Service 在配置發(fā)布后，需要通知所有的 Config Service 有配置發(fā)布，從而 Config Service 可以通知對應(yīng)的客戶端來拉取最新的配置。
從概念上來看，這是一個典型的消息使用場景，Admin Service 作為 producer 發(fā)出消息，各個 Config Service 作為 consumer 消費消息。通過一個消息組件（Message Queue）就能很好的實現(xiàn) Admin Service 和 Config Service 的解耦。
在實現(xiàn)上，考慮到 Apollo 的實際使用場景，以及為了盡可能減少外部依賴，我們沒有采用外部的消息中間件，而是通過數(shù)據(jù)庫實現(xiàn)了一個簡單的消息隊列。
3.3 客戶端設(shè)計
上圖簡要描述了 Apollo 客戶端的實現(xiàn)原理：
客戶端和服務(wù)端保持了一個長連接，從而能第一時間獲得配置更新的推送。（通過 Http Long Polling 實現(xiàn)）
客戶端還會定時從 Apollo 配置中心服務(wù)端拉取應(yīng)用的最新配置。
客戶端從 Apollo 配置中心服務(wù)端獲取到應(yīng)用的最新配置后，會保存在內(nèi)存中。
客戶端會把從服務(wù)端獲取到的配置在本地文件系統(tǒng)緩存一份。
應(yīng)用程序可以從 Apollo 客戶端獲取最新的配置、訂閱配置更新通知。
我們從基礎(chǔ)模型、服務(wù)端設(shè)計、客戶端設(shè)計三個維度來分析了 Apollo 總體設(shè)計，相信你對 Apollo 分布式配置中心有了全面且清晰的理解了。為了照顧沒有用過 Apollo 這款分布式配置中心的同學(xué)，老周這里還是簡單給個 Apollo 開發(fā)樣例演示，希望對你后面的動態(tài)調(diào)整線程池參數(shù)實踐有所幫助。
四、動態(tài)調(diào)整線程池參數(shù)實踐
我們了解原理以及架構(gòu)后，那我們開始實踐了。
4.1 服務(wù)端安裝
請看官方文檔進行相應(yīng)的安裝：https://ctripcorp.github.io/apollo/#/zh/deployment/quick-start
執(zhí)行啟動腳本后，當(dāng)看到如下輸出后，就說明啟動成功了！
啟動成功后訪問地址：http://localhost:8070
默認(rèn)輸入用戶名：apollo、密碼：admin，進行登錄。
點擊 SampleApp，我們看到在 DEV 環(huán)境包含一個 timeout 配置項，100 是這個配置項的值，下面我們在應(yīng)用程序讀取這個配置項：
4.2 應(yīng)用程序
4.2.1 引入依賴
<dependencies>
<dependency>
<groupId>com.ctrip.framework.apollo</groupId>
<artifactId>apollo-client</artifactId>
<version>1.7.0</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<version>1.7.25</version>
<scope>compile</scope>
</dependency>
</dependencies>
4.2.2 樣例測試
/**
* 動態(tài)獲取Apollo配置
* 注意要配置：-Dapp.id=myApp -Denv=DEV -Dapollo.cluster=default -Ddev_meta=http://localhost:8080
*
* @author 微信公眾號【老周聊架構(gòu)】
*/
public class GetApolloConfigTest {
public static void main(String[] args) throws InterruptedException {
Config config = ConfigService.getAppConfig();
config.addChangeListener(new ConfigChangeListener() {
@Override
public void onChange(ConfigChangeEvent changeEvent) {
System.out.println("Changes for namespace " + changeEvent.getNamespace());
for (String key : changeEvent.changedKeys()) {
ConfigChange change = changeEvent.getChange(key);
System.out.println(String.format("Found change - key: %s, oldValue: %s, newValue: %s, changeType: %s", change.getPropertyName(), change.getOldValue(), change.getNewValue(), change.getChangeType()));
}
}
});
Thread.sleep(1000000L);
}
}
我們現(xiàn)在把配置項默認(rèn)的值 100 改為 200 程序輸出結(jié)果如下：
控制臺會出現(xiàn)以下日志，表明動態(tài)獲取 Apollo 配置成功了。
4.3 動態(tài)線程池
上面我們把 Apollo 的動態(tài)監(jiān)聽修改配置的功能整明白了以后，再把線程池和 Apollo 結(jié)合起來構(gòu)建動態(tài)線程池那就方便多了。首先我們用默認(rèn)值構(gòu)建一個線程池，然后線程池會監(jiān)聽 Apollo 關(guān)于相關(guān)配置項，如果相關(guān)配置有變化則刷新相關(guān)參數(shù)。
代碼演示：
/**
* 動態(tài)線程池工廠
*
* @author 微信公眾號【老周聊架構(gòu)】
*/
@Slf4j
@Component
public class DynamicThreadPoolFactory {
/** 這里是你的namespace,我這里是默認(rèn)的application **/
private static final String NAME_SPACE = "application";
/** 線程執(zhí)行器 **/
private volatile ThreadPoolExecutor executor;
/** 核心線程數(shù) **/
private Integer corePoolSize = 10;
/** 最大值線程數(shù) **/
private Integer maximumPoolSize = 20;
/** 待執(zhí)行任務(wù)的隊列的長度 **/
private Integer workQueueSize = 1000;
/** 線程空閑時間 **/
private Long keepAliveTime = 1000L;
/** 線程名 **/
private String threadName;
public DynamicThreadPoolFactory() {
Config config = ConfigService.getConfig(NAME_SPACE);
init(config);
listen(config);
}
/**
* 初始化
*/
private void init(Config config) {
if (executor == null) {
synchronized (DynamicThreadPoolFactory.class) {
if (executor == null) {
String corePoolSizeProperty = config.getProperty(ParamsEnum.CORE_POOL_SIZE.getParam(), corePoolSize.toString());
String maximumPoolSizeProperty = config.getProperty(ParamsEnum.MAXIMUM_POOL_SIZE.getParam(), maximumPoolSize.toString());
String keepAliveTImeProperty = config.getProperty(ParamsEnum.KEEP_ALIVE_TIME.getParam(), keepAliveTime.toString());
BlockingQueue<Runnable> workQueueProperty = new LinkedBlockingQueue<>(workQueueSize);
executor = new ThreadPoolExecutor(Integer.valueOf(corePoolSizeProperty), Integer.valueOf(maximumPoolSizeProperty),
Long.valueOf(keepAliveTImeProperty), TimeUnit.MILLISECONDS, workQueueProperty);
}
}
}
}
/**
* 監(jiān)聽器
*/
private void listen(Config config) {
config.addChangeListener(new ConfigChangeListener() {
@Override
public void onChange(ConfigChangeEvent changeEvent) {
log.info("命名空間發(fā)生變化={}", changeEvent.getNamespace());
for (String key : changeEvent.changedKeys()) {
ConfigChange change = changeEvent.getChange(key);
String newValue = change.getNewValue();
refreshThreadPool(key, newValue);
log.info("發(fā)生變化key={},oldValue={},newValue={},changeType={}", change.getPropertyName(), change.getOldValue(), change.getNewValue(), change.getChangeType());
}
}
});
}
/**
* 刷新線程池
*/
private void refreshThreadPool(String key, String newValue) {
if (executor == null) {
return;
}
if (ParamsEnum.CORE_POOL_SIZE.getParam().equals(key)) {
executor.setCorePoolSize(Integer.valueOf(newValue));
log.info("修改核心線程數(shù)key={},value={}", key, newValue);
}
if (ParamsEnum.MAXIMUM_POOL_SIZE.getParam().equals(key)) {
executor.setMaximumPoolSize(Integer.valueOf(newValue));
log.info("修改最大線程數(shù)key={},value={}", key, newValue);
}
if (ParamsEnum.KEEP_ALIVE_TIME.getParam().equals(key)) {
executor.setKeepAliveTime(Integer.valueOf(newValue), TimeUnit.MILLISECONDS);
log.info("修改線程空閑時間key={},value={}", key, newValue);
}
}
public ThreadPoolExecutor getExecutor(String threadName) {
return executor;
}
}
@AllArgsConstructor
public enum ParamsEnum {
CORE_POOL_SIZE("corePoolSize", "核心線程數(shù)"),
MAXIMUM_POOL_SIZE("maximumPoolSize", "最大線程數(shù)"),
KEEP_ALIVE_TIME("keepAliveTime", "線程空閑時間"),
;
@Getter
private String param;
@Getter
private String desc;
}
/**
* 動態(tài)線程池執(zhí)行器
*
* @author 微信公眾號【老周聊架構(gòu)】
*/
@Component
public class DynamicThreadExecutor {
@Resource
private DynamicThreadPoolFactory threadPoolFactory;
public void execute(String bizName, Runnable job) {
threadPoolFactory.getExecutor(bizName).execute(job);
}
public Future<?> sumbit(String bizName, Runnable job) {
return threadPoolFactory.getExecutor(bizName).submit(job);
}
}
@Slf4j
public class DynamicThreadPoolExecutorTest {
@Resource
private DynamicThreadExecutor dynamicThreadExecutor;
/**
* 記得 IDEA VM options 要記得加下面的參數(shù)
* -Dapp.id=SampleApp -Denv=DEV -Dapollo.meta=http://localhost:8080
*/
@Test
public void testExecute() throws InterruptedException {
while (true) {
dynamicThreadExecutor.execute("bizName", new Runnable() {
@Override
public void run() {
System.out.println("bizInfo");
}
});
TimeUnit.SECONDS.sleep(1);
}
}
}
這里可以通過 JDK 自帶的 JVisualVM 工具可以查看到相應(yīng)的線程使用情況。
我們在配置中心修改配置項把核心線程數(shù)設(shè)置為 50，最大線程數(shù)設(shè)置為 100：
你會觀察到線程數(shù)顯著上升
這里還可以在代碼中通過打印相應(yīng)的線程狀態(tài)，更加直觀的從日志上觀察到核心線程、最大線程數(shù)的修改情況。
private static void threadPoolStatus(ThreadPoolExecutor executor, String name) {
BlockingQueue<Runnable> queue = executor.getQueue();
System.out.println(Thread.currentThread().getName() + "-" + name + "-:" +
"核心線程數(shù)：" + executor.getCorePoolSize() +
" 活動線程數(shù)：" + executor.getActiveCount() +
" 最大線程數(shù)：" + executor.getMaximumPoolSize() +
" 線程池活躍度：" + divide(executor.getActiveCount(), executor.getMaximumPoolSize()) +
" 任務(wù)完成數(shù)：" + executor.getCompletedTaskCount() +
" 隊列大?。? + (queue.size() + queue.remainingCapacity()) +
" 當(dāng)前排隊線程數(shù)：" + queue.size() +
" 隊列剩余大小：" + queue.remainingCapacity() +
" 隊列使用度：" + divide(queue.size(), queue.size() + queue.remainingCapacity()));
}
這樣的話就可以實現(xiàn)動態(tài)調(diào)整線程池參數(shù)，這就很好的解決了我們線程池現(xiàn)有的痛點，不至于線上出了問題還得改代碼部署那么漫長的修復(fù)時間了，動態(tài)線程池大大簡化了運維以及開發(fā)快速修復(fù)相關(guān)問題的難度。
歡迎大家關(guān)注我的公眾號【老周聊架構(gòu)】，Java后端主流技術(shù)棧的原理、源碼分析、架構(gòu)以及各種互聯(lián)網(wǎng)高并發(fā)、高性能、高可用的解決方案。
喜歡的話，點贊、再看、分享三連。
點個在看你最好看