一、背景及JADE介绍
买药秒送是健康即时零售业务新的核心流量场域，面对京东首页高流量曝光，我们对频道页整个技术架构方案进行升级，保障接口高性能、系统高可用。
动态线程池是买药频道应用的技术之一，我们通过3轮高保真压测最终初步确定了线程池的核心参数。但我们仍面临一些保障系统稳定性问题：如何监控线程池运行状态？以及因流量飙升出现任务堆积和拒绝时能否实时报警，线程池核心参数能否做到不重启应用，动态调整即时生效？
经调研，业界成熟的动态线程池开源项目有 dynamic-tp 和 hippo4j，在京东内部应用比较广泛的方案是 JADE ，几种方案实现思路大致相同，感兴趣可自行了解。JADE 是由零售中台-研发架构组维护的项目，动态线程池是JADE的组件之一，其稳定性已得到广泛验证（集团应用 300+，零售交易服务中台应用 250+ ，其中 0 级应用 130+），与JADE相辅相成的还有万象平台：是可视化的JADE管理端，集成配置、监控、审批等能力的JADE可视化平台，可以更高效的使用JADE组件，进一步提高工作效率。
实现效果
接入JADE和万象后，买药秒送线程池秒级监控效果如下：实时监控线程池运行状态 以及 阈值报警。
下面我们从实践到原理一探究竟。
二、JADE动态线程池+万象可视化平台接入实践
JADE动态线程池和万象整体流程图如下：应用中需要引入 JADE、DUCC和 PFinder SDK，通过JADE创建线程池，线程池核心参数通过万象平台配置，集成 DUCC 实现动态调参，即时生效。线程池运行状态监控通过 PFinder 实现秒级监控。
1、引入JADE POM依赖，jade从1.2.4版本开始支持万象
1<!-- JADE 核心包，最新版本与发布说明--> 2<dependency> 3 <groupId>com.jd.jade</groupId> 4 <artifactId>jade</artifactId> 5 <version>1.2.4</version> 6</dependency> 7<!-- 引用 PFinder SDK 库 --> 8<dependency> 9 <groupId>com.jd.pfinder</groupId> 10 <artifactId>pfinder-profiler-sdk</artifactId> 11 <version>1.1.5-FINAL</version> 12</dependency> 13<!-- JADE 配置目前基于 XStream 进行 XML 格式序列化，若通过非 DBConfig 动态调参，需自行引入 --> 14<dependency> 15 <groupId>com.thoughtworks.xstream</groupId> 16 <artifactId>xstream</artifactId> 17 <version>1.4.19</version> 18</dependency> 19<!-- JADE 尚未完全解除 dbconfig 依赖（主要是 ConfigBase 基类及 XmlConfigService 接口），非交易应用，需自行引入 dbconfig-client-api 精简包，交易应用一般已直接引用 dbconfig-client 实现包 --> 20<dependency> 21 <groupId>com.jd.purchase.config</groupId> 22 <artifactId>dbconfig-client-api</artifactId> 23 <version>1.0.8</version> 24</dependency>
2、创建 jade.properties配置文件，并通过 Spring加载该配置文件。
•注意名字不能修改，JADE初始化会从该命名文件中加载配置属性
1# 万象平台环境配置 2jade.wx.env=pre 3 4# 以下为调试设置，线上环境无需配置 5jade.log.level=debug 6jade.meter.debug-enabled=true
•Spring加载 JADE配置文件
1<bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"> 2 <property name="locations"> 3 <list> 4 <!--JADE配置--> 5 <value>classpath:jade.properties</value> 6 </list> 7 </property> 8 <property name="fileEncoding"> 9 <value>UTF-8</value> 10 </property> 11</bean>
3、 配置JADE启动类，负责 JADE 自定义初始化 。
•如果不集成万象平台，则可以使用配置的DUCC空间配置和修改线程池参数。
• 【推荐】如果使用万象，万象会为JDOS应用默认创建一个DUCC空间，使用万象的DUCC进行配置和更新。
1/** 2 * @description:JADE配置类 3 * @author: rongtao7 4 * @date: 2024/4/5 1:09 下午 5 */ 6@Configuration 7public class JadeConfig { 8 9 @Value("ucc://${ducc.application}:${ducc.token}@${ducc.hostPort}/v1/namespace/${ducc.namespace}/config/${ducc.config}/profiles/${ducc.profile}?longPolling=15000") 10 private String duccUrl; 11 12 @Value("${jade.wx.env}") 13 private String wxEnv; 14 15 @Bean 16 public InitializeBean jadeInitBean() { 17 InitializeBean initializeBean = new InitializeBean(); 18 19 // 注意这里，如果 uri 中 config 不是命名为 jade，则 name 属性需要设置为 jade 20 ConfiguratorManager instance = new ConfiguratorManager(); 21 instance.addResource("jade", duccUrl); 22 initializeBean.setConfigServiceProvider(instance); 23 24 // 万象环境 25 initializeBean.setWxEnv(wxEnv); 26 27 return initializeBean; 28 } 29}
4、使用JADE创建线程池，并通过PFinder包装增强以支持trace的传递
•prestart() 用于预热核心线程
1/** 2 * 线程池配置类，集成JADE和万象平台 3 */ 4@Configuration 5public class TaskExecutePoolConfig { 6 7 /** 8 * 买药秒送频道线程池 9 */ 10 @Bean 11 public ExecutorService msChannelPagePool(){ 12 //JADE组件创建线程池 13 ThreadPoolExecutor threadPoolExecutor = ThreadPoolExecutorBuilder.newBuilder() 14 .name(ThreadPoolName.MS_CHANNEL_PAGE_POOL.name()) // 线程池名称 15 .core(200) // 核心线程数 16 .max(200) // 最大线程数 17 .queue(100) // 设置队列长度，此队列支持动态调整 18 .callerRuns() // 拒绝策略，内置监控、日志 19 .keepAliveTime(60L, TimeUnit.SECONDS) //线程存活时间 20 .prestart() // 预初始化所有核心线程数 21 .build(); 22 23 // Pfinder增强 24 return PfinderContext.executorServiceWrapper(threadPoolExecutor); 25 } 26}
5、万象平台接入
1）创建万象环境：第一次接入需要创建预发和生产环境。
2）创建万象线程池组件
6、验证效果
•线程池参数动态变更 - 万象，更新后可观测到如下日志，说明修改成功
1update executor 'MS_CHANNEL_PAGE_POOL' corePoolSize from 500 to 50 2update executor 'MS_CHANNEL_PAGE_POOL' maxPoolSize from 500 to 200 3update executor 'MS_CHANNEL_PAGE_POOL' keepAliveTime from 60 to 120 in seconds 4update executor 'MS_CHANNEL_PAGE_POOL' queueCapacity from 100 to 90
•线程池监控 - PFinder， key格式为：executor.线程池名称.线程池状态（活跃/核心/最大线程数、队列大小、拒绝任务数）
•注：应用需开启pfinder监控并且PFinder SDK 要和 agent版本兼容
•线程池任务RT监控 & 线程池状态监控：
•线程池队列参数配置异常报警：
以上几步操作，就完成了JADE和万象的动态线程池接入。下面从源码角度浅析一下原理。
三、原理源码浅析
动态线程池 的 核心本质 是对JDK的ThreadPoolExecutor包装增强，集成UMP、PFinder、Ducc、万象平台，以实现线程池的可视化管理、动态调参、监控报警能力。
线程池参数如何实现变更呢？
线程池有4个关键参数，即：核心线程数、最大线程数、队列大小、存活时间4个。
•核心、最大线程数、存活时间3个参数通过JDK ThreadPoolExecutor 提供了 setCorePoolSize 、setMaximumPoolSize 和 setKeepAliveTime 支持更新参数。
•但队列长度 capacity 是不支持修改的，其使用private final 修饰。JADE是通过 ResizeableLinkedBlockingQueue 实现队列长度可变，实现方式是继承LinkedBlockingQueue，通过反射修改队列长度。
下面是JADE动态线程池简易原理图：
从万象平台更新参数开始，万象会将配置数据保存到MySQL数据库中，并通过发布操作将更新的配置推送到JADE的DUCC集成模块 DuccConfigService ，Linstener 监听到配置变更后调用 ThreadPoolExecutorUpdater 更新线程池参数，更新参数是通过继承JDK 的ThreadPoolExecutor 实现更新，以及通过ResizeableLinkedBlockingQueue 修改队列长度。
JADE线程池监控能力通过Meter监控点 及 MeterRegistry监控工厂集成PFinder和UMP实现。
了解基础原理后，从 JADE配置类初始化过程 及 线程池创建过程，分别看一下源码实现。
> JADE配置类初始化过程 - 源码探究
JADE InitBeanBase 注入了Spring容器，并利用Spring InitializingBean afterPropertiesSet() 执行自定义初始化逻辑。
JADE 自定义初始化逻辑 总共有8个初始化步骤，我们只需要关注其中几个即可。
1public abstract class InitBeanBase implements InitializingBean, ApplicationContextAware, ApplicationListener<ContextRefreshedEvent> { 2 @Override 3 public void afterPropertiesSet() throws Exception { 4 log.info("jade init begin"); 5 //1.读取配置文件，设置@Val属性值 6 initProperties(); 7 //2.初始化日志级别 8 initLogLevel(); 9 //3.初始化零售DBConfig 10 initDbConfig(); 11 //4.初始化DUCC 12 initConfig(); 13 //5.初始化万象配置 14 initWX(); 15 //6.初始化 jvm ump key 16 initUmps(); 17 //7.初始化PFinderMeterRegistry监控工厂 18 initMeter(); 19 //8.初始化JSF监听注册 JSF POOL 20 initJsf(); 21 UST.record(getClass()); 22 log.info("jade init end"); 23 } 24}
1、`` initProperties()用于读取jade.properties配置文件，设置@Val属性值
•从根目录读取jade.properties配置文件，名字不可变，否则获取不到。
1public final class JadeConfigs { 2 3 //从根目录读取 jade.properties 4 private static synchronized Config initConfig() { 5 //略... 6 Object cfg = Thread.currentThread().getContextClassLoader().getResourceAsStream("jade.properties"); 7 } 8}
•为Bean的@Val注解标注的属性设置值，如果jade.properties配置了则使用配置的，否则使用默认值。
1public abstract class InitBeanBase implements InitializingBean, ApplicationContextAware, ApplicationListener<ContextRefreshedEvent> { 2 3 //为@Val注解标注的属性设置值 4 private void parseSpringValue(Config cfg) { 5 6 //Spring PropertyPlaceholderHelper：解析和替换占位符的工具 7 PropertyPlaceholderHelper helper = new PropertyPlaceholderHelper("${", "}", ":", true); 8 9 //反射获取所有字段 10 for (Field f : FieldUtils.getAllFields(getClass())) { 11 f.setAccessible(true); 12 if (f.get(this) != null) { // may set explicitly 13 continue; 14 } 15 16 //获取 @Val 注解 17 Val valAnno = f.getAnnotation(Val.class); 18 if (valAnno != null && StringUtils.isNotEmpty(valAnno.value())) { 19 try { 20 21 //从Config(jade.properties) 配置文件读取属性值，没有则为默认值。 22 String actualVal = helper.replacePlaceholders(valAnno.value(), k -> { 23 String v = cfg.getString(k); 24 if (v == null) { 25 v = applicationContext.getEnvironment().getProperty(k); 26 } 27 return v; 28 }); 29 if (actualVal != null) { 30 Function<String, ?> parser = TYPE_PARSERS.get(f.getType()); 31 if (parser != null) { 32 Object parsedVal = parser.apply(actualVal); 33 f.set(this, parsedVal); 34 } 35 } 36 } catch (Exception e) { 37 log.error("parse field {} error", f.getName()); 38 throw e; 39 } 40 } 41 } 42 } 43}
2、initConfig()****初始化配置类中的jade配置的ducc，如果不集成万象，则使用这个ducc配置。使用万象，则使用万象平台配置的ducc。
•代码与万象初始化逻辑相同，参考下面的即可。
3、initWX()初始化万象平台配置。
•万象初始化流程主要有3步骤：1.拼接使用万象默认配置的Ducc空间；2.启动监听；3.拉取配置更新JADE组件
•万象的默认Ducc空间格式为：通过应用名和环境Env的拼接：{ns：wxbizapps} {appName：diansong} {env：pre}
1class WXInit { 2 3 //万象初始化 4 private void init0() { 5 6 //1.万象默认的DUCC配置 7 String duccHost = DuccResource.getDefautHost(); 8 Config config = JadeConfigs.getConfig(); 9 String app = config.getString("jade.wx.app", "jdos_wxbizapps"); 10 String token = config.getString("jade.wx.token", getDefaultDuccToken(duccHost)); 11 String ns = config.getString("jade.wx.ns", "wxbizapps"); 12 String cfg = config.getString("jade.wx.cfg", Env.getAppName()); 13 if (failOrLog(cfg, "jade.wx.cfg")) { 14 return; 15 } 16 String env = initBean.getWxEnv(); 17 if (StringUtils.isEmpty(env)) { 18 env = config.getString("jade.wx.env"); 19 } 20 if (failOrLog(env, "jade.wx.env")) { 21 return; 22 } 23 String currentApp = Env.getDeployAppName(); 24 if (failOrLog(currentApp, "current app name")) { 25 return; 26 } 27 //DUCC URL拼接 28 String url = String.format(DuccResource.URL_FORMAT, app, token, duccHost, ns, cfg, env, 1000 * 60, isRequired()); 29 log.info("connect to wanxiang via {}", url); // TODO: mark token 30 31 //Resource Name jade-wx 32 String resxName = "jade-wx"; 33 ConfiguratorManager cm = new ConfiguratorManager(); 34 cm.setApplication(currentApp); 35 cm.addResource(resxName, url); 36 cm.start(); 37 38 //2.启动监听Ducc jade-wx 39 ConfigService configService = new DuccConfigService(cm); 40 41 //3.从万象平台拉配置更新JADE组件 42 configService.getConfig(resxName, JadeConfig.class); // TODO: not found, throws? 43 44 UST.record(getClass()); 45 } 46}
• 启动监听DUCC 调用 DuccConfigService init()初始化方法
1public class DuccConfigService implements ConfigService { 2 //构造方法，注入DUCC ConfiguratorManager 3 public DuccConfigService(@NonNull ConfiguratorManager configuratorManager) { 4 if (configuratorManager == null) { 5 throw new NullPointerException("configuratorManager is marked non-null but is null"); 6 } else { 7 //初始化 8 this.init(configuratorManager); 9 } 10 } 11}
•init()初始化方法中会启动万象DUCC的线程，并添加监听事件，监听Resource name 为 jade-wx的变化，变化后的回调函数通过 DuccConfigService.this.updateConfig(configuration)用来更新JADE组件
1//初始化方法 2private void init(ConfiguratorManager configuratorManager) { 3 try { 4 this.configuratorManager = configuratorManager; 5 //1.启动Ducc线程 6 if (!configuratorManager.isStarted()) { 7 if (StringUtils.isNotEmpty(Env.getDeployAppName())) { 8 System.setProperty("application.name", Env.getDeployAppName()); 9 } 10 11 configuratorManager.start(); 12 } 13 14 List<Resource> resources = DuccUtil.getResources(configuratorManager); 15 Iterator var3 = resources.iterator(); 16 17 while(var3.hasNext()) { 18 final Resource resource = (Resource)var3.next(); 19 20 //2.Ducc添加监听事件，Name是：jade-wx 21 configuratorManager.addListener(new ConfigurationListener() { 22 public String getName() { 23 return resource.getName(); 24 } 25 26 //回调函数更新JADE组件 27 public void onUpdate(Configuration configuration) { 28 DuccConfigService.this.updateConfig(configuration); 29 } 30 }); 31 } 32 33 UST.record(this.getClass()); 34 } catch (Throwable var5) { 35 throw var5; 36 } 37}
•DuccConfigService更新方法调用 JadeConfig 的init()方法，根据万象平台配置更新JADE各个组件，包括动态线程池。
1public class JadeConfig implements JadeConfigSupport, InitializingObject { 2 public static void init(JadeConfigSupport cfg) { 3 //JADE-日志组件 更新 4 5 //JADE-动态线程池组件 更新 6 ThreadPoolExecutorUpdater.update(cfg.getExecutorConfig()); 7 8 //JADE-本地缓存组件 更新 9 10 //.... 11 } 12}
5、ThreadPoolExecutorUpdater更新线程池参数核心类
•核心、最大线程数、存活时间是通过继承JDK ThreadPoolExecutor 实现更新的。
•在核心类中，当调大核心线程数后，会调用prestartAllCoreThreads() 对核心线程进行预热，所以不必担心调大核心线程数后发生的“抖动”问题（实际是创建线程的开销）。
•注意 core和max是一起更新的，否则可能会导致更改不生效的问题。
ThreadPoolExecutorUpdater 更新线程池主要有以下5个步骤。
•updatePoolSize更新核心、最大线程数，注意需要一起同步更新，否则可能导致更新失败问题
•setKeepAliveTime更新KeepAliveTime存活时间
•setCapacity 反射修改队列容量
•prestartAllCoreThreads() 预热核心线程数
•updateRejectSetting() 更新拒绝策略
1private static void update0(ExecutorConfigSupport.ExecutorSetting executorSetting, ThreadPoolExecutor executor) { 2 //1.更新核心、最大线程数，注意需要一起同步更新，否则可能导致更新失败问题 3 updatePoolSize(executorSetting, executor); 4 //2.更新KeepAliveTime存活时间 5 if (executorSetting.getKeepAliveSeconds() != null && executorSetting.getKeepAliveSeconds() != executor.getKeepAliveTime(TimeUnit.SECONDS)) { 6 executor.setKeepAliveTime(executorSetting.getKeepAliveSeconds(), TimeUnit.SECONDS); 7 } 8 //3.更新队列 9 if (executorSetting.getQueueCapacity() != null) { 10 if (executor.getQueue() instanceof LinkedBlockingQueue) { 11 LinkedBlockingQueue currentQueue = (LinkedBlockingQueue) executor.getQueue(); 12 int currentQueueCapacity = ResizableLinkedBlockingQueue.getCapacity(currentQueue); 13 if (executorSetting.getQueueCapacity() > 0 && executorSetting.getQueueCapacity() != currentQueueCapacity) { 14 15 //反射修改队列数量，signalNotFull 16 ResizableLinkedBlockingQueue.setCapacity(currentQueue, executorSetting.getQueueCapacity()); 17 } else if (executorSetting.getQueueCapacity() == 0) { 18 19 //调整队列数量为0，注意丢任务风险。 20 if (BooleanUtils.isTrue(executorSetting.getForceResizeQueue())) { 21 setWorkQueue(executor, new SynchronousQueue()); 22 } else { 23 // log 24 } 25 } 26 } 27 //else 省略 28 } 29 30 //4.预热核心线程数 31 if (BooleanUtils.toBoolean(executorSetting.getPrestartAllCoreThreads()) && executor.getPoolSize() < executor.getCorePoolSize()) { 32 int threads = executor.prestartAllCoreThreads(); 33 } 34 35 //5.更新拒绝策略 36 updateRejectSetting(executorSetting, executor); 37}
•队列长度修改通过ResizableLinkedBlockingQueue反射实现。
1//可动态调整容量的 BlockingQueue 2//HACK: 内部直接继承自 LinkedBlockingQueue，通过反射修改其 private final capacity 字段 3public class ResizableLinkedBlockingQueue<E> extends LinkedBlockingQueue<E> { 4 //反射设置队列大小 5 static <E> void setCapacity(LinkedBlockingQueue<E> queue, int capacity) { 6 int oldCapacity = getCapacity(queue); 7 FieldUtils.writeField(queue, FN_CAPACITY, capacity, true); 8 int size = queue.size(); 9 //如果队列中的任务已经达到老队列容量限制，并且新的容量大于队列任务数 10 if (size >= oldCapacity && capacity > size) { 11 // thanks to https://www.cnblogs.com/thisiswhy/p/15457810.html 12 MethodUtils.invokeMethod(queue, true, "signalNotFull"); 13 } 14 } 15}
•这里有一个细节，如果队列容量满了，当调整完队列数后，手动调用signalNotFull发出队列非满通知，唤醒阻塞线程，可以继续向队列插入任务了。
> 创建
JADE线程池build()- 源码探究
以下是我们通过 JADE ThreadPoolExecutorBuilder 创建线程池的 Bean，核心逻辑在 build() 封装。
1/** 2 * 秒送频道页线程池 3 */ 4@Bean 5public ExecutorService msChannelPagePool(){ 6 ThreadPoolExecutor threadPoolExecutor = ThreadPoolExecutorBuilder.newBuilder() 7 .name(ThreadPoolName.MS_CHANNEL_PAGE_POOL.name()) // 线程池名称 8 .core(200) // 核心线程数 9 .max(200) // 最大线程数 10 .queue(1024) // 设置队列长度，此队列支持动态调整 11 .callerRuns() // 快捷设置拒绝策略为丢弃，内置监控、日志 12 .keepAliveTime(60L, TimeUnit.SECONDS) //线程存活时间 13 .prestart() // 预初始化所有核心线程数 14 .build(); 15 16 return PfinderContext.executorServiceWrapper(threadPoolExecutor); 17}
•build() 主要逻辑有3步，1.创建线程池 ，2.启动所有核心线程， 3.注册线程池监控点
1public abstract class AbstractExecutorBuilder<B extends AbstractExecutorBuilder, E extends ThreadPoolExecutor> { 2 public synchronized E build() { 3 //1.创建线程池 4 this.executor = createExecutor(); 5 //2.启动所有核心线程 6 if (this.prestartAllCoreThreads) { 7 executor.prestartAllCoreThreads(); 8 } 9 //3.创建监控 10 initMonitor(); 11 return this.executor; 12 } 13}
•initMonitor()创建PFinder线程池监控，即 活跃线程数、核心/最大线程数，队列数量等。格式为：executor.线程池名.activeCount. (注意线程池一定要有名字)
•gauge() 方法内部集成PFinder，使用代码编程的方式进行Gauge埋点，用于记录线程池的瞬时值指标：活动线程数、核心/最大、队列大小等。PFinder埋点方式详见PFinder文档。
1public abstract class MeterRegistry<C extends MeterRegistryConfig> { 2 3 public List<Gauge> gaugeExecutor(String executorName, ThreadPoolExecutor executor) { 4 String namePrefix = "executor." + executorName; 5 return gaugeExecutor0(namePrefix, executor); 6 } 7 8 private List<Gauge> gaugeExecutor0(String namePrefix, ThreadPoolExecutor executor) { 9 namePrefix += "."; 10 List<Gauge> gauges = new ArrayList<>(); 11 if (getConfig().isThreadPoolAllMetricsEnabled()) { 12 gauges.add(gauge(namePrefix + "taskCount", executor::getTaskCount)); 13 gauges.add(gauge(namePrefix + "completedTaskCount", executor::getCompletedTaskCount)); 14 } 15 gauges.add(gauge(namePrefix + "activeCount", executor::getActiveCount)); 16 gauges.add(gauge(namePrefix + "corePoolSize", executor::getCorePoolSize)); 17 gauges.add(gauge(namePrefix + "maxPoolSize", executor::getMaximumPoolSize)); 18 gauges.add(gauge(namePrefix + "poolSize", executor::getPoolSize)); 19 gauges.add(gauge(namePrefix + "queueSize", () -> executor.getQueue().size())); 20 // 21 return gauges; 22 } 23}
四、避坑指南
•线程池必须有名字，监控依赖，并且不能重名。当系统有问题时也便于通过jstack等工具排查定位问题。
•应用需开启pfinder监控并且PFinder SDK 要和 agent版本兼容
•线程池创建后，线程不会立即启动，而是在有任务提交时才启动，启动的瞬间会因为创建线程的开销造成性能“抖动”，可以使用prestartAllCoreThreads() 预热核心线程。
•线程池的核心线程，默认是不会回收的，如果一个线程池活跃度长时间很低，建议调整核心线程数，过多的线程会浪费内存资源，影响系统稳定性。
•Future、CompletableFuture异步任务使用线程池时设置合理的超时时间，避免因外部服务故障或网络等问题导致任务长时间阻塞，造成资源浪费，严重甚至拖垮整个线程池，导致线上问题。
•同理，系统中请求外部Http请求时，必须设置超时时间，避免资源被长时间占用无法释放，影响系统性能和稳定性。