Java 线程池

2022-05-02

4 min read

Java 线程池是并发编程中最核心的工具之一。本文从线程池初始化入手，逐步深入其原理，再到 Java 8 Stream 如何利用多线程，最后解决 ThreadLocal 在多线程环境下的失效问题并提供可落地的工具方案。

线程池的初始化

正确的方式：ThreadPoolExecutor

Java 线程池的核心类是 ThreadPoolExecutor 。阿里巴巴规约明确要求 必须通过 ThreadPoolExecutor 显式创建线程池 ，禁止使用 Executors 工厂方法。

七个核心参数

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

参数	含义	常见取值
corePoolSize	常驻线程数，即使空闲也不回收	CPU 核数 ~ CPU 核数*2
maximumPoolSize	最大线程数，队列满后才创建新线程	corePoolSize * 2~4
keepAliveTime	超过 corePoolSize 的线程空闲多久被回收	60s
workQueue	存放等待执行的任务	LinkedBlockingQueue / ArrayBlockingQueue
threadFactory	创建线程的工厂，便于统一命名	自定义，设置线程名前缀
handler	线程池满且队列满时的处理策略	AbortPolicy / CallerRunsPolicy

四种拒绝策略

策略	行为
AbortPolicy	抛 RejectedExecutionException（默认）
CallerRunsPolicy	由调用者线程直接执行任务
DiscardPolicy	直接丢弃新任务，不抛异常
DiscardOldestPolicy	丢弃队列中最老的任务，然后重试提交

初始化示例

生产环境推荐配置

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadPoolFactory {

    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();

    public static ThreadPoolExecutor createFixedThreadPool() {
        return new ThreadPoolExecutor(
            CPU_COUNT,
            CPU_COUNT * 2,
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000),
            new ThreadFactory() {
                private final AtomicInteger counter = new AtomicInteger(1);
                @Override
                public Thread newThread(Runnable r) {
                    Thread t = new Thread(r, "biz-pool-" + counter.getAndIncrement());
                    t.setDaemon(false);
                    return t;
                }
            },
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }
}

核心线程池参数也可以由 Spring 管理

@Configuration
public class ThreadPoolConfig {

    @Bean("taskExecutor")
    public ThreadPoolExecutor taskExecutor() {
        return new ThreadPoolExecutor(
            4, 8, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(500),
            new ThreadFactoryBuilder().setNameFormat("task-%d").build(),
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }
}

线程池原理

线程池的生命周期

线程池内部维护了一个 AtomicInteger ctl ，高 3 位表示整个线程池的运行状态，低 29 位表示线程数量。

状态	含义	触发条件
RUNNING	接受新任务，处理队列任务	初始化后
SHUTDOWN	不接受新任务，处理队列中剩余任务	shutdown()
STOP	不接受新任务，不处理队列任务，中断正在执行的任务	shutdownNow()
TIDYING	所有任务已终止，线程数为 0，即将调用 terminated()	过渡状态
TERMINATED	terminated() 执行完毕	终态

实际上，线程池的 5 种状态，本质上描述的是线程池“关闭”的生命周期。
kill -15 <pid> 是 Linux/Unix 的“优雅关闭”命令。 Spring Boot 应用程序可以捕获这个信号，并做善后工作。

任务执行流程
1. 当前线程数 < corePoolSize → 直接创建新线程执行任务
2. 当前线程数 >= corePoolSize → 将任务放入 workQueue 排队
3. workQueue 已满 && 线程数 < maximumPoolSize → 创建非核心线程执行任务
4. workQueue 已满 && 线程数 >= maximumPoolSize → 触发拒绝策略
线程复用的核心：Worker

线程池中的线程被封装为 Worker 对象。Worker 本质是一个 AQS 同步器*，同时实现了 *Runnable 接口。
Worker 启动后，通过 runWorker() 方法不断从队列中 poll 或 take 任务来执行，从而实现线程复用。
```
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    while (task != null || (task = getTask()) != null) {
        // 执行 task.run()
        task.run();
        task = null;
    }
}
```

Java 8 Stream 如何利用多线程

并行流

Java 8 提供了 parallelStream() 和 parallel() 方法，可以将串行流转为并行流，底层使用 *ForkJoinPool.commonPool()*：

ForkJoinPool 是 JDK 提供的面向 CPU 密集型任务的线程池，通过工作窃取（Work-Stealing）算法提高多核 CPU 的利用率。
普通线程池共享一个任务队列，而 ForkJoinPool 为每个工作线程维护一个任务队列，空闲线程会主动窃取其他线程队列中的任务。
```
// 方式一：parallelStream()
List<Integer> result = list.parallelStream()
    .map(x -> x * 2)
    .collect(Collectors.toList());

// 方式二：stream().parallel()
List<Integer> result = list.stream()
    .parallel()
    .map(x -> x * 2)
    .collect(Collectors.toList());
```

并行流 vs 线程池

场景	推荐方案
CPU 密集型计算（纯计算，无 IO）	parallelStream
IO 密集型（数据库、网络调用）	自定义 ThreadPoolExecutor
需要精细控制并发数	自定义线程池
简单集合的批量处理	parallelStream

ThreadLocal 多线程失效及解决方案

问题：ThreadLocal 在多线程中失效

ThreadLocal 通过线程本地变量实现数据隔离，但这一特性在跨线程时 *完全失效*。当任务提交到线程池执行时，子线程无法获取父线程的 ThreadLocal 值。

public class ThreadLocalExample {
    private static final ThreadLocal<String> USER_CONTEXT = new ThreadLocal<>();
    private static final ExecutorService EXECUTOR = Executors.newFixedThreadPool(2);

