動(dòng)態(tài)調(diào)整線程池參數(shù)實(shí)踐

一、線程池遇到的挑戰(zhàn)

我們上一篇 《一文讀懂線程池的實(shí)現(xiàn)原理 》已經(jīng)從線程池如何維護(hù)自身狀態(tài)、線程池如何管理任務(wù)、線程池如何管理線程三個(gè)維度來深入剖析線程池的底層原理與源碼剖析，這讓我們對(duì)線程池的原理有了較為深入的理解。這對(duì)我們多線程編程有很大的幫助，但在使用線程池時(shí)還是會(huì)面臨幾個(gè)棘手的問題。

• 開發(fā)人員個(gè)人經(jīng)驗(yàn)與水平參差不齊，配置線程池參數(shù)都是按照自己想法來，沒有統(tǒng)一的一個(gè)配置標(biāo)準(zhǔn)。

• 線程池執(zhí)行情況與任務(wù)類型相關(guān)性較大，IO 密集型和 CPU 密集型的任務(wù)運(yùn)行起來的情況差異非常大。

• 當(dāng)你配置好線程池后，有的時(shí)間段流量高峰期，導(dǎo)致線程池忙不過來；有的時(shí)間段流量低峰期，線程池比較空閑。這就會(huì)導(dǎo)致資源調(diào)度失衡，降低了系統(tǒng)的穩(wěn)定性。

我們先來看下美團(tuán)調(diào)研的業(yè)界一些線程池參數(shù)配置方案：

• 第一種方案是出自《Java并發(fā)編程實(shí)踐》，顯然和業(yè)務(wù)場景有所偏離。

• 第二種方案也不太合理，為什么呢？我們一個(gè)項(xiàng)目里一般來說不止一個(gè)自定義線程池吧？比如有專門處理數(shù)據(jù)異步持久化的線程池，有專門處理查詢請(qǐng)求的線程池，這樣去做一個(gè)簡單的線程隔離。但是如果都用這樣的參數(shù)配置的話，顯然是不合理的。

• 第三種方案雖然考慮到了業(yè)務(wù)場景，但這是理想狀態(tài)。流量是不可能這么均衡的，就拿美團(tuán)來說，下午 3、4 點(diǎn)的流量，能和 12 點(diǎn)左右午飯時(shí)的流量比嗎？

基于上面線程池的幾個(gè)痛點(diǎn)，那有沒有好的解決方案呢？有的，那就是動(dòng)態(tài)調(diào)整線程池參數(shù)。

盡管業(yè)界沒有一些成熟的經(jīng)驗(yàn)配置策略，那么我們是不是可以從修改線程池參數(shù)的成本入手？畢竟每次線上線程池故障的話，都得修改代碼里的線程池相應(yīng)的參數(shù)，然后再部署上線，這個(gè)過程在對(duì)可用性要求極高的項(xiàng)目中那是極其慢的，可能給公司造成巨大的損失。

既然改代碼里的線程池相應(yīng)的參數(shù)并上線這個(gè)過程慢，那我們是不是可以把相應(yīng)的線程池參數(shù)配置到分布式配置中心上去？實(shí)現(xiàn)線程池參數(shù)可動(dòng)態(tài)配置和即時(shí)生效，線程池參數(shù)動(dòng)態(tài)化前后的參數(shù)修改流程對(duì)比如下：

二、動(dòng)態(tài)化線程池

2.1 整體設(shè)計(jì)

動(dòng)態(tài)化線程池的核心設(shè)計(jì)包括以下三個(gè)方面：

2.1.1 簡化線程池配置

線程池構(gòu)造參數(shù)有 7 個(gè)，但是最核心的是 3 個(gè)：corePoolSize、maximumPoolSize、workQueue，它們最大程度地決定了線程池的任務(wù)分配和線程分配策略?？紤]到在實(shí)際應(yīng)用中我們獲取并發(fā)性的場景主要是兩種：

• 并行執(zhí)行子任務(wù)，提高響應(yīng)速度。這種情況下，應(yīng)該使用同步隊(duì)列，沒有什么任務(wù)應(yīng)該被緩存下來，而是應(yīng)該立即執(zhí)行。

