SpringBoot——结合nacos实现动态刷新自定义线程池

Bertha 。 2023-09-26 20:22 42阅读 0赞

# 需求 #

1.  实现自定义线程池
2.  实现线程池参数动态刷新

# 自定义动态线程池模板 #

## 自定义动态线程池代码 ##

@Data
    @Slf4j
    @Configuration
    @RefreshScope
    @ConfigurationProperties("custom.threadpool")
    public class DemoThreadPool implements InitializingBean {
    
        /**
         * thread name format
         */
        public static final String DEMO_THREAD_NAME = "DEMO-THREAD-%d";
    
        /**
         * core pool size, default: 8
         * 1、CPU-intensive: CPU-cores + 1
         * 2、IO intensive:CPU-cores/(1-blocking-factor)， blocking factor: 0.8~0.9
         */
        @Value("${corePoolSize:8}")
        private Integer corePoolSize;
    
        /**
         * max pool size, default: 100
         */
        @Value("${maxPoolSize:100}")
        private Integer maximumPoolSize;
    
        /**
         * idle thread keep alive time (unit: s), default: 60
         */
        @Value("${keepAliveTime:60}")
        private Long keepAliveTime;
    
        /**
         * queue capacity size, default: 16
         */
        @Value("${capacitySize:16}")
        private Integer capacitySize;
        /**
         * dynamic refresh switch, default: false
         */
        @Value("${dynamicRefreshSwitch:false}")
        private Boolean dynamicRefreshSwitch;
    
        /**
         * listened service's nacos dataId
         */
        @Value("${dataId:demo-service}")
        private String dataId;
    
        /**
         * thread pool executor
         */
        public static ThreadPoolExecutor demoThreadPoolExecutor;
    
        /**
         * nacos config manager
         */
        @Autowired
        private NacosConfigManager nacosConfigManager;
    
        /**
         * nacos config properties
         */
        @Autowired
        private NacosConfigProperties nacosConfigProperties;
    
        /**
         * keepAliveTime key
         */
        private static final String KEEPALIVE_TIME_KEY = "keepAliveTime";
    
        /**
         * maxPoolSize key
         */
        private static final String MAX_POOL_SIZE_KEY = "maxPoolSize";
    
        /**
         * corePoolSize key
         */
        private static final String CORE_POOL_SIZE_KEY = "corePoolSize";
    
        /**
         * init thread pool info
         *
         * @throws Exception
         */
        @Override
        public void afterPropertiesSet() throws Exception {
            final String activeProfile = nacosConfigProperties.getEnvironment().getActiveProfiles()[0];
            final String dataId = getDataId() + "-" + activeProfile + ".yml";
            final String group = nacosConfigProperties.getGroup();
            this.refreshThreadPoolParams();
            if (dynamicRefreshSwitch) {
                // listen nacos config properties
                nacosConfigManager.getConfigService().addListener(dataId, group, new Listener() {
                    @Override
                    public Executor getExecutor() {
                        return null;
                    }
    
                    /**
                     * receive nacos config info
                     *
                     * @param configInfo
                     */
                    @Override
                    public void receiveConfigInfo(String configInfo) {
                        log.info("============ refresh thread pool params ============");
                        Map<String, Object> config = new Yaml().load(configInfo);
                        Map<String, Object> customConfigMap = (Map<String, Object>) config.get("custom");
                        Map<String, Object> threadPoolConfigMap = (Map<String, Object>) customConfigMap.get("threadpool");
                        log.info("============ threadPoolConfigMap: {} ============", JSONObject.toJSONString(threadPoolConfigMap));
    
    
                        // set new corePoolSize
                        Integer corePoolSizeNew = threadPoolConfigMap.containsKey(CORE_POOL_SIZE_KEY) ?
                                (Integer) threadPoolConfigMap.get(CORE_POOL_SIZE_KEY) : corePoolSize;
                        // set new maxPoolSize
                        Integer maxPoolSizeNew = threadPoolConfigMap.containsKey(MAX_POOL_SIZE_KEY) ?
                                (Integer) threadPoolConfigMap.get(MAX_POOL_SIZE_KEY) : maximumPoolSize;
                        // set new keepAliveTime
                        Long keepAliveTimeNew = threadPoolConfigMap.containsKey(KEEPALIVE_TIME_KEY) ?
                                Long.valueOf(threadPoolConfigMap.get(KEEPALIVE_TIME_KEY).toString()) : keepAliveTime;
                        // refresh core pool size
                        demoThreadPoolExecutor.setCorePoolSize(corePoolSizeNew);
                        // refresh max pool size
                        demoThreadPoolExecutor.setMaximumPoolSize(maxPoolSizeNew);
                        // idle thread can keep alive time
                        demoThreadPoolExecutor.setKeepAliveTime(keepAliveTimeNew, TimeUnit.SECONDS);
                        printThreadPoolInfo();
                    }
                });
            }
        }
    
        /**
         * refresh thread pool params
         */
        private void refreshThreadPoolParams() {
            ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat(DEMO_THREAD_NAME).get();
            demoThreadPoolExecutor = new ThreadPoolExecutor(
                    // core pool size
                    corePoolSize,
                    // max pool size
                    maximumPoolSize,
                    // idle thread alive time
                    keepAliveTime,
                    // alive time unit, s
                    TimeUnit.SECONDS,
                    // 吞吐量通常高于ArrayBlockingQueue
                    new LinkedBlockingDeque<>(capacitySize),
                    // custom thread factory
                    threadFactory,
                    // CallerRunsPolicy：the task will be executed by the caller thread
                    new ThreadPoolExecutor.CallerRunsPolicy());
        }
    
    
        /**
         * print thread pool info
         */
        public void printThreadPoolInfo() {
            String printLog = String.format("\r\ncore size: %s\r\nactive count: %s\r\n" +
                            "max pool size: %s\r\nqueue size: %s\r\ntask count: %s",
                    // core pool size
                    demoThreadPoolExecutor.getCorePoolSize(),
                    // current active count
                    demoThreadPoolExecutor.getActiveCount(),
                    // max pool size
                    demoThreadPoolExecutor.getMaximumPoolSize(),
                    // current queue size
                    demoThreadPoolExecutor.getQueue().size(),
                    // task count
                    demoThreadPoolExecutor.getTaskCount());
            log.info("thread pool info ====>>> {}", printLog);
        }
    }
    
    复制代码

## nacos配置 ##

custom:
      threadpool: 
        corePoolSize: 5
        maxPoolSize: 100
        keepAliveTime: 60
        dynamicRefreshSwitch: true
    复制代码

# 注意 #

1.  本文动态刷新结合nacos实现。
2.  本文动态刷新只实现线程池的`corePoolSize`，`maximumPoolSize`,`keepAliveTime`参数刷新。