    public static void main(String[] args) {
        USER_CONTEXT.set("admin");

        EXECUTOR.submit(() -> {
            String user = USER_CONTEXT.get(); // 结果为 null！
            System.out.println("当前用户: " + user);
        });
    }
}

解决方案一：InheritableThreadLocal

InheritableThreadLocal 在 创建子线程时 会将父线程的值拷贝给子线程。但 线程池会复用线程 ，只会在线程第一次创建时拷贝，后续复用不会更新。
```
private static final InheritableThreadLocal<String> CONTEXT = new InheritableThreadLocal<>();
```

解决方案二：TransmittableThreadLocal

阿里巴巴开源的 transmittable-thread-local 解决了线程池复用场景下的上下文传递问题。它通过 在执行任务前捕获、执行后恢复 的机制，确保每次提交任务时都能正确传递 ThreadLocal 值。

引入依赖

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>transmittable-thread-local</artifactId>
    <version>2.14.4</version>
</dependency>

基础使用

import com.alibaba.ttl.TransmittableThreadLocal;
import com.alibaba.ttl.TtlRunnable;

public class TtlDemo {
    private static final TransmittableThreadLocal<String> CONTEXT = new TransmittableThreadLocal<>();
    private static final ExecutorService EXECUTOR = Executors.newFixedThreadPool(2);

    public static void main(String[] args) {
        CONTEXT.set("admin");

        // 使用 TtlRunnable 包装任务
        EXECUTOR.submit(TtlRunnable.get(() -> {
            String user = CONTEXT.get(); // 结果: "admin"
            System.out.println("当前用户: " + user);
        }));
    }
}

装饰线程池（推荐）

每次 TtlRunnable.get() 是侵入性的。更好的方式是使用 TtlExecutors 装饰整个线程池：

ExecutorService executor = Executors.newFixedThreadPool(4);
executor = TtlExecutors.getTtlExecutorService(executor);

// 之后所有提交的任务自动传递 ThreadLocal 值
executor.submit(() -> {
    String user = CONTEXT.get(); // 正常获取
});

方案的对比

方案	优点	缺点	推荐场景
InheritableThreadLocal	JDK 原生，零依赖	线程池复用时不生效	不需要线程池的简单场景
TransmittableThreadLocal	功能完善，生态成熟	需要引入第三方依赖	生产环境首选

Java 线程池

线程池的初始化

线程池原理

Java 8 Stream 如何利用多线程

ThreadLocal 多线程失效及解决方案

On This Page

See Also