Dynamic Thread Pool Component with Cat Monitoring and Nacos Configuration for Spring Cloud

Thread pools are frequently used for asynchronous and batch processing, but static configurations often require application restarts to adjust core parameters or to monitor queue sizes, leading to operational challenges.

The author created a dynamic thread‑pool module in the open‑source Kitty project, adding Cat monitoring, dynamic core‑parameter changes, and queue‑capacity alerts. The component is not published to Maven Central, so it must be built locally or from a private repository.

Usage

Add the Maven dependency:

<dependency>
    <groupId>com.cxytiandi</groupId>
    <artifactId>kitty-spring-cloud-starter-dynamic-thread-pool</artifactId>
</dependency>

Configure the thread‑pool in Nacos (e.g., kitty-cloud-thread-pool.properties) with properties such as corePoolSize, maximumPoolSize, queueCapacity, and alert thresholds.

kitty.threadpools.nacosDataId=kitty-cloud-thread-pool.properties
kitty.threadpools.nacosGroup=BIZ_GROUP
kitty.threadpools.accessToken=ae6eb1e9e6964d686d2f2e8127d0ce5b31097ba23deee6e4f833bc0a77d5b71d
kitty.threadpools.secret=SEC6ec6e31d1aa1bdb2f7fd5eb5934504ce09b65f6bdc398d00ba73a9857372de00
kitty.threadpools.owner=尹吉欢
kitty.threadpools.executors[0].threadPoolName=TestThreadPoolExecutor
kitty.threadpools.executors[0].corePoolSize=4
kitty.threadpools.executors[0].maximumPoolSize=4
kitty.threadpools.executors[0].queueCapacity=5
kitty.threadpools.executors[0].queueCapacityThreshold=5
kitty.threadpools.executors[1].threadPoolName=TestThreadPoolExecutor2
kitty.threadpools.executors[1].corePoolSize=2
kitty.threadpools.executors[1].maximumPoolSize=4

Inject and use the pool:

@Autowired
private DynamicThreadPoolManager dynamicThreadPoolManager;

dynamicThreadPoolManager.getThreadPoolExecutor("TestThreadPoolExecutor").execute(() -> {
    log.info("线程池的使用");
    try {
        Thread.sleep(30000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}, "getArticle");

Extended Features

Task execution monitoring in Cat reports (transaction type is the thread‑pool name, task name is displayed).

Dynamic modification of core parameters ( corePoolSize, maximumPoolSize, queueCapacity, etc.) via Nacos.

Queue‑capacity alert when the number of pending tasks exceeds queueCapacityThreshold.

Reject‑count alert for tasks rejected by the configured RejectedExecutionHandler (e.g., AbortPolicy).

Runtime endpoint /actuator/thread-pool exposing pool statistics.

Custom rejection strategies via SPI (e.g., implementing RejectedExecutionHandler).

Custom alarm notifications by implementing ThreadPoolAlarmNotify (e.g., SMS alerts).

Code Implementation Highlights

Creating the pool uses a custom KittyThreadPoolExecutor that integrates Cat tracing:

/**
 * 创建线程池
 * @param threadPoolProperties
 */
private void createThreadPoolExecutor(DynamicThreadPoolProperties threadPoolProperties) {
    threadPoolProperties.getExecutors().forEach(executor -> {
        KittyThreadPoolExecutor threadPoolExecutor = new KittyThreadPoolExecutor(
                executor.getCorePoolSize(),
                executor.getMaximumPoolSize(),
                executor.getKeepAliveTime(),
                executor.getUnit(),
                getBlockingQueue(executor.getQueueType(), executor.getQueueCapacity(), executor.isFair()),
                new KittyThreadFactory(executor.getThreadPoolName()),
                getRejectedExecutionHandler(executor.getRejectedExecutionType(), executor.getThreadPoolName()),
                executor.getThreadPoolName());
        threadPoolExecutorMap.put(executor.getThreadPoolName(), threadPoolExecutor);
    });
}

Configuration changes are listened to via Nacos and applied at runtime:

/**
 * 监听配置修改，spring-cloud-alibaba 2.1.0版本不支持@NacosConfigListener的监听
 */
public void initConfigUpdateListener(DynamicThreadPoolProperties dynamicThreadPoolProperties) {
    ConfigService configService = nacosConfigProperties.configServiceInstance();
    try {
        configService.addListener(dynamicThreadPoolProperties.getNacosDataId(), dynamicThreadPoolProperties.getNacosGroup(), new AbstractListener() {
            @Override
            public void receiveConfigInfo(String configInfo) {
                new Thread(() -> refreshThreadPoolExecutor()).start();
                log.info("线程池配置有变化，刷新完成");
            }
        });
    } catch (NacosException e) {
        log.error("Nacos配置监听异常", e);
    }
}

Refreshing updates core size, max size, keep‑alive, rejection handler, and queue capacity (if the queue is a ResizableCapacityLinkedBlockingQueue).

Cat reporting is done via a StatusExtension that periodically pushes metrics such as active count, completed tasks, largest pool size, reject count, and waiting task count.

A custom actuator endpoint exposes the pool data as JSON for external monitoring tools.

Before‑execute and after‑execute hooks are overridden to create Cat transactions around each task, ensuring accurate execution‑time monitoring.

The article concludes with a link to the full source code and invites readers to star the repository.