• 并行執(zhí)行大批次任務(wù)，提升吞吐量。這種情況下，應(yīng)該使用有界隊(duì)列，使用隊(duì)列去緩沖大批量的任務(wù)，隊(duì)列容量必須聲明，防止任務(wù)無限制堆積。

所以線程池只需要提供這三個(gè)關(guān)鍵參數(shù)的配置，并且提供兩種隊(duì)列的選擇，就可以滿足絕大多數(shù)的業(yè)務(wù)需求。

2.1.2 參數(shù)可動(dòng)態(tài)修改

為了解決參數(shù)不好配，修改參數(shù)成本高等問題。在 Java 線程池留有高擴(kuò)展性的基礎(chǔ)上，封裝線程池，允許線程池監(jiān)聽同步外部的消息，根據(jù)消息進(jìn)行修改配置。將線程池的配置放置在平臺(tái)側(cè)，允許開發(fā)同學(xué)簡單的查看、修改線程池配置。

2.1.3 增加線程池監(jiān)控

對(duì)某事物缺乏狀態(tài)的觀測(cè)，就對(duì)其改進(jìn)無從下手。在線程池執(zhí)行任務(wù)的生命周期添加監(jiān)控能力，幫助開發(fā)同學(xué)了解線程池狀態(tài)。

動(dòng)態(tài)化線程池提供如下功能：

• 動(dòng)態(tài)調(diào)參：支持線程池參數(shù)動(dòng)態(tài)調(diào)整、界面化操作；包括修改線程池核心大小、最大核心大小、隊(duì)列長度等；參數(shù)修改后及時(shí)生效。

• 任務(wù)監(jiān)控：支持應(yīng)用粒度、線程池粒度、任務(wù)粒度的 Transaction 監(jiān)控；可以看到線程池的任務(wù)執(zhí)行情況、最大任務(wù)執(zhí)行時(shí)間、平均任務(wù)執(zhí)行時(shí)間、P95/P99線等。

• 負(fù)載告警：支持告警規(guī)則配置，當(dāng)超過閾值時(shí)（線程池隊(duì)列任務(wù)積壓到一定值、線程池負(fù)載數(shù)達(dá)到一定閾值）會(huì)通知相關(guān)的開發(fā)負(fù)責(zé)人。

• 操作監(jiān)控：創(chuàng)建、修改和刪除線程池都會(huì)通知到應(yīng)用的開發(fā)負(fù)責(zé)人。

• 操作日志：可以查看線程池參數(shù)的修改記錄，誰在什么時(shí)候修改了線程池參數(shù)、修改前的參數(shù)值是什么。

• 權(quán)限校驗(yàn)：只有應(yīng)用開發(fā)負(fù)責(zé)人才能夠修改應(yīng)用的線程池參數(shù)。

三、Apollo 分布式配置中心

了解完動(dòng)態(tài)化線程池的整體設(shè)計(jì)與功能架構(gòu)后，我相信你也可以設(shè)計(jì)出一款動(dòng)態(tài)線程池組件出來的。下面跟著老周來實(shí)踐一下動(dòng)態(tài)調(diào)整線程池參數(shù)，可能不像上面設(shè)計(jì)的那樣那么全面，但會(huì)把動(dòng)態(tài)調(diào)整線程池參數(shù)的核心給實(shí)現(xiàn)一下。

不難發(fā)現(xiàn)動(dòng)態(tài)化線程池的核心是配置管理，那我們就得找一個(gè)分布式配置中心，這里老周用的 Apollo，還有其它的像 Spring Cloud Config、disconf、某些大型互聯(lián)網(wǎng)公司自研的分布式配置中心等，根據(jù)自己的項(xiàng)目情況以及使用場景來選擇就行。

3.1 Apollo 總體設(shè)計(jì)

Apollo（阿波羅）是攜程框架部門研發(fā)的分布式配置中心，能夠集中化管理應(yīng)用不同環(huán)境、不同集群的配置，配置修改后能夠?qū)崟r(shí)推送到應(yīng)用端，并且具備規(guī)范的權(quán)限、流程治理等特性，適用于微服務(wù)配置管理場景。

