线程池实时管理与监控工具的实现与思考
0. 前言
Java线程池作为最常使用到的并发工具，大多数同学都有使用的经验，但你确定你是在正确的使用线程池吗？阿里Java代码规范要求我们不使用 Executors来快速创建线程池，但是抛弃Executors，使用其它方式创建线程池就一定不会出现问题吗？本文详细描述了一款Java线程池动态管理和实时监控插件的开发过程，希望大家对线程池有新的认识，主要内容如下：
我们在使用线程池过程中碰到了什么问题？ Java线程池是如何进行容量管理的？ 通过分析线程池源码来讲解如何动态修改Java线程池参数并进行有效的监控 线上运行的实际成果展示
1. 问题
Spring提供了非常友好的@Async注解帮助我们快速方便构造异步线程池，默认情况下Spring Async使用SimpleAsyncTaskExecutor来处理线程，本质上SimpleAsyncTaskExecutor不算线程池，每次方法调用都会创建新的线程，但是它提供了限流机制，通过concurrencyLimit属性来控制限流的开启（>=0:开启，-1:关闭，默认值为-1），所以通常情况下我们需要自定义线程池配置：
@EnableAsync@Configurationpublic class SpringAsyncConfig {@Bean(name = "commonExecutor")public Executor commonExecutor() {ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();taskExecutor.setCorePoolSize(5);taskExecutor.setMaxPoolSize(20);taskExecutor.setQueueCapacity(50);taskExecutor.setKeepAliveSeconds(60);taskExecutor.setAllowCoreThreadTimeOut(true);taskExecutor.setThreadNamePrefix("CommonExecutor-");taskExecutor.initialize();return taskExecutor;}}@Async("commonExecutor")public void doSth() {...}
于是在项目中我们这样使用线程池来进行异步化处理
@Async("commonExecutor")public void serviceA() {...}@Async("commonExecutor")public void serviceB() {...}@Async("commonExecutor")public void serviceC() {...}
在实际应用当中，这是一个十分危险的操作方式，在线上运行过程中，serviceA的QPS远高于serviceB和serviceC, 这样导致线程池被A大量占用，B和C的请求直接被reject或者因为长时间排队而超时。
2. 思考
针对上述问题，团队做了如下思考：
如何针对不同的接口使用相互隔离的线程池？ 如何合理的配置线程池参数？ 如何动态的对线程池参数进行调整？
2.1. 如何针对不同的业务接口使用相互隔离的线程池？
这个问题让我们很容易联想到Hystrix的资源隔离，同样是使用线程池技术来实现的，hystrix资源隔离的原理图如下图2所示：
在上图中，线程池A用来处理对service1和service2的请求，线程池B处理service3的请求，线程池C处理service4的请求，Hystrix通过@HystrixCommand( threadPoolKey="xxx" ) 指定当前HystrixCommand实例的threadPoolKey，相同threadPoolKey的方法将使用相同的线程池实例，为了让大家聚焦到线程池本身，这里不再对hystrix线程隔离的具体原理进行说明。类似的处理方式，我们在Spring Async 中通过配置不同的线程池实例来实现：
@EnableAsync@Configurationpublic class SpringAsyncConfig {@Bean(name = "executorA")public Executor executorA() {ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();taskExecutor.setCorePoolSize(5);taskExecutor.setMaxPoolSize(20);taskExecutor.setQueueCapacity(50);taskExecutor.setKeepAliveSeconds(60);taskExecutor.setAllowCoreThreadTimeOut(true);taskExecutor.setThreadNamePrefix("ExecutorA-");taskExecutor.initialize();return taskExecutor;}@Bean(name = "executorB")public Executor executorB() {ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();taskExecutor.setCorePoolSize(2);taskExecutor.setMaxPoolSize(8);taskExecutor.setQueueCapacity(20);taskExecutor.setKeepAliveSeconds(60);taskExecutor.setAllowCoreThreadTimeOut(true);taskExecutor.setThreadNamePrefix("ExecutorB-");taskExecutor.initialize();return taskExecutor;}@Bean(name = "executorC")public Executor executorC() {ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();taskExecutor.setCorePoolSize(10);taskExecutor.setMaxPoolSize(40);taskExecutor.setQueueCapacity(100);taskExecutor.setKeepAliveSeconds(60);taskExecutor.setAllowCoreThreadTimeOut(true);taskExecutor.setThreadNamePrefix("ExecutorC-");taskExecutor.initialize();return taskExecutor;}}
我们在不同的业务方法中使用@Async注解传入不同的实例name来使用不同的线程池实例：
@Async("executorA")public void doSthA() {...}@Async("executorB")public void doSthB() {...}@Async("executorC")public void doSthC() {...}
当然，在实际开发过程当中，不可能每个业务接口都单独使用一个线程池，否则会引入额外的开销。我们要根据实际的业务情况来进行线程池资源的规划。
2.2. 如何合理的配置线程池参数？
公式1：Nthreads = Ncpu
其中：Ncpu = cpu的核心数 ，Ucpu = cpu的利用率 W = 线程等待时间，C = 线程计算时间
此方案偏理论化，cpu的实际利用率（即分配多少cpu给线程池使用）和线程的计算，等待时间非常难评估，并且最后计算出来的结果也很容易偏离实际应用场景。
#####公式2：coreSize = 2
公式3：coreSize = tps
此种计算方式考虑到了实际的业务情况，引入了TPS和执行时间，假定每个接口的流量分配是平均的情况下是比较合理的，但是实际情况是接口的流量是随机的，在业务低峰期很低，在业务高峰期很高，如果按平均tps去配置线程池，可能没法在业务高峰期扛住系统的压力；按峰值来配置的话，在大多数时候线程池都是空闲了，增加了系统的开销
综上，不管哪种计算方式都无法准确的评估线程池的规模，我们需要找到一种比较灵活的配置方式。
####2.3. 如何动态的对线程池参数进行调整？有没有办法动态的调整线程池的各项参数，要解决这个问题，我们需要对Java线程池的原理有深入的了解。下图3是Java线程池创建工作线程Worker的流程图：
图3描述了一个线程池执行一个工作任务的基本流程，具体原理这里不再赘述，这里聊一下自己对JAVA线程池模型的一些思考：
Java线程池并没有使用非常复杂的数据结构来管理工作线程，代码也非常简洁，只用了一个HashSet来存储工作线程对象，Worker通过继承AQS并自己实现了一个不可重入的独占锁来保证多线程的并发安全； Java线程池本质是一个生产者消费者模型，Worker作为消费者不停的消费提交过来的Task, 所以如何提交任务消费的吞吐量和效率，同时控制消费者的规模，避免过多的占用系统资源，是线程池要解决的核心问题。
那么，Java线程池是如何对线程的容量进行管理的？对应的源码如下：
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));private static final int COUNT_BITS = Integer.SIZE - 3;private static final int CAPACITY = (1 << COUNT_BITS) - 1;// runState is stored in the high-order bitsprivate static final int RUNNING = -1 << COUNT_BITS;private static final int SHUTDOWN = 0 << COUNT_BITS;private static final int STOP = 1 << COUNT_BITS;private static final int TIDYING = 2 << COUNT_BITS;private static final int TERMINATED = 3 << COUNT_BITS;// Packing and unpacking ctlprivate static int runStateOf(int c) { return c & ~CAPACITY; }private static int workerCountOf(int c) { return c & CAPACITY; }private static int ctlOf(int rs, int wc) { return rs | wc; }
通过源码可以发现，ThreadPoolExecutor使用一个32位的AtomicInteger类型来同时标识线程池的运行状态和线程数量，然后通过位运算的左移操作来修改状态值，这是一种非常巧妙的算法设计，具体计算过程如下图所示：
当需要修改工作线程的数量时，通过CAS(Compare And Swap)操作来保证并发安全，CAS使用JVM底层Unsafe提供的API来直接修改变量的值，类似于乐观锁的机制。
/*** Attempts to CAS-increment the workerCount field of ctl.*/private boolean compareAndIncrementWorkerCount(int expect) {return ctl.compareAndSet(expect, expect + 1);}/*** Attempts to CAS-decrement the workerCount field of ctl.*/private boolean compareAndDecrementWorkerCount(int expect) {return ctl.compareAndSet(expect, expect - 1);}/*** Decrements the workerCount field of ctl. This is called only on* abrupt termination of a thread (see processWorkerExit). Other* decrements are performed within getTask.*/private void decrementWorkerCount() {do {} while (! compareAndDecrementWorkerCount(ctl.get()));}
下面对线程池的几个重要参数进行说明，部分细节会引用线程池的源码。
核心线程数：corePoolSize ， 最大线程数：maxPoolSize
如果workCount < corePoolSize，则创建新的Worker对象来处理任务，即使线程池容器中的其他Worker是空闲的。 如果corePoolSize < workCount < maximumPoolSize，若workQueue未满，则将请求放入workQueue中，等待有空闲的线程去从workQueue中取任务并处理，只有当workQueue满时才创建新的Worker去处理任务。 如果workCount > maximumPoolSize且workQueue已满，通过handler所指定的拒绝策略来处理。所以，任务提交时，判断的顺序为 corePoolSize –> queueSize –> maximumPoolSize。
最大线程数量: maximumPoolSize
线程池会保证所有可用状态的Worker线程的数量不会超过此最大值
工作队列: workQueue
当工作任务的数量超过coreSize时会将新创建的Worker对象放入等待队列，队列的实现可以自己选择，可选的实现有SynchronousQueue，LinkedBlockingQueue，ArrayBlockingQueue等，具体原理本文不再详述，请参考文章末尾引用的文档。
非核心线程所允许的空闲时间: keepAliveTime
当线程池中的线程数量大于corePoolSize的时候，如果这时没有新的任务提交，核心线程外的线程不会立即销毁，而是会等待，直到超过了keepAliveTime。
拒绝策略执行器: RejectedExecutionHandler
表示线程池的饱和策略。如果阻塞队列满了并且没有空闲的线程，这时如果继续提交任务，就需要采取一种策略处理该任务。线程池提供了4种策略：
AbortPolicy：直接抛出异常（默认策略） CallerRunsPolicy：用调用者所在的线程来执行任务 DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务 DiscardPolicy：直接丢弃任务
那我们有没有办法在线程池运行过程当中，动态的修改corePoolSize和maxPoolSize的值呢，以setCorePoolSize方法为例，我们看一下ThreadPoolExecutor的源码:
public void setCorePoolSize(int corePoolSize) {if (corePoolSize < 0)throw new IllegalArgumentException();int delta = corePoolSize - this.corePoolSize;this.corePoolSize = corePoolSize;if (workerCountOf(ctl.get()) > corePoolSize)interruptIdleWorkers();else if (delta > 0) {// We don't really know how many new threads are "needed".// As a heuristic, prestart enough new workers (up to new// core size) to handle the current number of tasks in// queue, but stop if queue becomes empty while doing so.int k = Math.min(delta, workQueue.size());while (k-- > 0 && addWorker(null, true)) {if (workQueue.isEmpty())break;}}}
通过源码可以看出，当新设置的corePoolSize的值大于当前值时，会按照等待队列中的任务数量来创建新的工作线程；当新设置的corePoolSize小于当前工作线程时，则会调用interruptIdleWorkers方法来中断空闲的工作线程，我们继续看一下interruptIdleWorkers()的源码：
private void interruptIdleWorkers(boolean onlyOne) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (Worker w : workers) {Thread t = w.thread;if (!t.isInterrupted() && w.tryLock()) {try {t.interrupt();} catch (SecurityException ignore) {} finally {w.unlock();}}if (onlyOne)break;}} finally {mainLock.unlock();}}
interruptIdleWorkers方法遍历workers集合中所有的工作线程，如果通过tryLock获取锁成功，就中断该线程。这里为什么需要使用mainLock？因为workers是HashSet类型的，不能保证线程安全。我们再来看看其它线程池参数的set方法：
public void setMaximumPoolSize(int maximumPoolSize) {if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize)throw new IllegalArgumentException();this.maximumPoolSize = maximumPoolSize;if (workerCountOf(ctl.get()) > maximumPoolSize)interruptIdleWorkers();}public void setKeepAliveTime(long time, TimeUnit unit) {if (time < 0)throw new IllegalArgumentException();if (time == 0 && allowsCoreThreadTimeOut())throw new IllegalArgumentException("Core threads must have nonzero keep alive times");long keepAliveTime = unit.toNanos(time);long delta = keepAliveTime - this.keepAliveTime;this.keepAliveTime = keepAliveTime;if (delta < 0)interruptIdleWorkers();}
发现都会调用interruptIdleWorkers()方法来对线程池的容量进行调节，所以结论是我们可以在线程池运行的过程中实时的对线程池的规模重新进行调节，并且是安全可靠的。
3. 实现
经过充分的调研，我们决定实现一个非常轻量级的动态创建线程池SpringBoot插件，主要功能如下图所示：
3.1 配置管理
将线程池的基本参数存储在Apollo配置中心和MySQL,可以很方便的进行配置的修改操作（第一个版本只支持Apollo配置中心，后续会提供MySQL的支持），Apollo配置参考如下：
#DataMonitor 监控业务报表数据naughty.threadpools.executors[0].corePoolSize = 10naughty.threadpools.executors[0].maximumPoolSize = 30naughty.threadpools.executors[0].keepAliveTime = 300naughty.threadpools.waitRefreshConfigSeconds = 10naughty.threadpools.executors[0].threadPoolName = DataMonitornaughty.threadpools.executors[0].queueCapacity = 5naughty.threadpools.executors[0].rejectedExecutionType = AbortPolicy#ExposureExecutor 用户准入接口异步调用naughty.threadpools.executors[1].threadPoolName = ExposureExecutornaughty.threadpools.executors[1].queueCapacity = 1naughty.threadpools.executors[1].rejectedExecutionType = CallerRunsPolicynaughty.threadpools.executors[1].corePoolSize = 5naughty.threadpools.executors[1].maximumPoolSize = 20naughty.threadpools.executors[1].keepAliveTime = 300
3.2 配置监听
我们利用Apollo的ChangeListener来实现对配置变更的监听，（如果是MySQL，可以修改完配置后直接通过HTTP接口通知客户端进行配置刷新），代码片段如下：
public class ThreadPoolConfigUpdateListener {@Value("${apollo.bootstrap.namespaces:application}")private String namespace;@Autowiredprivate DynamicThreadPoolFacade dynamicThreadPoolManager;@Autowiredprivate DynamicThreadPoolProperties poolProperties;@PostConstructpublic void init() {initConfigUpdateListener();}public void initConfigUpdateListener() {String apolloNamespace = namespace;if (StringUtils.hasText(poolProperties.getApolloNamespace())) {apolloNamespace = poolProperties.getApolloNamespace();}String finalApolloNamespace = apolloNamespace;Config config = ConfigService.getConfig(finalApolloNamespace);config.addChangeListener(changeEvent -> {try {Thread.sleep(poolProperties.getWaitRefreshConfigSeconds() * 1000);} catch (InterruptedException e) {log.error("配置刷新异常",e);}dynamicThreadPoolManager.refreshThreadPoolExecutor();log.info("线程池配置有变化，刷新完成");});}}
线程池配置的刷新的逻辑简单描述如下：
对于新增线程池配置，我们需要创建一个线程池实例 ，调用Spring容器提供的defaultListableBeanFactory.registerSingleton()方法将ThreadPoolTaskExecutor实例添加到Spring容器，最后调用autowireCapableBeanFactory.autowireBean(executor)方法注入相关的依赖。 如果是更新已经存在的线程池配置，直接从Spring容器中取出该实例然后刷新相应的配置即可。
public void refreshThreadPoolExecutor(DynamicThreadPoolProperties dynamicThreadPoolProperties) {dynamicThreadPoolProperties.getExecutors().forEach(poolProperties -> {NaughtyThreadPoolTaskExecutor executor = getExecutor(poolProperties.getThreadPoolName());if (executor == null) {executor = new NaughtyThreadPoolTaskExecutor();managerExecutor(executor, poolProperties);executor.setBlockingQueue(getBlockingQueue(poolProperties.getQueueType(), poolProperties.getQueueCapacity()));executor.initialize();//将new出的对象放入Spring容器中defaultListableBeanFactory.registerSingleton(poolProperties.getThreadPoolName(), executor);//自动注入依赖autowireCapableBeanFactory.autowireBean(executor);}else{managerExecutor(executor, poolProperties);BlockingQueue<Runnable> queue = executor.getThreadPoolExecutor().getQueue();if (queue instanceof ResizableCapacityLinkedBlockIngQueue) {((ResizableCapacityLinkedBlockIngQueue<Runnable>) queue).setCapacity(poolProperties.getQueueCapacity());}}});}private void managerExecutor(NaughtyThreadPoolTaskExecutor executor, ThreadPoolProperties poolProperties) {try {if (executor!=null) {executor.setBeanName(poolProperties.getThreadPoolName());executor.setCorePoolSize(poolProperties.getCorePoolSize());executor.setMaxPoolSize(poolProperties.getMaximumPoolSize());executor.setKeepAliveSeconds((int) poolProperties.getKeepAliveTime());executor.setRejectedExecutionHandler(this.getRejectedExecutionHandler(poolProperties.getRejectedExecutionType(), poolProperties.getThreadPoolName()));executor.setThreadPoolName(poolProperties.getThreadPoolName());}}catch(Exception e){log.error("Executor 参数设置异常",e);}}
3.3 状态监控
ThreadPoolExecutor提供了beforeExecute, afterExecute 等钩子方法，我们可以可以在钩子方法中对线程池任务的执行时间上报CAT,代码片段如下：
@Overrideprotected void beforeExecute(Thread t, Runnable r) {String threadName = Thread.currentThread().getName();Transaction transaction = Cat.newTransaction(threadPoolName, runnableNameMap.get(r.getClass().getSimpleName()));transactionMap.put(threadName, transaction);super.beforeExecute(t, r);}@Overrideprotected void afterExecute(Runnable r, Throwable t) {super.afterExecute(r, t);String threadName = Thread.currentThread().getName();Transaction transaction = transactionMap.get(threadName);transaction.setStatus(Message.SUCCESS);if (t != null) {Cat.logError(t);transaction.setStatus(t);}transaction.complete();transactionMap.remove(threadName);}
对应的CAT监控数据如下图所示:
通过使用CAT的StatusExtension，可以定时将线程池的运行时状态数据发送到CAT并生成柱状图，相关实现代码如下：
public StatusExtension registerStatusExtension(ThreadPoolProperties prop, Object object) {NaughtyThreadPoolTaskExecutor executor = (NaughtyThreadPoolTaskExecutor) object;StatusExtension statusExtension = new StatusExtension() {@Overridepublic String getId() {return "thread.pool.info." + prop.getThreadPoolName();}@Overridepublic String getDescription() {return "线程池监控";}@Overridepublic Map<String, String> getProperties() {AtomicLong rejectCount = getRejectCount(prop.getThreadPoolName());Map<String, String> pool = new HashMap<>();pool.put("activeCount", String.valueOf(executor.getActiveCount()));pool.put("keepAliveTime", String.valueOf(executor.getKeepAliveSeconds()));int coreSize = executor.getCorePoolSize();int maxSize = executor.getMaxPoolSize();if (coreSize!=0){pool.put("active/core", String.valueOf(Float.valueOf(executor.getActiveCount())/Float.valueOf(coreSize)));}if (maxSize!=0){pool.put("active/max", String.valueOf(Float.valueOf(executor.getActiveCount())/Float.valueOf(maxSize)));}pool.put("coreSize", String.valueOf(executor.getCorePoolSize()));pool.put("maxSize", String.valueOf(executor.getMaxPoolSize()));ThreadPoolExecutor threadPoolExecutor = executor.getThreadPoolExecutor();pool.put("completedTaskCount", String.valueOf(threadPoolExecutor.getCompletedTaskCount()));pool.put("largestPoolSize", String.valueOf(threadPoolExecutor.getLargestPoolSize()));pool.put("taskCount", String.valueOf(threadPoolExecutor.getTaskCount()));pool.put("rejectCount", String.valueOf(rejectCount == null ? 0 : rejectCount.get()));pool.put("queueSize", String.valueOf(threadPoolExecutor.getQueue().size()));return pool;}};StatusExtensionRegister.getInstance().register(statusExtension);return statusExtension;}
各项监控指标的说明如下：（以下部分观点可能需要经过进一步的验证，仅供大家参考）
active/coreSize ：活动线程数和核心线程数的比值， 其中active = executor.getActiveCount()，表示所有运行中的工作线程的数量，这个比值反应线程池的线程活跃状态，如果一直维持在一个很低的水平，则说明线程池需要进行缩容；如果长时间维持一个很大的数值，说明活跃度好，线程池利用率高。 active/maxSize ：活动线程数和最大线程数的比值，这个值可以配合上面的 active/coreSize 来看，当active/coreSize大于100%的时候，如果active/maxSize维持在一个较低的值，则说明当前线程池的负载偏低，如果大于60%或者更高，则说明线程池过载，需要及时调整线程池容量配置。 completedTaskCount：执行完毕的工作线程的总数，包含历史所有。 largestPoolSize：历史上线程池容量触达过的最大值 rejectCount：被拒绝的线程的数量，如果大量线程被拒绝，则说明当前线程池已经溢出了，需要及时调整线程池配置 queueSize：队列中工作线程的数量，如果大量的线程池在排队，说明coreSize已经不够用了，可以根据实际情况来调整，对于执行时间要求很严格的业务场景，可能需要通过提升coreSize来减少排队情况。
实际生产环境的线程池状态监控如下图所示：
展望
项目在使用线程池监控插件以后，获得了如下收益：
大量在后台使用线程池异步运行的批处理任务得到了有效隔离，管理和监控，通过查看CAT的各项埋点能够有针对性的对某个业务的批处理任务进行优化，避免影响服务的稳定性 对于高并发的线程池场景，在高负载的情况下能够第一时间收到告警，基本杜绝了因线程池溢出导致的接口不可用 作为一个轻量插件，通过极小的投入获得了还不错的收益，我们需要挖掘类似的高性价比的技术解决方案，不断提高研发团队的效能。
未来我们会考虑进行后续版本的迭代，尝试加入以下功能：
支持MySQL配置或其它配置中心 提供配置管理Web界面 提供可以独立部署的线程池监控Portal，不依赖CAT
作者介绍
大脑壳，信也科技北京研发中心资深专家，目前主要负责金融流量平台(fincloud.ppdapi.com)的架构设计和业务迭代。 workertang，信也科技北京研发中心研发专家，负责金融流量平台业的研发工作