线程池原理与最佳实践

线程池是 Java 并发编程中最重要的基础设施之一。创建线程的成本不低——需要调用操作系统 API、分配栈内存、初始化线程上下文，反复创建销毁线程会严重拖慢系统性能。线程池通过复用已创建的线程来执行任务，避免频繁创建销毁的开销。

为什么需要线程池

三个核心好处：

1. 降低资源消耗：复用线程，避免反复创建/销毁
2. 提高响应速度：任务到达时线程已存在，无需等待创建
3. 便于管理：控制最大并发数，防止资源耗尽

ThreadPoolExecutor 的 7 个参数

这是生产环境的高频踩坑点。ThreadPoolExecutor 的完整构造方法：

public ThreadPoolExecutor(
    int corePoolSize,                  // 1. 核心线程数
    int maximumPoolSize,               // 2. 最大线程数
    long keepAliveTime,                // 3. 非核心线程空闲存活时间
    TimeUnit unit,                     // 4. 存活时间单位
    BlockingQueue<Runnable> workQueue, // 5. 工作队列
    ThreadFactory threadFactory,       // 6. 线程工厂
    RejectedExecutionHandler handler   // 7. 拒绝策略
)

参数详解

1. corePoolSize（核心线程数）

线程池中始终保持存活的线程数量（即使空闲）。当任务提交时，如果当前线程数 < corePoolSize，会优先创建新线程执行任务，即使有空闲线程存在。

2. maximumPoolSize（最大线程数）

线程池允许的最大线程数。只有当工作队列满了才会创建超过 corePoolSize 的线程。

3-4. keepAliveTime + unit（空闲存活时间）

非核心线程（线程数 > corePoolSize 的部分）空闲超过这个时间后会被回收。如果调用了 allowCoreThreadTimeOut(true)，核心线程也会被回收。

5. workQueue（工作队列）

当核心线程都在忙时，新任务进入这个队列等待。队列的选择直接影响线程池的行为：

6. threadFactory（线程工厂）

自定义线程的创建方式，通常用于设置线程名、守护线程标志等。

ThreadFactory factory = r -> {
    Thread t = new Thread(r);
    t.setName("my-pool-thread-" + t.getId());
    t.setDaemon(false);
    return t;
};

7. handler（拒绝策略）

当线程池饱和（线程数达到 maximumPoolSize 且队列已满）时的处理策略：

实际生产中通常自定义拒绝策略——记录日志、发告警、写入 MQ 等。

任务提交的完整流程

                    提交任务
                       │
                       ▼
            ┌─ 线程数 < corePoolSize？─┐
            │                          │
           是                         否
            │                          │
            ▼                          ▼
       创建核心线程              ┌─ 工作队列满了？─┐
       执行任务                  │                │
                               否              是
                                │                │
                                ▼                ▼
                          任务入队等待    ┌─ 线程数 < maxPoolSize？─┐
                                        │                        │
                                       是                       否
                                        │                        │
                                        ▼                        ▼
                                   创建非核心线程            执行拒绝策略
                                   执行任务

关键注意点：

1. 先入队，后扩容。当核心线程都在忙时，新任务首先进入队列等待，而不是直接创建新线程。只有队列满了，才会创建超过 corePoolSize 的线程。
2. 如果使用无界队列（如默认的 LinkedBlockingQueue），队列永远不会满，maximumPoolSize 形同虚设。

常见线程池配置（Executors 工厂方法）

// 1. 固定大小线程池
Executors.newFixedThreadPool(10);
// 等价于 new ThreadPoolExecutor(10, 10, 0, SECONDS, new LinkedBlockingQueue<>())
// 问题：无界队列，任务堆积可能导致 OOM

// 2. 缓存线程池
Executors.newCachedThreadPool();
// 等价于 new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, SECONDS, new SynchronousQueue<>())
// 问题：最大线程数无上限，高并发时可能创建大量线程导致 OOM

// 3. 单线程池
Executors.newSingleThreadExecutor();
// 等价于 new ThreadPoolExecutor(1, 1, 0, SECONDS, new LinkedBlockingQueue<>())
// 问题：同 FixedThreadPool，无界队列

// 4. 定时线程池
Executors.newScheduledThreadPool(10);
// 基于 ScheduledThreadPoolExecutor

阿里巴巴 Java 开发手册明确禁止使用 Executors 创建线程池。原因：FixedThreadPool 和 SingleThreadPool 的队列长度为 Integer.MAX_VALUE，可能堆积大量任务导致 OOM；CachedThreadPool 的最大线程数为 Integer.MAX_VALUE，可能创建大量线程导致 OOM。

正确做法是手动创建 ThreadPoolExecutor，明确设置核心参数：

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10,                                // 核心线程数
    20,                                // 最大线程数
    60, TimeUnit.SECONDS,             // 空闲线程存活时间
    new ArrayBlockingQueue<>(200),    // 有界队列
    new ThreadFactoryBuilder()
        .setNameFormat("biz-pool-%d")
        .build(),
    new ThreadPoolExecutor.CallerRunsPolicy()  // 拒绝策略
);

线程池参数如何设置

没有通用公式，但有经验参考：

