買藥秒送 JADE動態(tài)線程池實(shí)踐及原理淺析

一、背景及JADE介紹

買藥秒送是健康即時零售業(yè)務(wù)新的核心流量場域，面對京東首頁高流量曝光，我們對頻道頁整個技術(shù)架構(gòu)方案進(jìn)行升級，保障接口高性能、系統(tǒng)高可用。

動態(tài)線程池是買藥頻道應(yīng)用的技術(shù)之一，我們通過3輪高保真壓測最終初步確定了線程池的核心參數(shù)。但我們?nèi)悦媾R一些保障系統(tǒng)穩(wěn)定性問題：如何監(jiān)控線程池運(yùn)行狀態(tài)？以及因流量飆升出現(xiàn)任務(wù)堆積和拒絕時能否實(shí)時報警，線程池核心參數(shù)能否做到不重啟應(yīng)用，動態(tài)調(diào)整即時生效？

經(jīng)調(diào)研，業(yè)界成熟的動態(tài)線程池開源項目有 dynamic-tp 和 hippo4j，在京東內(nèi)部應(yīng)用比較廣泛的方案是 JADE ，幾種方案實(shí)現(xiàn)思路大致相同，感興趣可自行了解。JADE 是由零售中臺-研發(fā)架構(gòu)組維護(hù)的項目，動態(tài)線程池是JADE的組件之一，其穩(wěn)定性已得到廣泛驗(yàn)證（集團(tuán)應(yīng)用 300+，零售交易服務(wù)中臺應(yīng)用 250+ ，其中 0 級應(yīng)用 130+），與JADE相輔相成的還有萬象平臺：是可視化的JADE管理端，集成配置、監(jiān)控、審批等能力的JADE可視化平臺，可以更高效的使用JADE組件，進(jìn)一步提高工作效率。

實(shí)現(xiàn)效果

接入JADE和萬象后，買藥秒送線程池秒級監(jiān)控效果如下：實(shí)時監(jiān)控線程池運(yùn)行狀態(tài)以及閾值報警。

下面我們從實(shí)踐到原理一探究竟。

二、JADE動態(tài)線程池+萬象可視化平臺接入實(shí)踐

JADE動態(tài)線程池和萬象整體流程圖如下：應(yīng)用中需要引入 JADE、DUCC和 PFinder SDK，通過JADE創(chuàng)建線程池，線程池核心參數(shù)通過萬象平臺配置，集成 DUCC 實(shí)現(xiàn)動態(tài)調(diào)參，即時生效。線程池運(yùn)行狀態(tài)監(jiān)控通過 PFinder 實(shí)現(xiàn)秒級監(jiān)控。

1、引入JADE POM依賴，jade從1.2.4版本開始支持萬象

com.jd.jade jade 1.2.4 com.jd.pfinder pfinder-profiler-sdk 1.1.5-FINAL com.thoughtworks.xstream xstream 1.4.19 com.jd.purchase.config dbconfig-client-api 1.0.8

2、創(chuàng)建jade.properties配置文件，并通過Spring加載該配置文件。

?注意名字不能修改，JADE初始化會從該命名文件中加載配置屬性

# 萬象平臺環(huán)境配置 jade.wx.env=pre # 以下為調(diào)試設(shè)置，線上環(huán)境無需配置 jade.log.level=debug jade.meter.debug-enabled=true

?Spring加載JADE配置文件

classpath:jade.properties UTF-8

3、配置JADE啟動類，負(fù)責(zé) JADE 自定義初始化 。

?如果不集成萬象平臺，則可以使用配置的DUCC空間配置和修改線程池參數(shù)。

?【推薦】如果使用萬象，萬象會為JDOS應(yīng)用默認(rèn)創(chuàng)建一個DUCC空間，使用萬象的DUCC進(jìn)行配置和更新。

