项目线程池调优

0. 写在前面

1. 在网上搜集资料的时候，真正见证了什么叫做天下文章一大抄，更可恨的是错误的东西、未证明的东西也疯狂的抄袭。整个中文互联网环境全部污染了，真让人痛心疾首，这里实在忍不住要吐槽下。
2. 由于涉及公司业务，所以做了很大的业务脱敏，导致很多细节不清晰，大家关注方法就好。

1. 背景

在分析了我司某个应用之后，发现该应用线程模型不合理。大致有以下几个问题：

• 核心链路个数过多，作用不明晰，存在不必要的线程池
• 参数设置（core、max、拒绝策略）不合理
• 缺少监控和动态调整

因此，对项目的线程模型进行系统性的调优。

2. 项目分析

2.1 线程池工作原理

注意：

• 当线程数小于核心线程数时，就会创建线程，即使线程处于空闲时间。创建线程会加锁，源码中加锁是使用的CAS，在并发较高的情况下，会出现 CPU 自旋问题。
• 比较理想的情况是日常任务核心线程处理，少量任务进入队列，如遇高峰期，队列任务数可以排满，新建少量线程帮助处理任务。
• 如果系统有持续稳定的流量，即使线程池的线程数超过了核心线程数，也几乎很难触发 keepalive 机制进行销毁，这点非常非常重要，且及其容易被忽略。（本文会详细叙述这一点）

2.2 线程数配置理论支撑

计算密集型和 IO 密集型的线程池参数设置肯定不同，看了《Java并发编程实战》和《Java并发编程的艺术》这两本书，书上的方案偏理论，难落地。

Google 了一堆文章，大多是把这两本书上的内容用自己的话翻译了一下。总结下来，业界常用线程池参数设置方案如下图（截图来自于《美团技术团队》）：

第一种方案很难去确定计算时间和等待时间的比例，第二种方案缺少理论支撑，并且一个应用内大多会有多个线程池。考虑到我们系统的流量还算平均，第三种方案可以试一试。

而且发现国外不少文章在谈到线程池的时候会引用 “Little’s Law“（利特尔法则），第三种方案其实就是 “Little’s Law“ 思想。有兴趣可以自行搜索“利特尔法则”。

实际项目中，因为业务场景不同，线程池的参数设置没有“银弹”，很难找到一个公式去设置。可以采用 “评估+压测+监控+动态调整”的方式处理。

这里再次吐槽，网上的文章各种抄来抄去，我看到一大堆文章都提到对于 CPU 密集型的 Core 设置为 N+1，对于 IO 密集型的 Core 设置为 2N。也不知道这个结论哪来的，有什么理论支撑。根据我个人经验来看，对于重 IO 的高并发场景，核心线程数设置为 30N~50N 问题都不大。

2.3 项目链路梳理

因为涉及到业务隐私，所以这里很多流程和数据都做了泛化和简化处理。

整个链路大致如下：

Kafka Consumer 主线程（接收 Kafka 消息）——> 线程池 A（主要进行授权处理、业务过滤） ——> 线程池 B（组装数据 + Http 同步）。

业务逻辑有以下几点值得注意：

1. Consumer 主线程几乎没有任何业务逻辑，只是做基础的数据解析和内存级别的过滤， RT 极低；
2. 线程池 A 中的授权可能需要通过 Http 请求第三方获取 Token（绝大部分可走内存 + Redis），RT 较低，只有极端情况下缓存命中率低；
3. 线程池 B 是一个重 IO 操作，RT 较高。

2.4 原先架构存在的问题

• Kafka Consumer 主线程几乎无耗时，所以当极端场景有突发 Kafka 流量时，线程池 A 会 Reject；
• 线程池 A 的存在合理性有待商榷，我猜测因为历史原因胡乱添加线程池来提高所谓的性能，根本没有考虑链路使用的合理性；
• 有些平台授权在 Consumer 主线程中，有些在 线程池 A 中，乱七八糟；
• 线程池 B 的线程数配置应远大于 线程池 A，而实际配置却相反，瞎搞；

3. 优化方案

3.1 去除线程池 A

1. 授权、业务过滤统一到 Kafka Consumer 主线程中，对授权逻辑进行调优，引入 RT 监控；
2. 加入兜底策略，授权链路在完全优化完毕之前可能出现阻塞、卡死、RT 极高的极端情况，需要对这种 case 单独处理，避免 Consumer 主线程因为没有及时消费完毕消息导致消费集群 Rebalance；

3.2 重新评估线程池 B 的参数设置

这一块的核心是通过日常流量、峰值流量、任务 RT、能容忍的响应延迟来设置 Core、Max、队列长度的值。

初始设置值可以参考下面的公式：

corePoolSize = 日常 QPS RT(ms) / 1000，maxPoolSize = 峰值 QPS RT(ms) / 1000，maxPoolSize 可以稍微大一点，不超过 400 一般问题不大。

比如：单机 QPS 为 1000，RT 为 100ms，corePoolSize = 1000 * 100ms / 1000 = 100 个。如果你的 RT 超过了 1000ms，请先做 RT 优化。

queueCapacity = (corePoolSize/RT) * responsetime，其中 responsetime 是你的系统能容忍的响应延迟。

注意：根据压测和线上监控指标合理的优化设置值，不要设置完初始值之后就不管了。

3.3 引入线程池监控和告警

监控：

