ExecutorService 和线程池：为不同场景选择合适的池类型

1. 概览

为每个任务创建 new Thread() 很容易，但当负载增加时很难控制。ExecutorService 通过重用线程和排队任务来解决这个问题。Executors 类提供了多种不同的线程池，选择错误的类型可能会导致 OutOfMemoryError 或应用程序挂起。

本文逐一介绍了各种线程池类型、ThreadPoolExecutor 如何在内部处理任务、如何选择池大小以及如何在 Spring Boot 3.x 中配置。读完本文后，你会知道应该为 CPU-bound 任务、I/O-bound 任务或定时任务使用哪种线程池。

2. ExecutorService 基础

以下示例使用一个简单的线程池来运行一个返回结果的任务：

import java.util.concurrent.*;

public class OrderReportApp {

    public static void main(String[] args) throws Exception {
        try (ExecutorService executor = Executors.newFixedThreadPool(4)) {
            Future<Integer> total = executor.submit(() -> countOrders("2026-10"));
            System.out.println("总订单数：" + total.get());
        }
    }

    static int countOrders(String month) {
        return 128; // 模拟数据库查询
    }
}

总订单数：128

代码的工作原理如下：

• submit() 接受一个 Callable 并立即返回 Future<Integer>。任务在线程池的一个线程上运行。
• total.get() 阻塞当前线程，直到有结果。
• 从 Java 19 开始，ExecutorService 实现了 AutoCloseable。离开 try 块时，close() 调用 shutdown()，然后等待所有任务完成。

注意： 如果你不关闭 executor，非 daemon 线程会保持 JVM 活跃。main 程序将无法退出。

3. Executors 中的各种线程池类型

3.1. newFixedThreadPool(n)

该池恰好有 n 个线程。额外的任务被放入一个无限制的 LinkedBlockingQueue 中。

ExecutorService invoiceExecutor = Executors.newFixedThreadPool(8);

当你想限制同时运行的任务数时，这种类型是合适的，例如 CPU-bound 任务。风险在于队列。如果 producer 推送任务的速度快于处理速度，队列会不断膨胀，直到堆内存耗尽。

3.2. newCachedThreadPool()

该池没有固定的线程。它使用 SynchronousQueue，意味着没有真正的队列。每个新任务都分配给一个空闲线程（如果有的话），否则线程池创建一个新线程。超过 60 秒未使用的线程将被销毁。

ExecutorService notificationExecutor = Executors.newCachedThreadPool();

该池适用于许多短任务和不均匀的低负载。当负载突然增加时，它可能创建数千个平台线程。每个线程默认占用约 1 MB 的栈空间，所以应用程序容易遇到 OutOfMemoryError: unable to create native thread。

3.3. newSingleThreadExecutor()

该池只有一个线程，任务按发送顺序严格依次运行。

ExecutorService auditLogExecutor = Executors.newSingleThreadExecutor();
auditLogExecutor.submit(() -> System.out.println("订单 #1 已创建"));
auditLogExecutor.submit(() -> System.out.println("订单 #1 已支付"));

订单 #1 已创建
订单 #1 已支付

这两行日志始终按上述顺序打印。当你需要按顺序记录日志、写入文件或更新状态而不想使用锁时，这种类型很有用。如果线程因异常而死亡，线程池会创建一个新线程代替。它的队列也是无限制的。

3.4. newScheduledThreadPool(n)

这种类型用于定时或重复的任务：

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

scheduler.scheduleAtFixedRate(
        () -> System.out.println("库存同步时间：" + java.time.LocalTime.now()),
        0, 5, TimeUnit.SECONDS);

scheduleAtFixedRate 每 5 秒运行一次任务，从上一次运行开始的时间点开始计算。相反，scheduleWithFixedDelay 从上一次结束后开始计算 5 秒。当任务运行时间不稳定时，你应该选择 scheduleWithFixedDelay。

注意： 如果定时任务抛出异常，所有后续运行都会被静默取消，没有任何日志。你应该始终在 try/catch 中包装任务体。

3.5. newWorkStealingPool()

该方法返回一个 ForkJoinPool，其线程数等于 CPU 数。每个线程有自己的队列。空闲线程将从其他线程的队列中“窃取”工作。

ExecutorService pricingExecutor = Executors.newWorkStealingPool();

这种类型适用于可以递归分解的 CPU-bound 任务，例如对大型数据集进行并行计算。线程池的线程是 daemon，且线程池不保证执行顺序。

