在这里插入图片描述

此项目是已组件的形式进行搭建
分为 4 个部分 config、domain、registry、trigger

  • config：自动装配 Bean，获取项目中的线程池，创建 redis 消息中心
  • domain：存储线程池的 Bean，修改线程池
  • registry：redis 注册中心，将线程池的信息上传到 redis
  • trigger：
    • job：定时任务，将线程池信息上传到 redis 中
    • listener：订阅 redis 注册中心，等待注册中心推送消息，然后根据消息修改线程池，再将线程池配置信息上传到 redis 中

domain

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
public class ThreadPoolConfigEntity {

    /**
     * 应用名称
     */
    private String appName;

    /**
     * 线程池名称
     */
    private String threadPoolName;

    /**
     * 核心线程数
     */
    private int corePoolSize;

    /**
     * 最大线程数
     */
    private int maximumPoolSize;

    /**
     * 当前活跃线程数
     */
    private int activeCount;

    /**
     * 当前池中线程数
     */
    private int poolSize;

    /**
     * 队列类型
     */
    private String queueType;

    /**
     * 当前队列任务数
     */
    private int queueSize;

    /**
     * 队列剩余任务数
     */
    private int remainingCapacity;

    public ThreadPoolConfigEntity() {
    }

    public ThreadPoolConfigEntity(String appName, String threadPoolName) {
        this.appName = appName;
        this.threadPoolName = threadPoolName;
    }
}

线程池的配置信息，要存储在 redis 中

1
2
3
4
5
6
7
8
public interface IDynamicThreadPoolService {

    List<ThreadPoolConfigEntity> queryThreadPoolList();

    ThreadPoolConfigEntity queryThreadPoolConfigByName(String threadPoolName);

    void updateThreadPoolConfig(ThreadPoolConfigEntity threadPoolConfigEntity);
}

动态线程服务池接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
public class DynamicThreadPoolService implements IDynamicThreadPoolService {

    private final Logger logger = LoggerFactory.getLogger(DynamicThreadPoolService.class);

    private final String applicationName;
    private final Map<String, ThreadPoolExecutor> threadPoolExecutorMap;

    public DynamicThreadPoolService(String applicationName, Map<String, ThreadPoolExecutor> threadPoolMap) {
        this.applicationName = applicationName;
        this.threadPoolExecutorMap = threadPoolMap;
    }


    @Override
    public List<ThreadPoolConfigEntity> queryThreadPoolList() {
        Set<String> threadPoolBeanNames = threadPoolExecutorMap.keySet();
        List<ThreadPoolConfigEntity> threadPoolVOS = new ArrayList<>(threadPoolBeanNames.size());
        for (String beanName : threadPoolBeanNames) {
            ThreadPoolExecutor threadPoolExecutor = threadPoolExecutorMap.get(beanName);
            ThreadPoolConfigEntity threadPoolConfigVO = new ThreadPoolConfigEntity(applicationName, beanName);
            threadPoolConfigVO.setCorePoolSize(threadPoolExecutor.getCorePoolSize());
            threadPoolConfigVO.setMaximumPoolSize(threadPoolExecutor.getMaximumPoolSize());
            threadPoolConfigVO.setActiveCount(threadPoolExecutor.getActiveCount());
            threadPoolConfigVO.setPoolSize(threadPoolExecutor.getPoolSize());
            threadPoolConfigVO.setQueueType(threadPoolExecutor.getQueue().getClass().getSimpleName());
            threadPoolConfigVO.setQueueSize(threadPoolExecutor.getQueue().size());
            threadPoolConfigVO.setRemainingCapacity(threadPoolExecutor.getQueue().remainingCapacity());
            threadPoolVOS.add(threadPoolConfigVO);
        }
        return threadPoolVOS;
    }

