作者：京东零售 张宾
1.背景
在后台开发中，会经常用到线程池技术，对于线程池核心参数的配置很大程度上依靠经验。然而，由于系统运行过程中存在的不确定性，我们很难一劳永逸地规划一个合理的线程池参数。在对线程池配置参数进行调整时，一般需要对服务进行重启，这样修改的成本就会偏高。一种解决办法就是，将线程池的配置放到配置平台侧，系统运行期间开发人员根据系统运行情况对核心参数进行动态配置。
本文以公司DUCC配置平台作为服务配置中心，以修改线程池核心线程数、最大线程数为例，实现一个简单的动态化线程池。
2.代码实现
当前项目中使用的是Spring 框架提供的线程池类ThreadPoolTaskExecutor，而ThreadPoolTaskExecutor底层又使用里了JDK中线程池类ThreadPoolExecutor，线程池类ThreadPoolExecutor有两个成员方法setCorePoolSize、setMaximumPoolSize可以在运行时设置核心线程数和最大线程数。
setCorePoolSize方法执行流程是：首先会覆盖之前构造函数设置的corePoolSize，然后，如果新的值比原始值要小，当多余的工作线程下次变成空闲状态的时候会被中断并销毁，如果新的值比原来的值要大且工作队列不为空，则会创建新的工作线程。流程图如下：
setMaximumPoolSize方法： 首先会覆盖之前构造函数设置的maximumPoolSize，然后，如果新的值比原来的值要小，当多余的工作线程下次变成空闲状态的时候会被中断并销毁。
Spring 框架提供的线程池类ThreadPoolTaskExecutor，此类封装了对ThreadPoolExecutor有两个成员方法setCorePoolSize、setMaximumPoolSize的调用。
基于以上源代码分析，要实现一个简单的动态线程池需要以下几步：
（1）定义一个动态线程池类，继承ThreadPoolTaskExecutor，目的跟非动态配置的线程池类ThreadPoolTaskExecutor区分开；
（2）定义和实现一个动态线程池配置定时刷的类，目的定时对比ducc配置的线程池数和本地应用中线程数是否一致，若不一致，则更新本地动态线程池线程池数；
（3）引入公司ducc配置平台相关jar包并创建一个动态线程池配置key；
（4）定义和实现一个应用启动后根据动态线程池Bean和从ducc配置平台拉取配置刷新应用中的线程数配置；
接下来代码一一实现：
（1）动态线程池类
1/** 2 * 动态线程池 3 * 4 */ 5public class DynamicThreadPoolTaskExecutor extends ThreadPoolTaskExecutor { 6}
（2）动态线程池配置定时刷新类
1@Slf4j 2public class DynamicThreadPoolRefresh implements InitializingBean { 3 /** 4 * Maintain all automatically registered and manually registered DynamicThreadPoolTaskExecutor. 5 */ 6 private static final ConcurrentMap<String, DynamicThreadPoolTaskExecutor> DTP_REGISTRY = new ConcurrentHashMap<>(); 7 8 /** 9 * @param threadPoolBeanName 10 * @param threadPoolTaskExecutor 11 */ 12 public static void registerDynamicThreadPool(String threadPoolBeanName, DynamicThreadPoolTaskExecutor threadPoolTaskExecutor) { 13 log.info("DynamicThreadPool register ThreadPoolTaskExecutor, threadPoolBeanName: {}, executor: {}", threadPoolBeanName, ExecutorConverter.convert(threadPoolBeanName, threadPoolTaskExecutor.getThreadPoolExecutor())); 14 DTP_REGISTRY.putIfAbsent(threadPoolBeanName, threadPoolTaskExecutor); 15 } 16 17 @Override 18 public void afterPropertiesSet() throws Exception { 19 this.refresh(); 20 //创建定时任务线程池 21 ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1, (new BasicThreadFactory.Builder()).namingPattern("DynamicThreadPoolRefresh-%d").daemon(true).build()); 22 //延迟1秒执行,每个1分钟check一次 23 executorService.scheduleAtFixedRate(new RefreshThreadPoolConfig(), 1000L, 60000L, TimeUnit.MILLISECONDS); 24 } 25 26 private void refresh() { 27 String dynamicThreadPool = ""; 28 try { 29 if (DTP_REGISTRY.isEmpty()) { 30 log.debug("DynamicThreadPool refresh DTP_REGISTRY is empty"); 31 return; 32 } 33 dynamicThreadPool = DuccConfigUtil.getValue(DuccConfigConstants.DYNAMIC_THREAD_POOL); 34 if (StringUtils.isBlank(dynamicThreadPool)) { 35 log.debug("DynamicThreadPool refresh dynamicThreadPool not config"); 36 return; 37 } 38 log.debug("DynamicThreadPool refresh dynamicThreadPool:{}", dynamicThreadPool); 39 List<ThreadPoolProperties> threadPoolPropertiesList = JsonUtil.json2Object(dynamicThreadPool, new TypeReference<List<ThreadPoolProperties>>() { 40 }); 41 if (CollectionUtils.isEmpty(threadPoolPropertiesList)) { 42 log.error("DynamicThreadPool refresh dynamicThreadPool json2Object error!{}", dynamicThreadPool); 43 return; 44 } 45 for (ThreadPoolProperties properties : threadPoolPropertiesList) { 46 doRefresh(properties); 47 } 48 } catch (Exception e) { 49 log.error("DynamicThreadPool refresh exception!dynamicThreadPool:{}", dynamicThreadPool, e); 50 } 51 } 52 53 /** 54 * @param properties 55 */ 56 private void doRefresh(ThreadPoolProperties properties) { 57 if (StringUtils.isBlank(properties.getThreadPoolBeanName()) 58 || properties.getCorePoolSize() < 1 59 || properties.getMaxPoolSize() < 1 60 || properties.getMaxPoolSize() < properties.getCorePoolSize()) { 61 log.error("DynamicThreadPool refresh, invalid parameters exist, properties: {}", properties); 62 return; 63 } 64 DynamicThreadPoolTaskExecutor threadPoolTaskExecutor = DTP_REGISTRY.get(properties.getThreadPoolBeanName()); 65 if (Objects.isNull(threadPoolTaskExecutor)) { 66 log.warn("DynamicThreadPool refresh, DTP_REGISTRY not found {}", properties.getThreadPoolBeanName()); 67 return; 68 } 69 ThreadPoolProperties oldProp = ExecutorConverter.convert(properties.getThreadPoolBeanName(), threadPoolTaskExecutor.getThreadPoolExecutor()); 70 if (Objects.equals(oldProp.getCorePoolSize(), properties.getCorePoolSize()) 71 && Objects.equals(oldProp.getMaxPoolSize(), properties.getMaxPoolSize())) { 72 log.warn("DynamicThreadPool refresh, properties of [{}] have not changed.", properties.getThreadPoolBeanName()); 73 return; 74 } 75 if (!Objects.equals(oldProp.getCorePoolSize(), properties.getCorePoolSize())) { 76 threadPoolTaskExecutor.setCorePoolSize(properties.getCorePoolSize()); 77 log.info("DynamicThreadPool refresh, corePoolSize changed!{} {}", properties.getThreadPoolBeanName(), properties.getCorePoolSize()); 78 } 79 if (!Objects.equals(oldProp.getMaxPoolSize(), properties.getMaxPoolSize())) { 80 threadPoolTaskExecutor.setMaxPoolSize(properties.getMaxPoolSize()); 81 log.info("DynamicThreadPool refresh, maxPoolSize changed!{} {}", properties.getThreadPoolBeanName(), properties.getMaxPoolSize()); 82 } 83 84 ThreadPoolProperties newProp = ExecutorConverter.convert(properties.getThreadPoolBeanName(), threadPoolTaskExecutor.getThreadPoolExecutor()); 85 log.info("DynamicThreadPool refresh result!{} oldProp:{},newProp：{}", properties.getThreadPoolBeanName(), oldProp, newProp); 86 } 87 88 private class RefreshThreadPoolConfig extends TimerTask { 89 private RefreshThreadPoolConfig() { 90 } 91 92 @Override 93 public void run() { 94 DynamicThreadPoolRefresh.this.refresh(); 95 } 96 } 97 98}
线程池配置类
1@Data 2public class ThreadPoolProperties { 3 /** 4 * 线程池名称 5 */ 6 private String threadPoolBeanName; 7 /** 8 * 线程池核心线程数量 9 */ 10 private int corePoolSize; 11 /** 12 * 线程池最大线程池数量 13 */ 14 private int maxPoolSize; 15}
(3)引入公司ducc配置平台相关jar包并创建一个动态线程池配置key
ducc配置平台使用见：https://cf.jd.com/pages/viewpage.action?pageId=403477057
动态线程池配置key：dynamic.thread.pool
配置value:
1[ 2 { 3 "threadPoolBeanName": "submitOrderThreadPoolTaskExecutor", 4 "corePoolSize": 32, 5 "maxPoolSize": 128 6 } 7]
(4) 应用启动刷新应用本地动态线程池配置
1@Slf4j 2public class DynamicThreadPoolPostProcessor implements BeanPostProcessor { 3 4 @Override 5 public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { 6 if (bean instanceof DynamicThreadPoolTaskExecutor) { 7 DynamicThreadPoolRefresh.registerDynamicThreadPool(beanName, (DynamicThreadPoolTaskExecutor) bean); 8 } 9 return bean; 10 } 11}
3.动态线程池应用
动态线程池Bean声明
1 <!-- 普通线程池 --> 2 <bean id="threadPoolTaskExecutor" class="com.jd.concurrent.ThreadPoolTaskExecutorWrapper"> 3 <!-- 核心线程数，默认为 --> 4 <property name="corePoolSize" value="128"/> 5 <!-- 最大线程数，默认为Integer.MAX_VALUE --> 6 <property name="maxPoolSize" value="512"/> 7 <!-- 队列最大长度，一般需要设置值>=notifyScheduledMainExecutor.maxNum；默认为Integer.MAX_VALUE --> 8 <property name="queueCapacity" value="500"/> 9 <!-- 线程池维护线程所允许的空闲时间，默认为60s --> 10 <property name="keepAliveSeconds" value="60"/> 11 <!-- 线程池对拒绝任务（无线程可用）的处理策略，目前只支持AbortPolicy、CallerRunsPolicy；默认为后者 --> 12 <property name="rejectedExecutionHandler"> 13 <!-- AbortPolicy:直接抛出java.util.concurrent.RejectedExecutionException异常 --> 14 <!-- CallerRunsPolicy:主线程直接执行该任务，执行完之后尝试添加下一个任务到线程池中，可以有效降低向线程池内添加任务的速度 --> 15 <!-- DiscardOldestPolicy:抛弃旧的任务、暂不支持；会导致被丢弃的任务无法再次被执行 --> 16 <!-- DiscardPolicy:抛弃当前任务、暂不支持；会导致被丢弃的任务无法再次被执行 --> 17 <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy"/> 18 </property> 19 </bean> 20 <!-- 动态线程池 --> 21 <bean id="submitOrderThreadPoolTaskExecutor" class="com.jd.concurrent.DynamicThreadPoolTaskExecutor"> 22 <!-- 核心线程数，默认为 --> 23 <property name="corePoolSize" value="32"/> 24 <!-- 最大线程数，默认为Integer.MAX_VALUE --> 25 <property name="maxPoolSize" value="128"/> 26 <!-- 队列最大长度，一般需要设置值>=notifyScheduledMainExecutor.maxNum；默认为Integer.MAX_VALUE --> 27 <property name="queueCapacity" value="500"/> 28 <!-- 线程池维护线程所允许的空闲时间，默认为60s --> 29 <property name="keepAliveSeconds" value="60"/> 30 <!-- 线程池对拒绝任务（无线程可用）的处理策略，目前只支持AbortPolicy、CallerRunsPolicy；默认为后者 --> 31 <property name="rejectedExecutionHandler"> 32 <!-- AbortPolicy:直接抛出java.util.concurrent.RejectedExecutionException异常 --> 33 <!-- CallerRunsPolicy:主线程直接执行该任务，执行完之后尝试添加下一个任务到线程池中，可以有效降低向线程池内添加任务的速度 --> 34 <!-- DiscardOldestPolicy:抛弃旧的任务、暂不支持；会导致被丢弃的任务无法再次被执行 --> 35 <!-- DiscardPolicy:抛弃当前任务、暂不支持；会导致被丢弃的任务无法再次被执行 --> 36 <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy"/> 37 </property> 38 </bean> 39 <!-- 动态线程池刷新配置 --> 40 <bean class="com.jd.concurrent.DynamicThreadPoolPostProcessor"/> 41 <bean class="com.jd.concurrent.DynamicThreadPoolRefresh"/>
业务类注入Spring Bean后，直接使用即可
1 @Resource 2 private ThreadPoolTaskExecutor submitOrderThreadPoolTaskExecutor; 3 4 5 Runnable asyncTask = ()->{...}; 6 CompletableFuture.runAsync(asyncTask, this.submitOrderThreadPoolTaskExecutor);
4.小结
本文从实际项目的业务痛点场景出发，并基于公司已有的ducc配置平台简单实现了线程池线程数量可配置。