3.1.1 基礎(chǔ)模型

如下圖即是 Apollo 的基礎(chǔ)模型：

1. 用戶在配置中心對(duì)配置進(jìn)行修改并發(fā)布

2. 配置中心通知 Apollo 客戶端有配置更新

3. Apollo 客戶端從配置中心拉取最新的配置、更新本地配置并通知到應(yīng)用

• Config Service 提供配置的讀取、推送等功能，服務(wù)對(duì)象是 Apollo 客戶端

• Admin Service 提供配置的修改、發(fā)布等功能，服務(wù)對(duì)象是 Apollo Portal（管理界面）

• Config Service 和 Admin Service 都是多實(shí)例、無狀態(tài)部署，所以需要將自己注冊(cè)到 Eureka 中并保持心跳

• 在 Eureka 之上我們架了一層 Meta Server 用于封裝 Eureka 的服務(wù)發(fā)現(xiàn)接口

• Client 通過域名訪問 Meta Server 獲取 Config Service 服務(wù)列表（IP+Port），而后直接通過 IP+Port 訪問服務(wù)，同時(shí)在 Client 側(cè)會(huì)做 load balance、錯(cuò)誤重試

• Portal 通過域名訪問 Meta Server 獲取 Admin Service 服務(wù)列表（IP+Port），而后直接通過 IP+Port 訪問服務(wù)，同時(shí)在 Portal 側(cè)會(huì)做 load balance、錯(cuò)誤重試

• 為了簡化部署，我們實(shí)際上會(huì)把 Config Service、Eureka 和 Meta Server 三個(gè)邏輯角色部署在同一個(gè) JVM 進(jìn)程中

3.2 服務(wù)端設(shè)計(jì)

3.2.1 配置發(fā)布后的實(shí)時(shí)推送設(shè)計(jì)

在配置中心中，一個(gè)重要的功能就是配置發(fā)布后實(shí)時(shí)推送到客戶端。本文重點(diǎn)分析配置更新推送方式，下面我們簡要看一下這塊是怎么設(shè)計(jì)實(shí)現(xiàn)的。

1. 用戶在 Portal 操作配置發(fā)布

2. Portal 調(diào)用 Admin Service 的接口操作發(fā)布

3. Admin Service 發(fā)布配置后，發(fā)送 ReleaseMessage 給各個(gè) Config Service

4. Config Service 收到 ReleaseMessage 后，通知對(duì)應(yīng)的客戶端

上圖的發(fā)送 ReleaseMessage 的實(shí)現(xiàn)方式詳情請(qǐng)往下繼續(xù)看：

Admin Service 在配置發(fā)布后，需要通知所有的 Config Service 有配置發(fā)布，從而 Config Service 可以通知對(duì)應(yīng)的客戶端來拉取最新的配置。

從概念上來看，這是一個(gè)典型的消息使用場景，Admin Service 作為 producer 發(fā)出消息，各個(gè) Config Service 作為 consumer 消費(fèi)消息。通過一個(gè)消息組件（Message Queue）就能很好的實(shí)現(xiàn) Admin Service 和 Config Service 的解耦。

在實(shí)現(xiàn)上，考慮到 Apollo 的實(shí)際使用場景，以及為了盡可能減少外部依賴，我們沒有采用外部的消息中間件，而是通過數(shù)據(jù)庫實(shí)現(xiàn)了一個(gè)簡單的消息隊(duì)列。

1. 客戶端和服務(wù)端保持了一個(gè)長連接，從而能第一時(shí)間獲得配置更新的推送。（通過 Http Long Polling 實(shí)現(xiàn)）

2. 客戶端還會(huì)定時(shí)從 Apollo 配置中心服務(wù)端拉取應(yīng)用的最新配置。

3. 客戶端從 Apollo 配置中心服務(wù)端獲取到應(yīng)用的最新配置后，會(huì)保存在內(nèi)存中。

4. 客戶端會(huì)把從服務(wù)端獲取到的配置在本地文件系統(tǒng)緩存一份。