    @Override
    public ThreadPoolConfigEntity queryThreadPoolConfigByName(String threadPoolName) {
        ThreadPoolExecutor threadPoolExecutor = threadPoolExecutorMap.get(threadPoolName);
        if (null == threadPoolExecutor) return new ThreadPoolConfigEntity(applicationName, threadPoolName);

        // 线程池配置数据
        ThreadPoolConfigEntity threadPoolConfigVO = new ThreadPoolConfigEntity(applicationName, threadPoolName);
        threadPoolConfigVO.setCorePoolSize(threadPoolExecutor.getCorePoolSize());
        threadPoolConfigVO.setMaximumPoolSize(threadPoolExecutor.getMaximumPoolSize());
        threadPoolConfigVO.setActiveCount(threadPoolExecutor.getActiveCount());
        threadPoolConfigVO.setPoolSize(threadPoolExecutor.getPoolSize());
        threadPoolConfigVO.setQueueType(threadPoolExecutor.getQueue().getClass().getSimpleName());
        threadPoolConfigVO.setQueueSize(threadPoolExecutor.getQueue().size());
        threadPoolConfigVO.setRemainingCapacity(threadPoolExecutor.getQueue().remainingCapacity());

        if (logger.isDebugEnabled()) {
            logger.info("动态线程池，配置查询 应用名:{} 线程名:{} 池化配置:{}", applicationName, threadPoolName, JSON.toJSONString(threadPoolConfigVO));
        }

        return threadPoolConfigVO;
    }

    @Override
    public void updateThreadPoolConfig(ThreadPoolConfigEntity threadPoolConfigEntity) {
        if (threadPoolConfigEntity ==  null || !applicationName.equals(threadPoolConfigEntity.getAppName())) {
            return;
        }
        ThreadPoolExecutor threadPoolExecutor = threadPoolExecutorMap.get(threadPoolConfigEntity.getThreadPoolName());
        if (null == threadPoolExecutor) {
            return;
        }

        threadPoolExecutor.setMaximumPoolSize(threadPoolConfigEntity.getMaximumPoolSize());
        threadPoolExecutor.setCorePoolSize(threadPoolConfigEntity.getCorePoolSize());
        logger.info("更新线程池 {} 的配置信息", JSON.toJSONString(threadPoolExecutor));
    }
}

线程池服务，获取线程池配置信息、更改线程配置

registry

1
2
3
4
5
6
7
public interface IRegistry {

    void reportThreadPool(List<ThreadPoolConfigEntity> threadPoolEntities);

    void reportThreadPoolConfigParameter(ThreadPoolConfigEntity threadPoolConfigEntity);

}

注册中心，数据上报，可以使用 redis、MySQL 等实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public class RedisRegistry implements IRegistry {

    private final RedissonClient redissonClient;


    public RedisRegistry(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }

    // 上报数据
    @Override
    public void reportThreadPool(List<ThreadPoolConfigEntity> threadPoolEntities) {
        RList<ThreadPoolConfigEntity> list = redissonClient.getList(RegistryEnumVO.THREAD_POOL_CONFIG_LIST_KEY.getKey());
        list.addAll(threadPoolEntities);
    }

    @Override
    public void reportThreadPoolConfigParameter(ThreadPoolConfigEntity threadPoolConfigEntity) {
        String strKey = RegistryEnumVO.THREAD_POOL_CONFIG_PARAMETER_LIST_KEY.getKey() + "_" + threadPoolConfigEntity.getAppName() + "_" + threadPoolConfigEntity.getThreadPoolName();
        RBucket<ThreadPoolConfigEntity> bucket = redissonClient.getBucket(strKey);
        bucket.set(threadPoolConfigEntity, Duration.ofDays(30));
    }

}

trigger

job

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public class ThreadPoolDataReportJob {

    private final Logger logger = LoggerFactory.getLogger(ThreadPoolDataReportJob.class);

    private final RedisRegistry redisRegistry;
    private final DynamicThreadPoolService dynamicThreadPoolService;

    public ThreadPoolDataReportJob(RedisRegistry redisRegistry, DynamicThreadPoolService dynamicThreadPoolService) {
        this.redisRegistry = redisRegistry;
        this.dynamicThreadPoolService = dynamicThreadPoolService;
    }

    @Scheduled(cron = "*/20 * * * * *")
    public void execReportThreadPoolList() {
        List<ThreadPoolConfigEntity> threadPoolConfigEntities = dynamicThreadPoolService.queryThreadPoolList();
        redisRegistry.reportThreadPool(threadPoolConfigEntities);
        logger.info("上部线程池数据, {}", JSON.toJSONString(threadPoolConfigEntities));

        for (ThreadPoolConfigEntity threadPoolConfigEntity : threadPoolConfigEntities) {
            redisRegistry.reportThreadPoolConfigParameter(threadPoolConfigEntity);
            logger.info("上报线程池的配置信息, {}", JSON.toJSONString(threadPoolConfigEntity));
        }

    }
}