3.6. newVirtualThreadPerTaskExecutor()（Java 21）

使用这种类型，每个任务都在一个新的虚拟线程上运行。虚拟线程非常轻量级，所以你可以创建数十万个。

try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 10_000; i++) {
        int customerId = i;
        executor.submit(() -> {
            Thread.sleep(100); // 模拟对 CustomerService 的 HTTP 调用
            return customerId;
        });
    }
}

上面的代码运行 10,000 个阻塞 100 ms 的任务，并在大约几百毫秒内完成。如果使用固定的 10 个线程的线程池，相同的工作量需要大约 100 秒。

这是 I/O-bound 任务（如 HTTP 调用、数据库查询或文件读取）的一个很好的默认选择。你需要记住三件事：

• 不要池化虚拟线程。 它们被设计为一次性使用后丢弃。
• 如果要限制外部资源的负载，请使用 Semaphore。 例如，当数据库只支持 20 个连接时，不要依赖线程池大小来限制。
• 在 Java 21 中，含有阻塞操作的 synchronized 块会将虚拟线程“固定”到 carrier 线程。 这样虚拟线程的优势就消失了。在这些情况下应优先使用 ReentrantLock。

以下是使用 Semaphore 限制并发查询数的示例：

Semaphore dbPermits = new Semaphore(20);

executor.submit(() -> {
    dbPermits.acquire();
    try {
        return orderRepository.findByCustomerId(customerId);
    } finally {
        dbPermits.release();
    }
});

4. 自定义配置 ThreadPoolExecutor

Executors 中的 factory 方法通常隐藏了一个危险的参数：无限制的队列或无限制的线程数。对于生产系统，你应该自己初始化 ThreadPoolExecutor 来控制一切：

ThreadFactory factory = Thread.ofPlatform().name("payment-", 0).factory();

ThreadPoolExecutor paymentExecutor = new ThreadPoolExecutor(
        4,                                   // corePoolSize
        8,                                   // maximumPoolSize
        60, TimeUnit.SECONDS,                // 超过 core 的线程的 keepAlive 时间
        new ArrayBlockingQueue<>(100),       // 有限制的队列
        factory,                             // 设置名称：payment-0, payment-1...
        new ThreadPoolExecutor.CallerRunsPolicy());

当一个新任务到达时，ThreadPoolExecutor 按照以下顺序处理：

1. 如果线程数少于 corePoolSize，线程池创建一个新线程。
2. 否则，线程池将任务放入队列。
3. 如果队列已满且线程数少于 maximumPoolSize，线程池创建额外的线程。
4. 如果已达到两个限制，线程池调用 RejectedExecutionHandler。

其结果是 maximumPoolSize 仅当队列已满时才有作用。如果你使用无限制的队列，线程池永远不会超过 corePoolSize。这是一个非常常见的误解。

Java 提供了四种内置的拒绝策略：

策略	行为
AbortPolicy（默认）	抛出 RejectedExecutionException
CallerRunsPolicy	调用 execute() 的线程自己运行任务，帮助减缓 producer（背压）
DiscardPolicy	丢弃任务，不报告任何内容
DiscardOldestPolicy	丢弃队列中最旧的任务，然后重试

通过 ThreadFactory 给线程命名可以使线程转储和日志更易读。比较 payment-3 和 pool-7-thread-3 就能看出区别。

5. 选择正确的类型

5.1. 线程池大小

对于 CPU-bound 任务，线程数应该大约等于核心数：

int cores = Runtime.getRuntime().availableProcessors();
ExecutorService cpuExecutor = Executors.newFixedThreadPool(cores);

超过核心数的线程只会增加上下文切换的成本。

对于在平台线程上运行的 I/O-bound 任务，你可以使用一个近似公式：

线程数 ≈ 核心数 × (1 + 等待时间 / 计算时间)

例如：机器有 8 个核心，每个请求等待 I/O 90 ms，计算 10 ms。结果是 8 × (1 + 9) = 80 个线程。这只是起点。你需要通过负载测试来进行实际测量。

5.2. 决策表

问题	选择
计算密集的 CPU 任务	newFixedThreadPool(cores) 或 ForkJoinPool
大量 API 调用、数据库查询	newVirtualThreadPerTaskExecutor() + Semaphore
需要顺序执行	newSingleThreadExecutor()
定时任务	newScheduledThreadPool(n)
生产环境，需要背压	ThreadPoolExecutor 配合有限制的队列
突发负载，无限制	避免使用 newCachedThreadPool()