CPU 密集型任务

特点：计算量大，几乎不做 I/O。

corePoolSize = CPU 核心数 + 1

+1 是为了在某个线程因页缺失（Page Fault）等短暂等待时，多出的线程可以利用 CPU。

I/O 密集型任务

特点：大量时间花在 I/O 等待（网络请求、数据库查询、文件读写）。

corePoolSize = CPU 核心数 × 2

更精确的公式（Brian Goetz《Java Concurrency in Practice》）：

线程数 = CPU 核心数 × 目标 CPU 利用率 × (1 + W/C)

W = 线程等待时间（I/O 等待）
C = 线程计算时间（CPU 执行）

例如：一个 Web 接口，平均响应 100ms，其中 5ms 在 CPU 计算，95ms 在等数据库。在 8 核 CPU 上：

线程数 = 8 × 1.0 × (1 + 95/5) = 8 × 20 = 160

混合型任务

最佳实践是拆分为两个线程池——CPU 密集型和 I/O 密集型各用一个。

实际生产中，以上公式只是起点。真正的最优参数需要通过压测 + 监控 + 动态调整来确定。美团等公司的实践是支持线程池参数的动态修改：

// ThreadPoolExecutor 支持运行时修改参数
executor.setCorePoolSize(newCore);
executor.setMaximumPoolSize(newMax);

线程池的生命周期

                ┌──────────────┐
                │   RUNNING    │ ← 初始状态，接受任务，处理队列
                └──────┬───────┘
                       │ shutdown()
                       ▼
                ┌──────────────┐
                │   SHUTDOWN   │ ← 不接受新任务，处理队列中剩余任务
                └──────┬───────┘
                       │ 队列为空 && 线程数为 0
                       ▼
                ┌──────────────┐
                │   TIDYING    │ ← 所有任务完成
                └──────┬───────┘
                       │ terminated() 回调
                       ▼
                ┌──────────────┐
                │  TERMINATED  │ ← 终止
                └──────────────┘

或直接：
      RUNNING ──shutdownNow()──► STOP ──► TIDYING ──► TERMINATED
      STOP: 不接受新任务，中断执行中的任务，清空队列

shutdown() vs shutdownNow()

优雅停机模式：

executor.shutdown();                               // 不再接受新任务
if (!executor.awaitTermination(60, SECONDS)) {    // 等待最多 60 秒
    executor.shutdownNow();                        // 超时则强制关闭
    if (!executor.awaitTermination(60, SECONDS)) {
        System.err.println("线程池无法正常关闭");
    }
}

ForkJoinPool（JDK 7+）

ForkJoinPool 专为分治算法设计，采用**工作窃取（Work-Stealing）**算法，适合可递归分解的计算密集型任务。

核心思想

将大任务递归拆分（Fork）为小任务，各小任务并行执行后合并结果（Join）。

                    大任务
                   /      \
              子任务1      子任务2
             /    \       /    \
          小任务1 小任务2 小任务3 小任务4
             \    /       \    /
              结果1        结果2
                  \      /
                  最终结果

工作窃取算法

每个工作线程有自己的双端队列（Deque）。当自己的队列为空时，会从其他线程的队列尾部窃取任务执行。

线程 A 的队列         线程 B 的队列
┌─────────┐          ┌─────────┐
│ 任务 1  │ ← 执行   │ 任务 4  │ ← 执行
│ 任务 2  │          │         │
│ 任务 3  │ ← 窃取 ──│         │  B 队列空了，从 A 的尾部窃取
└─────────┘          └─────────┘

RecursiveTask 示例

// 计算 1 到 N 的和
public class SumTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 1000;
    private final long start, end;

    public SumTask(long start, long end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        if (end - start <= THRESHOLD) {
            // 基准条件：直接计算
            long sum = 0;
            for (long i = start; i <= end; i++) sum += i;
            return sum;
        }
        // 分治：拆分为两个子任务
        long mid = (start + end) / 2;
        SumTask left = new SumTask(start, mid);
        SumTask right = new SumTask(mid + 1, end);

        left.fork();                // 异步执行左半部分
        long rightResult = right.compute();  // 当前线程执行右半部分
        long leftResult = left.join();       // 等待左半部分完成

        return leftResult + rightResult;
    }
}

// 使用
ForkJoinPool pool = new ForkJoinPool();
long result = pool.invoke(new SumTask(1, 100_000_000));

注意：left.fork() + right.compute() 比 left.fork() + right.fork() 更高效——后者会让当前线程闲等，前者让当前线程直接执行 right 子任务。

CompletableFuture 与 ForkJoinPool

CompletableFuture 默认使用 ForkJoinPool.commonPool() 作为执行器。如果你的 CompletableFuture 链中有 I/O 操作，应该指定自定义线程池，避免阻塞公共 ForkJoinPool 的工作线程。

// 错误：I/O 操作在 commonPool 中执行
CompletableFuture.supplyAsync(() -> httpClient.get(url));

// 正确：指定 I/O 线程池
ExecutorService ioPool = Executors.newFixedThreadPool(20);
CompletableFuture.supplyAsync(() -> httpClient.get(url), ioPool);

生产环境核心踩坑点