定时将线程池信息上传到 redis 中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public class ThreadPoolDataReportJob {

    private final Logger logger = LoggerFactory.getLogger(ThreadPoolDataReportJob.class);

    private final RedisRegistry redisRegistry;
    private final DynamicThreadPoolService dynamicThreadPoolService;

    public ThreadPoolDataReportJob(RedisRegistry redisRegistry, DynamicThreadPoolService dynamicThreadPoolService) {
        this.redisRegistry = redisRegistry;
        this.dynamicThreadPoolService = dynamicThreadPoolService;
    }

    @Scheduled(cron = "*/20 * * * * *")
    public void execReportThreadPoolList() {
        List<ThreadPoolConfigEntity> threadPoolConfigEntities = dynamicThreadPoolService.queryThreadPoolList();
        redisRegistry.reportThreadPool(threadPoolConfigEntities);
        logger.info("上部线程池数据, {}", JSON.toJSONString(threadPoolConfigEntities));

        for (ThreadPoolConfigEntity threadPoolConfigEntity : threadPoolConfigEntities) {
            redisRegistry.reportThreadPoolConfigParameter(threadPoolConfigEntity);
            logger.info("上报线程池的配置信息, {}", JSON.toJSONString(threadPoolConfigEntity));
        }
    }
}

listener

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
public class ThreadPoolConfigAdjustListener implements MessageListener<ThreadPoolConfigEntity> {

    private final Logger logger = LoggerFactory.getLogger(ThreadPoolConfigAdjustListener.class);

    private final RedisRegistry redisRegistry;
    private final DynamicThreadPoolService dynamicThreadPoolService;

    public ThreadPoolConfigAdjustListener(RedisRegistry redisRegistry, DynamicThreadPoolService dynamicThreadPoolService) {
        this.dynamicThreadPoolService = dynamicThreadPoolService;
        this.redisRegistry = redisRegistry;
    }


    @Override
    public void onMessage(CharSequence charSequence, ThreadPoolConfigEntity threadPoolConfigEntity) {
        logger.info("动态线程池，调整线程池配置。线程池名称:{} 核心线程数:{} 最大线程数:{}",
                threadPoolConfigEntity.getThreadPoolName(), threadPoolConfigEntity.getPoolSize(), threadPoolConfigEntity.getMaximumPoolSize());

        dynamicThreadPoolService.updateThreadPoolConfig(threadPoolConfigEntity);

        List<ThreadPoolConfigEntity> threadPoolConfigEntities = dynamicThreadPoolService.queryThreadPoolList();
        redisRegistry.reportThreadPool(threadPoolConfigEntities);

        ThreadPoolConfigEntity poolConfigEntity = dynamicThreadPoolService.queryThreadPoolConfigByName(threadPoolConfigEntity.getThreadPoolName());
        redisRegistry.reportThreadPoolConfigParameter(poolConfigEntity);

    }
}

使用 redis 中的消息队列订阅消息，当腰更改线程池配置时，会发送一个消息，此监听者会接收到消息然后进行更改

config

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@Configuration
// 如果不添加这个 DynamicThreadPoolAutoProperties 会报错
@EnableConfigurationProperties(DynamicThreadPoolAutoProperties.class)
// 开启定时任务
@EnableScheduling
public class DynamicThreadPoolAutoConfig {

    private final Logger logger = LoggerFactory.getLogger(DynamicThreadPoolAutoConfig.class);
    private String applicationName;
    @Bean
    public DynamicThreadPoolService dynamicThreadPoolNameService(ApplicationContext applicationContext, Map<String, ThreadPoolExecutor> stringThreadPoolExecutorMap, RedissonClient redissonClient) {

        this.applicationName = applicationContext.getEnvironment().getProperty("spring.application.name");
        logger.info(applicationName);

        // 获取缓存数据，设置本地线程池配置
        Set<String> threadPoolKeys = stringThreadPoolExecutorMap.keySet();
        for (String threadPoolKey : threadPoolKeys) {
            ThreadPoolConfigEntity threadPoolConfigEntity = redissonClient.<ThreadPoolConfigEntity>getBucket(RegistryEnumVO.THREAD_POOL_CONFIG_PARAMETER_LIST_KEY.getKey() + "_" + applicationName + "_" + threadPoolKey).get();
            if (null == threadPoolConfigEntity) continue;
            ThreadPoolExecutor threadPoolExecutor = stringThreadPoolExecutorMap.get(threadPoolKey);
            threadPoolExecutor.setCorePoolSize(threadPoolConfigEntity.getCorePoolSize());
            threadPoolExecutor.setMaximumPoolSize(threadPoolConfigEntity.getMaximumPoolSize());
        }

        logger.info(stringThreadPoolExecutorMap.toString());

        return new DynamicThreadPoolService(applicationName, stringThreadPoolExecutorMap);
    }