5. 應(yīng)用程序可以從 Apollo 客戶端獲取最新的配置、訂閱配置更新通知。

我們從基礎(chǔ)模型、服務(wù)端設(shè)計(jì)、客戶端設(shè)計(jì)三個(gè)維度來分析了 Apollo 總體設(shè)計(jì)，相信你對(duì) Apollo 分布式配置中心有了全面且清晰的理解了。為了照顧沒有用過 Apollo 這款分布式配置中心的同學(xué)，老周這里還是簡單給個(gè) Apollo 開發(fā)樣例演示，希望對(duì)你后面的動(dòng)態(tài)調(diào)整線程池參數(shù)實(shí)踐有所幫助。

四、動(dòng)態(tài)調(diào)整線程池參數(shù)實(shí)踐

我們了解原理以及架構(gòu)后，那我們開始實(shí)踐了。

4.1 服務(wù)端安裝

請(qǐng)看官方文檔進(jìn)行相應(yīng)的安裝：https://ctripcorp.github.io/apollo/#/zh/deployment/quick-start

執(zhí)行啟動(dòng)腳本后，當(dāng)看到如下輸出后，就說明啟動(dò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 樣例測(cè)試

/**
 * 動(dòng)態(tài)獲取Apollo配置
 * 注意要配置：-Dapp.id=myApp -Denv=DEV -Dapollo.cluster=default -Ddev_meta=http://localhost:8080
 *
 * @author 微信公眾號(hào)【老周聊架構(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)在把配置項(xiàng)默認(rèn)的值 100 改為 200 程序輸出結(jié)果如下：

4.3 動(dòng)態(tài)線程池

上面我們把 Apollo 的動(dòng)態(tài)監(jiān)聽修改配置的功能整明白了以后，再把線程池和 Apollo 結(jié)合起來構(gòu)建動(dòng)態(tài)線程池那就方便多了。首先我們用默認(rèn)值構(gòu)建一個(gè)線程池，然后線程池會(huì)監(jiān)聽 Apollo 關(guān)于相關(guān)配置項(xiàng)，如果相關(guān)配置有變化則刷新相關(guān)參數(shù)。

代碼演示：

/**
 * 動(dòng)態(tài)線程池工廠
 *
 * @author 微信公眾號(hào)【老周聊架構(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ù)的隊(duì)列的長度 **/
    private Integer workQueueSize = 1000;

    /** 線程空閑時(shí)間 **/
    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("修改線程空閑時(shí)間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", "線程空閑時(shí)間"),
    ;

    @Getter
    private String param;

    @Getter
    private String desc;
}

/**
 * 動(dòng)態(tài)線程池執(zhí)行器
 *  
 * @author 微信公眾號(hào)【老周聊架構(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)的線程使用情況。

我們?cè)谂渲弥行男薷呐渲庙?xiàng)把核心線程數(shù)設(shè)置為 50，最大線程數(shù)設(shè)置為 100：

這里還可以在代碼中通過打印相應(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() +
            " 活動(dòng)線程數(shù)：" + executor.getActiveCount() +
            " 最大線程數(shù)：" + executor.getMaximumPoolSize() +
            " 線程池活躍度：" + divide(executor.getActiveCount(), executor.getMaximumPoolSize()) +
            " 任務(wù)完成數(shù)：" + executor.getCompletedTaskCount() +
            " 隊(duì)列大?。? + (queue.size() + queue.remainingCapacity()) +
            " 當(dāng)前排隊(duì)線程數(shù)：" + queue.size() +
            " 隊(duì)列剩余大?。? + queue.remainingCapacity() +
            " 隊(duì)列使用度：" + divide(queue.size(), queue.size() + queue.remainingCapacity()));
}

這樣的話就可以實(shí)現(xiàn)動(dòng)態(tài)調(diào)整線程池參數(shù)，這就很好的解決了我們線程池現(xiàn)有的痛點(diǎn)，不至于線上出了問題還得改代碼部署那么漫長的修復(fù)時(shí)間了，動(dòng)態(tài)線程池大大簡化了運(yùn)維以及開發(fā)快速修復(fù)相關(guān)問題的難度。