Prometheus + Grafana 大法好，可以定时（比如每隔 1 分钟），输出下线程池的各项参数值。

告警：

• 活跃线程数 > 最大线程数 * 0.8，最大线程数过小，如果长时间，则核心线程数也过小；
• 当前等待队列长度 > 设置值 * 0.8。

阈值可根据自身系统来设置。

3.4 线程池动态调整

技术原理：

JDK 的 ThreadPoolExecutor 提供了setCorePoolSize 和 setMaximumPoolSize 方法修改核心线程数、最大线程数。

以 setCorePoolSize 为例，如果新设置的核心数小于旧值，会回收空闲线程；如果大于旧值，会创建新线程。具体可以去看 ThreadPoolExecutor 类的相关源码和文档。

我们项目中实际使用的是 Spring 在 JDK 基础上封装的线程池 ThreadPoolTaskExecutor。代码如下所示：

/** * Set the ThreadPoolExecutor's core pool size. * Default is 1. * This setting can be modified at runtime, for example through JMX. */
public void setCorePoolSize(int corePoolSize) {
  synchronized (this.poolSizeMonitor) { 
    this.corePoolSize = corePoolSize;
    if (this.threadPoolExecutor != null) {                                                        	     this.threadPoolExecutor.setCorePoolSize(corePoolSize);
   }   
  }
}

setCorePoolSize 方法注释上面有写到 This setting can be modified at runtime，可以在运行时进行修改。

setMaximumPoolSize 方法大致同上。

除此之外，Spring 的 ThreadPoolTaskExecutor 还提供了设置队列长度的方法 setQueueCapacity，这个方法JDK 是没有的（JDK 的队列加了 final 关键字）。

ThreadPoolTaskExecutor 的 setQueueCapacity 只是初始化线程池用的，根本没有改变队列长度的功能。这点需要注意下，如果有动态调整队列长度的需求，可以自行实现阻塞队列。

4. 踩坑集-keepAlive 的坑

这里很多人可能会有误解，在经历一个流量高峰期后，可能线程池的线程数大于了核心线程数。如果此时经历流量低峰期，是不是任务都由核心线程执行，非核心线程会空闲出来，然后被回收销毁掉？

其实不是这样的，线程池是不区分核心线程和非核心线程，只要创建了，就是线程。并且，在稳定流量的系统的中，已创建线程很难通过 keepAlive 的配置进行销毁，因为线程执行任务是随机的。

举个例子，在低峰期时，假设 QPS 为 200，RT 为 100ms，任务只需要 20 个线程就可以执行完毕，但是此时线程池中有 40 个线程，这 40 个线程执行任务的概率都是相同的。

假设 keepAlive 配置为 60s，即某个线程在 60s 内没有执行任何任务，才会被销毁掉。每个线程每次获取执行任务的概率为 1/ 2，在这 60s 内，总共有 600 次机会执行任务，没有执行一次的概率为：0.5^600，约为 0。

结论：在稳定流量的系统中，线程池中的线程一旦被创建，很难通过 keepAlive 进行回收。

5.扩展: Spring线程池ThreadPoolTaskExecutor

本质是对 juc 线程池的封装。

TaskDecorator 装饰器

/**
 * A callback interface for a decorator to be applied to any {@link Runnable}
 * about to be executed.
 *
 * <p>Note that such a decorator is not necessarily being applied to the
 * user-supplied {@code Runnable}/{@code Callable} but rather to the actual
 * execution callback (which may be a wrapper around the user-supplied task).
 *
 * <p>The primary use case is to set some execution context around the task's
 * invocation, or to provide some monitoring/statistics for task execution.
 *
 * @author Juergen Hoeller
 * @since 4.3
 * @see TaskExecutor#execute(Runnable)
 * @see SimpleAsyncTaskExecutor#setTaskDecorator
 */
public interface TaskDecorator {

	/**
	 * Decorate the given {@code Runnable}, returning a potentially wrapped
	 * {@code Runnable} for actual execution.
	 * @param runnable the original {@code Runnable}
	 * @return the decorated {@code Runnable}
	 */
	Runnable decorate(Runnable runnable);

}

这个类是对任务的一个装饰器，类的注释最后一句 to provide some monitoring/statistics for task execution，非常适合用于指标监控。

关于 decorate 的说明：传参是实际的 Runnable，返回结果是装饰后的 Runnable。

public class TraceDecorator implements TaskDecorator {
    @Override
    public Runnable decorate(Runnable runnable) {
        return () -> {
            long startTime = System.currentTimeMillis();
            try {
                runnable.run();
            } finally {
                long endTime = System.currentTimeMillis();
                // 生产环境这里打trace监控任务耗时
                System.out.println("耗时为:" + (endTime - startTime) + "ms");
            }
        };
    }
}

测试类：

public class DecoratorTest {

    public static void main(String[] args) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(2);
        executor.setQueueCapacity(1000);
        executor.setTaskDecorator(new TraceDecorator());
        executor.initialize();
        executor.execute(() -> {
            // 模拟业务耗时
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

}

执行结果：

15:46:20.903 [main] INFO org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor - Initializing ExecutorService
耗时为:3004ms

参考资料

1. 《Java并发编程的艺术》、《Java并发编程实战》
2. Java线程池实现原理及其在美团业务中的实践
3. https://engineering.zalando.com/posts/2019/04/how-to-set-an-ideal-thread-pool-size.html
4. https://juejin.im/post/6844904101881315335