    @Bean("redissonClient")
    public RedissonClient redissonClient(DynamicThreadPoolAutoProperties properties) {
        Config config = new Config();
        // 根据需要可以设定编解码器；https://github.com/redisson/redisson/wiki/4.-%E6%95%B0%E6%8D%AE%E5%BA%8F%E5%88%97%E5%8C%96
        config.setCodec(JsonJacksonCodec.INSTANCE);

        config.useSingleServer()
                .setAddress("redis://" + properties.getHost() + ":" + properties.getPort())
                .setPassword(properties.getPassword())
                .setConnectionPoolSize(properties.getPoolSize())
                .setConnectionMinimumIdleSize(properties.getMinIdleSize())
                .setIdleConnectionTimeout(properties.getIdleTimeout())
                .setConnectTimeout(properties.getConnectTimeout())
                .setRetryAttempts(properties.getRetryAttempts())
                .setRetryInterval(properties.getRetryInterval())
                .setPingConnectionInterval(properties.getPingInterval())
                .setKeepAlive(properties.isKeepAlive())
        ;

        RedissonClient redissonClient = Redisson.create(config);



        logger.info("动态线程池，注册器（redis）链接初始化完成。{} {} {}", properties.getHost(), properties.getPoolSize(), !redissonClient.isShutdown());

        return redissonClient;
    }

    @Bean("RedisRegistry")
    public RedisRegistry redisRegistry(RedissonClient redissonClient) {
        return new RedisRegistry(redissonClient);
    }

    @Bean
    public ThreadPoolDataReportJob threadPoolDataReportJob(RedisRegistry redisRegistry, DynamicThreadPoolService dynamicThreadPoolService) {
        return new ThreadPoolDataReportJob(redisRegistry, dynamicThreadPoolService);
    }

    @Bean
    public ThreadPoolConfigAdjustListener threadPoolConfigAdjustListener(RedisRegistry redisRegistry, DynamicThreadPoolService dynamicThreadPoolNameService) {
        return new ThreadPoolConfigAdjustListener(redisRegistry, dynamicThreadPoolNameService);
    }

    @Bean(name = "dynamicThreadPoolRedisTopic")
    public RTopic threadPoolConfigAdjustListener(RedissonClient redissonClient, ThreadPoolConfigAdjustListener threadPoolConfigAdjustListener) {
        RTopic rTopic = redissonClient.getTopic(RegistryEnumVO.DYNAMIC_THREAD_POOL_REDIS_TOPIC + "_" + applicationName);
        rTopic.addListener(ThreadPoolConfigEntity.class, threadPoolConfigAdjustListener);
        return rTopic;
    }
}

注册各种 Bean

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@Bean
public DynamicThreadPoolService dynamicThreadPoolNameService(ApplicationContext applicationContext, Map<String, ThreadPoolExecutor> stringThreadPoolExecutorMap, RedissonClient redissonClient) {

    this.applicationName = applicationContext.getEnvironment().getProperty("spring.application.name");
    logger.info(applicationName);

    // 获取缓存数据，设置本地线程池配置
    Set<String> threadPoolKeys = stringThreadPoolExecutorMap.keySet();
    for (String threadPoolKey : threadPoolKeys) {
        ThreadPoolConfigEntity threadPoolConfigEntity = redissonClient.<ThreadPoolConfigEntity>getBucket(RegistryEnumVO.THREAD_POOL_CONFIG_PARAMETER_LIST_KEY.getKey() + "_" + applicationName + "_" + threadPoolKey).get();
        if (null == threadPoolConfigEntity) continue;
        ThreadPoolExecutor threadPoolExecutor = stringThreadPoolExecutorMap.get(threadPoolKey);
        threadPoolExecutor.setCorePoolSize(threadPoolConfigEntity.getCorePoolSize());
        threadPoolExecutor.setMaximumPoolSize(threadPoolConfigEntity.getMaximumPoolSize());
    }

    logger.info(stringThreadPoolExecutorMap.toString());

    return new DynamicThreadPoolService(applicationName, stringThreadPoolExecutorMap);
}

当项目引用此组件时，原项目的线程池就会作为参数进入这个方法

许可协议

本文采用 署名-非商业性使用-相同方式共享 4.0 国际 许可协议，转载请注明出处。