6. Spring Boot 3.x 中的线程池

首先，你需要 Maven 依赖项：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

然后，为发送电子邮件声明一个单独的 executor：

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean(name = "emailExecutor")
    public ThreadPoolTaskExecutor emailExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(8);
        executor.setQueueCapacity(200);
        executor.setThreadNamePrefix("email-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(30);
        return executor;
    }
}

@Service
public class EmailService {

    @Async("emailExecutor")
    public void sendOrderConfirmation(String customerEmail) {
        System.out.println(Thread.currentThread().getName() + " 发送邮件到 " + customerEmail);
    }
}

email-1 发送邮件到 an.nguyen@example.com

需要注意的几点：

• ThreadPoolTaskExecutor 的 queueCapacity 默认值是 Integer.MAX_VALUE。如果你不设置这个值，maxPoolSize 就没有意义，如第 4 节所述。
• setWaitForTasksToCompleteOnShutdown(true) 确保在应用程序关闭时不会中断正在发送的电子邮件。
• 如果你想为 @Async、Tomcat 和 @Scheduled 使用虚拟线程，你只需要添加一行配置：

spring.threads.virtual.enabled=true

7. 测试线程池行为

下面的测试确认了第 4 节中描述的正确处理顺序：先是 core 线程，然后是队列，然后是额外线程，最后是拒绝。

import org.junit.jupiter.api.Test;
import java.util.concurrent.*;
import static org.assertj.core.api.Assertions.*;

class PaymentExecutorTest {

    @Test
    void shouldRejectWhenThreadsAndQueueAreFull() throws Exception {
        CountDownLatch release = new CountDownLatch(1);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1, 2, 30, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                new ThreadPoolExecutor.AbortPolicy());

        Runnable blockingPayment = () -> {
            try {
                release.await();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };

        try {
            executor.execute(blockingPayment); // 1：创建 core 线程
            executor.execute(blockingPayment); // 2：进入队列
            executor.execute(blockingPayment); // 3：队列满 -> 创建第二个线程

            assertThat(executor.getPoolSize()).isEqualTo(2);
            assertThat(executor.getQueue()).hasSize(1);
            assertThatThrownBy(() -> executor.execute(blockingPayment))
                    .isInstanceOf(RejectedExecutionException.class);
        } finally {
            release.countDown();
            executor.shutdown();
            assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
        }
    }
}

CountDownLatch 将所有任务保持在阻塞状态，因此当你检查时线程池状态保持稳定。第四个任务没有空间，所以 AbortPolicy 抛出异常。finally 块释放闭锁并关闭线程池，所以测试不会留下任何泄漏的线程。

8. 常见错误

• 对不受控制的 producer 使用 Executors.newFixedThreadPool。 无限制的队列会膨胀并导致 OOM。应该使用有限制的 ArrayBlockingQueue。
• 在每个请求中创建一个新的池。 线程池应该被共享，例如声明为 bean 或 static final 字段。创建后忘记关闭会导致线程泄漏。
• 吞掉异常。 对于 submit()，异常被保存在 Future 中。如果你不调用 get()，你永远不会看到错误。应该调用 get() 或立即在任务中捕获错误。
• 因任务等待同一小池中的子任务而导致的死锁。 例如：2 个线程的池，两个父任务都对队列中的子任务调用 get()。没有线程可以运行子任务，所以整个系统都会挂起。
• 在虚拟线程中放置大型的 ThreadLocal。 你可能有数百万个虚拟线程，每个都持有一个副本，所以内存增长非常快。
• 调用 shutdownNow() 但任务不处理中断。 任务仍然会继续运行。长期运行的任务应该检查 Thread.currentThread().isInterrupted()。

注意： 不应该将 ExecutorService 用于复杂的多步骤相互依赖的处理流程。这种情况应该使用 CompletableFuture，它仍然可以在你自定义配置的 executor 上运行。

9. 结论

每种线程池类型都解决一个特定的问题：

• Fixed 池或 ForkJoinPool 用于 CPU-bound 任务。
• 虚拟线程加 Semaphore 用于 Java 21 上的 I/O-bound 任务。
• Single thread executor 当需要顺序执行时。
• Scheduled 池 用于定时任务。

在生产环境中，应该自己配置 ThreadPoolExecutor 或 ThreadPoolTaskExecutor，使用有限制的队列、明确的拒绝策略和易识别的线程名称。最后，你应该通过负载测试来验证线程池大小，而不是仅依赖公式。