/** * @description:JADE配置類 * @author: rongtao7 * @date: 2024/4/5 1:09 下午 */ @Configuration public class JadeConfig { @Value("ucc://${ducc.application}:${ducc.token}@${ducc.hostPort}/v1/namespace/${ducc.namespace}/config/${ducc.config}/profiles/${ducc.profile}?longPolling=15000") private String duccUrl; @Value("${jade.wx.env}") private String wxEnv; @Bean public InitializeBean jadeInitBean() { InitializeBean initializeBean = new InitializeBean(); // 注意這里，如果 uri 中 config 不是命名為 jade，則 name 屬性需要設(shè)置為 jade ConfiguratorManager instance = new ConfiguratorManager(); instance.addResource("jade", duccUrl); initializeBean.setConfigServiceProvider(instance); // 萬象環(huán)境 initializeBean.setWxEnv(wxEnv); return initializeBean; } }

4、使用JADE創(chuàng)建線程池，并通過PFinder包裝增強(qiáng)以支持trace的傳遞

?prestart()用于預(yù)熱核心線程

/** * 線程池配置類，集成JADE和萬象平臺 */ @Configuration public class TaskExecutePoolConfig { /** * 買藥秒送頻道線程池 */ @Bean public ExecutorService msChannelPagePool(){ //JADE組件創(chuàng)建線程池 ThreadPoolExecutor threadPoolExecutor = ThreadPoolExecutorBuilder.newBuilder() .name(ThreadPoolName.MS_CHANNEL_PAGE_POOL.name()) // 線程池名稱 .core(200) // 核心線程數(shù) .max(200) // 最大線程數(shù) .queue(100) // 設(shè)置隊列長度，此隊列支持動態(tài)調(diào)整 .callerRuns() // 拒絕策略，內(nèi)置監(jiān)控、日志 .keepAliveTime(60L, TimeUnit.SECONDS) //線程存活時間 .prestart() // 預(yù)初始化所有核心線程數(shù) .build(); // Pfinder增強(qiáng) return PfinderContext.executorServiceWrapper(threadPoolExecutor); } }

5、萬象平臺接入

1）創(chuàng)建萬象環(huán)境：第一次接入需要創(chuàng)建預(yù)發(fā)和生產(chǎn)環(huán)境。

2）創(chuàng)建萬象線程池組件

6、驗(yàn)證效果

?線程池參數(shù)動態(tài)變更 - 萬象，更新后可觀測到如下日志，說明修改成功

update executor 'MS_CHANNEL_PAGE_POOL' corePoolSize from 500 to 50 update executor 'MS_CHANNEL_PAGE_POOL' maxPoolSize from 500 to 200 update executor 'MS_CHANNEL_PAGE_POOL' keepAliveTime from 60 to 120 in seconds update executor 'MS_CHANNEL_PAGE_POOL' queueCapacity from 100 to 90

?線程池監(jiān)控 - PFinder，key格式為：executor.線程池名稱.線程池狀態(tài)（活躍/核心/最大線程數(shù)、隊列大小、拒絕任務(wù)數(shù)）

?注：應(yīng)用需開啟pfinder監(jiān)控并且PFinder SDK 要和 agent版本兼容

?線程池任務(wù)RT監(jiān)控 & 線程池狀態(tài)監(jiān)控：

?線程池隊列參數(shù)配置異常報警：

以上幾步操作，就完成了JADE和萬象的動態(tài)線程池接入。下面從源碼角度淺析一下原理。

三、原理源碼淺析

動態(tài)線程池的核心本質(zhì)是對JDK的ThreadPoolExecutor包裝增強(qiáng)，集成UMP、PFinder、Ducc、萬象平臺，以實(shí)現(xiàn)線程池的可視化管理、動態(tài)調(diào)參、監(jiān)控報警能力。

線程池參數(shù)如何實(shí)現(xiàn)變更呢？

線程池有4個關(guān)鍵參數(shù)，即：核心線程數(shù)、最大線程數(shù)、隊列大小、存活時間4個。

?核心、最大線程數(shù)、存活時間3個參數(shù)通過JDK ThreadPoolExecutor提供了setCorePoolSize、setMaximumPoolSize和setKeepAliveTime支持更新參數(shù)。

?但隊列長度capacity是不支持修改的，其使用private final修飾。JADE是通過ResizeableLinkedBlockingQueue實(shí)現(xiàn)隊列長度可變，實(shí)現(xiàn)方式是繼承LinkedBlockingQueue，通過反射修改隊列長度。

下面是JADE動態(tài)線程池簡易原理圖：

從萬象平臺更新參數(shù)開始，萬象會將配置數(shù)據(jù)保存到MySQL數(shù)據(jù)庫中，并通過發(fā)布操作將更新的配置推送到JADE的DUCC集成模塊DuccConfigService，Linstener監(jiān)聽到配置變更后調(diào)用ThreadPoolExecutorUpdater更新線程池參數(shù)，更新參數(shù)是通過繼承JDK的ThreadPoolExecutor實(shí)現(xiàn)更新，以及通過ResizeableLinkedBlockingQueue修改隊列長度。

JADE線程池監(jiān)控能力通過Meter監(jiān)控點(diǎn) 及MeterRegistry監(jiān)控工廠集成PFinder和UMP實(shí)現(xiàn)。

了解基礎(chǔ)原理后，從JADE配置類初始化過程及線程池創(chuàng)建過程，分別看一下源碼實(shí)現(xiàn)。

> JADE配置類初始化過程 - 源碼探究

JADEInitBeanBase注入了Spring容器，并利用SpringInitializingBeanafterPropertiesSet()執(zhí)行自定義初始化邏輯。

JADE 自定義初始化邏輯總共有8個初始化步驟，我們只需要關(guān)注其中幾個即可。

public abstract class InitBeanBase implements InitializingBean, ApplicationContextAware, ApplicationListener { @Override public void afterPropertiesSet() throws Exception { log.info("jade init begin"); //1.讀取配置文件，設(shè)置@Val屬性值 initProperties(); //2.初始化日志級別 initLogLevel(); //3.初始化零售DBConfig initDbConfig(); //4.初始化DUCC initConfig(); //5.初始化萬象配置 initWX(); //6.初始化 jvm ump key initUmps(); //7.初始化PFinderMeterRegistry監(jiān)控工廠 initMeter(); //8.初始化JSF監(jiān)聽注冊 JSF POOL initJsf(); UST.record(getClass()); log.info("jade init end"); } }

1、initProperties()用于讀取jade.properties配置文件，設(shè)置@Val屬性值

?從根目錄讀取jade.properties配置文件，名字不可變，否則獲取不到。

public final class JadeConfigs { //從根目錄讀取 jade.properties private static synchronized Config initConfig() { //略... Object cfg = Thread.currentThread().getContextClassLoader().getResourceAsStream("jade.properties"); } }

?為Bean的@Val注解標(biāo)注的屬性設(shè)置值，如果jade.properties配置了則使用配置的，否則使用默認(rèn)值。

public abstract class InitBeanBase implements InitializingBean, ApplicationContextAware, ApplicationListener { //為@Val注解標(biāo)注的屬性設(shè)置值 private void parseSpringValue(Config cfg) { //Spring PropertyPlaceholderHelper：解析和替換占位符的工具 PropertyPlaceholderHelper helper = new PropertyPlaceholderHelper("${", "}", ":", true); //反射獲取所有字段 for (Field f : FieldUtils.getAllFields(getClass())) { f.setAccessible(true); if (f.get(this) != null) { // may set explicitly continue; } //獲取 @Val 注解 Val valAnno = f.getAnnotation(Val.class); if (valAnno != null && StringUtils.isNotEmpty(valAnno.value())) { try { //從Config(jade.properties) 配置文件讀取屬性值，沒有則為默認(rèn)值。 String actualVal = helper.replacePlaceholders(valAnno.value(), k -> { String v = cfg.getString(k); if (v == null) { v = applicationContext.getEnvironment().getProperty(k); } return v; }); if (actualVal != null) { Function parser = TYPE_PARSERS.get(f.getType()); if (parser != null) { Object parsedVal = parser.apply(actualVal); f.set(this, parsedVal); } } } catch (Exception e) { log.error("parse field {} error", f.getName()); throw e; } } } } }

2、initConfig()初始化配置類中的jade配置的ducc，如果不集成萬象，則使用這個ducc配置。使用萬象，則使用萬象平臺配置的ducc。

?代碼與萬象初始化邏輯相同，參考下面的即可。

3、initWX()初始化萬象平臺配置。

?萬象初始化流程主要有3步驟：1.拼接使用萬象默認(rèn)配置的Ducc空間；2.啟動監(jiān)聽；3.拉取配置更新JADE組件

?萬象的默認(rèn)Ducc空間格式為：通過應(yīng)用名和環(huán)境Env的拼接：{ns：wxbizapps} {appName：diansong} {env：pre}

class WXInit { //萬象初始化 private void init0() { //1.萬象默認(rèn)的DUCC配置 String duccHost = DuccResource.getDefautHost(); Config config = JadeConfigs.getConfig(); String app = config.getString("jade.wx.app", "jdos_wxbizapps"); String token = config.getString("jade.wx.token", getDefaultDuccToken(duccHost)); String ns = config.getString("jade.wx.ns", "wxbizapps"); String cfg = config.getString("jade.wx.cfg", Env.getAppName()); if (failOrLog(cfg, "jade.wx.cfg")) { return; } String env = initBean.getWxEnv(); if (StringUtils.isEmpty(env)) { env = config.getString("jade.wx.env"); } if (failOrLog(env, "jade.wx.env")) { return; } String currentApp = Env.getDeployAppName(); if (failOrLog(currentApp, "current app name")) { return; } //DUCC URL拼接 String url = String.format(DuccResource.URL_FORMAT, app, token, duccHost, ns, cfg, env, 1000 * 60, isRequired()); log.info("connect to wanxiang via {}", url); // TODO: mark token //Resource Name jade-wx String resxName = "jade-wx"; ConfiguratorManager cm = new ConfiguratorManager(); cm.setApplication(currentApp); cm.addResource(resxName, url); cm.start(); //2.啟動監(jiān)聽Ducc jade-wx ConfigService configService = new DuccConfigService(cm); //3.從萬象平臺拉配置更新JADE組件 configService.getConfig(resxName, JadeConfig.class); // TODO: not found, throws? UST.record(getClass()); } }

? 啟動監(jiān)聽DUCC調(diào)用DuccConfigService init()初始化方法

public class DuccConfigService implements ConfigService { //構(gòu)造方法，注入DUCC ConfiguratorManager public DuccConfigService(@NonNull ConfiguratorManager configuratorManager) { if (configuratorManager == null) { throw new NullPointerException("configuratorManager is marked non-null but is null"); } else { //初始化 this.init(configuratorManager); } } }

?init()初始化方法中會啟動萬象DUCC的線程，并添加監(jiān)聽事件，監(jiān)聽Resource name 為jade-wx的變化，變化后的回調(diào)函數(shù)通過DuccConfigService.this.updateConfig(configuration)用來更新JADE組件

//初始化方法 private void init(ConfiguratorManager configuratorManager) { try { this.configuratorManager = configuratorManager; //1.啟動Ducc線程 if (!configuratorManager.isStarted()) { if (StringUtils.isNotEmpty(Env.getDeployAppName())) { System.setProperty("application.name", Env.getDeployAppName()); } configuratorManager.start(); } List resources = DuccUtil.getResources(configuratorManager); Iterator var3 = resources.iterator(); while(var3.hasNext()) { final Resource resource = (Resource)var3.next(); //2.Ducc添加監(jiān)聽事件，Name是：jade-wx configuratorManager.addListener(new ConfigurationListener() { public String getName() { return resource.getName(); } //回調(diào)函數(shù)更新JADE組件 public void onUpdate(Configuration configuration) { DuccConfigService.this.updateConfig(configuration); } }); } UST.record(this.getClass()); } catch (Throwable var5) { throw var5; } }

?DuccConfigService更新方法調(diào)用JadeConfig的init()方法，根據(jù)萬象平臺配置更新JADE各個組件，包括動態(tài)線程池。

public class JadeConfig implements JadeConfigSupport, InitializingObject { public static void init(JadeConfigSupport cfg) { //JADE-日志組件 更新 //JADE-動態(tài)線程池組件 更新 ThreadPoolExecutorUpdater.update(cfg.getExecutorConfig()); //JADE-本地緩存組件 更新 //.... } }

5、ThreadPoolExecutorUpdater更新線程池參數(shù)核心類

?核心、最大線程數(shù)、存活時間是通過繼承JDK ThreadPoolExecutor實(shí)現(xiàn)更新的。

?在核心類中，當(dāng)調(diào)大核心線程數(shù)后，會調(diào)用prestartAllCoreThreads()對核心線程進(jìn)行預(yù)熱，所以不必?fù)?dān)心調(diào)大核心線程數(shù)后發(fā)生的“抖動”問題（實(shí)際是創(chuàng)建線程的開銷）。

?注意core和max是一起更新的，否則可能會導(dǎo)致更改不生效的問題。

ThreadPoolExecutorUpdater更新線程池主要有以下5個步驟。

?updatePoolSize更新核心、最大線程數(shù)，注意需要一起同步更新，否則可能導(dǎo)致更新失敗問題

?setKeepAliveTime更新KeepAliveTime存活時間

?setCapacity反射修改隊列容量

?prestartAllCoreThreads()預(yù)熱核心線程數(shù)

?updateRejectSetting()更新拒絕策略

private static void update0(ExecutorConfigSupport.ExecutorSetting executorSetting, ThreadPoolExecutor executor) { //1.更新核心、最大線程數(shù)，注意需要一起同步更新，否則可能導(dǎo)致更新失敗問題 updatePoolSize(executorSetting, executor); //2.更新KeepAliveTime存活時間 if (executorSetting.getKeepAliveSeconds() != null && executorSetting.getKeepAliveSeconds() != executor.getKeepAliveTime(TimeUnit.SECONDS)) { executor.setKeepAliveTime(executorSetting.getKeepAliveSeconds(), TimeUnit.SECONDS); } //3.更新隊列 if (executorSetting.getQueueCapacity() != null) { if (executor.getQueue() instanceof LinkedBlockingQueue) { LinkedBlockingQueue currentQueue = (LinkedBlockingQueue) executor.getQueue(); int currentQueueCapacity = ResizableLinkedBlockingQueue.getCapacity(currentQueue); if (executorSetting.getQueueCapacity() > 0 && executorSetting.getQueueCapacity() != currentQueueCapacity) { //反射修改隊列數(shù)量，signalNotFull ResizableLinkedBlockingQueue.setCapacity(currentQueue, executorSetting.getQueueCapacity()); } else if (executorSetting.getQueueCapacity() == 0) { //調(diào)整隊列數(shù)量為0，注意丟任務(wù)風(fēng)險。 if (BooleanUtils.isTrue(executorSetting.getForceResizeQueue())) { setWorkQueue(executor, new SynchronousQueue()); } else { // log } } } //else 省略 } //4.預(yù)熱核心線程數(shù) if (BooleanUtils.toBoolean(executorSetting.getPrestartAllCoreThreads()) && executor.getPoolSize() < executor.getCorePoolSize()) { int threads = executor.prestartAllCoreThreads(); } //5.更新拒絕策略 updateRejectSetting(executorSetting, executor); }

?隊列長度修改通過ResizableLinkedBlockingQueue反射實(shí)現(xiàn)。

//可動態(tài)調(diào)整容量的 BlockingQueue //HACK: 內(nèi)部直接繼承自 LinkedBlockingQueue，通過反射修改其 private final capacity 字段 public class ResizableLinkedBlockingQueue extends LinkedBlockingQueue { //反射設(shè)置隊列大小 static void setCapacity(LinkedBlockingQueue queue, int capacity) { int oldCapacity = getCapacity(queue); FieldUtils.writeField(queue, FN_CAPACITY, capacity, true); int size = queue.size(); //如果隊列中的任務(wù)已經(jīng)達(dá)到老隊列容量限制，并且新的容量大于隊列任務(wù)數(shù) if (size >= oldCapacity && capacity > size) { // thanks to https://www.cnblogs.com/thisiswhy/p/15457810.html MethodUtils.invokeMethod(queue, true, "signalNotFull"); } } }

?這里有一個細(xì)節(jié)，如果隊列容量滿了，當(dāng)調(diào)整完隊列數(shù)后，手動調(diào)用signalNotFull發(fā)出隊列非滿通知，喚醒阻塞線程，可以繼續(xù)向隊列插入任務(wù)了。

> 創(chuàng)建JADE線程池build()- 源碼探究

以下是我們通過 JADE ThreadPoolExecutorBuilder 創(chuàng)建線程池的 Bean，核心邏輯在 build() 封裝。

/** * 秒送頻道頁線程池 */ @Bean public ExecutorService msChannelPagePool(){ ThreadPoolExecutor threadPoolExecutor = ThreadPoolExecutorBuilder.newBuilder() .name(ThreadPoolName.MS_CHANNEL_PAGE_POOL.name()) // 線程池名稱 .core(200) // 核心線程數(shù) .max(200) // 最大線程數(shù) .queue(1024) // 設(shè)置隊列長度，此隊列支持動態(tài)調(diào)整 .callerRuns() // 快捷設(shè)置拒絕策略為丟棄，內(nèi)置監(jiān)控、日志 .keepAliveTime(60L, TimeUnit.SECONDS) //線程存活時間 .prestart() // 預(yù)初始化所有核心線程數(shù) .build(); return PfinderContext.executorServiceWrapper(threadPoolExecutor); }

?build()主要邏輯有3步，1.創(chuàng)建線程池 ，2.啟動所有核心線程， 3.注冊線程池監(jiān)控點(diǎn)

public abstract class AbstractExecutorBuilder{ public synchronized E build() { //1.創(chuàng)建線程池 this.executor = createExecutor(); //2.啟動所有核心線程 if (this.prestartAllCoreThreads) { executor.prestartAllCoreThreads(); } //3.創(chuàng)建監(jiān)控 initMonitor(); return this.executor; } }

?initMonitor()創(chuàng)建PFinder線程池監(jiān)控，即 活躍線程數(shù)、核心/最大線程數(shù)，隊列數(shù)量等。格式為：executor.線程池名.activeCount.(注意線程池一定要有名字)

?gauge()方法內(nèi)部集成PFinder，使用代碼編程的方式進(jìn)行Gauge埋點(diǎn)，用于記錄線程池的瞬時值指標(biāo)：活動線程數(shù)、核心/最大、隊列大小等。PFinder埋點(diǎn)方式詳見PFinder文檔。

public abstract class MeterRegistry { public List gaugeExecutor(String executorName, ThreadPoolExecutor executor) { String namePrefix = "executor." + executorName; return gaugeExecutor0(namePrefix, executor); } private List gaugeExecutor0(String namePrefix, ThreadPoolExecutor executor) { namePrefix += "."; List gauges = new ArrayList(); if (getConfig().isThreadPoolAllMetricsEnabled()) { gauges.add(gauge(namePrefix + "taskCount", executor::getTaskCount)); gauges.add(gauge(namePrefix + "completedTaskCount", executor::getCompletedTaskCount)); } gauges.add(gauge(namePrefix + "activeCount", executor::getActiveCount)); gauges.add(gauge(namePrefix + "corePoolSize", executor::getCorePoolSize)); gauges.add(gauge(namePrefix + "maxPoolSize", executor::getMaximumPoolSize)); gauges.add(gauge(namePrefix + "poolSize", executor::getPoolSize)); gauges.add(gauge(namePrefix + "queueSize", () -> executor.getQueue().size())); // return gauges; } }

四、避坑指南

?線程池必須有名字，監(jiān)控依賴，并且不能重名。當(dāng)系統(tǒng)有問題時也便于通過jstack等工具排查定位問題。

?應(yīng)用需開啟pfinder監(jiān)控并且PFinder SDK 要和 agent版本兼容

?線程池創(chuàng)建后，線程不會立即啟動，而是在有任務(wù)提交時才啟動，啟動的瞬間會因?yàn)閯?chuàng)建線程的開銷造成性能“抖動”，可以使用prestartAllCoreThreads()預(yù)熱核心線程。

?線程池的核心線程，默認(rèn)是不會回收的，如果一個線程池活躍度長時間很低，建議調(diào)整核心線程數(shù)，過多的線程會浪費(fèi)內(nèi)存資源，影響系統(tǒng)穩(wěn)定性。

?Future、CompletableFuture異步任務(wù)使用線程池時設(shè)置合理的超時時間，避免因外部服務(wù)故障或網(wǎng)絡(luò)等問題導(dǎo)致任務(wù)長時間阻塞，造成資源浪費(fèi)，嚴(yán)重甚至拖垮整個線程池，導(dǎo)致線上問題。

?同理，系統(tǒng)中請求外部Http請求時，必須設(shè)置超時時間，避免資源被長時間占用無法釋放，影響系統(tǒng)性能和穩(wěn)定性。

審核編輯 黃宇