Java多线程（六）：线程池与自定义ThreadPoolExecutor

前言

前面的案例都直接创建 Thread。这种方式适合学习线程机制，也适合数量很少、生命周期清楚的临时任务。但服务器可能在短时间内接收成千上万个请求，如果每个请求都创建一个线程，创建和销毁开销会不断累积，线程数量失控还会耗尽内存并造成大量上下文切换。

线程池的思想是提前准备并管理一组工作线程，任务到来时把任务提交给线程池，工作线程执行完任务后不立即销毁，而是继续处理后续任务。本篇介绍线程池的基本使用、任务提交方式、自定义 ThreadPoolExecutor 的七个核心参数、任务处理流程和拒绝策略。

一、为什么需要线程池

直接创建线程主要存在以下问题：

1. 创建和销毁有成本

线程需要分配栈空间和操作系统资源，频繁创建、销毁会产生额外开销。对于大量执行时间很短的任务，管理线程的成本甚至可能接近任务本身。

2. 数量难以控制

如果请求一来就 new Thread(...).start()，请求峰值可能迅速转化为线程峰值。过多线程会消耗内存，并让 CPU 花费大量时间切换上下文。

3. 缺少统一管理

手动创建的线程不方便统一命名、监控、关闭、限制并发量和处理异常。线程池把这些能力集中到一个组件中。

线程池的主要价值可以概括为：复用线程、控制并发、管理任务和线程生命周期。

二、ExecutorService 的基本使用

ExecutorService 是常用线程池服务接口。学习阶段可以通过 Executors 快速创建固定大小的线程池：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class FixedPoolDemo {
    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(3);

        try {
            for (int i = 1; i <= 10; i++) {
                int taskId = i;
                pool.submit(() -> {
                    System.out.println(
                            Thread.currentThread().getName()
                                    + " 正在执行任务 " + taskId
                    );
                });
            }
        } finally {
            pool.shutdown();
        }
    }
}

这里只创建 3 个工作线程，却可以处理 10 个任务。执行完某个任务的线程会继续从队列中获取下一个任务。

线程池使用完后要关闭：

• shutdown()：不再接收新任务，但已经提交的任务会继续执行。
• shutdownNow()：尝试中断正在执行的任务，并返回尚未开始的任务；它是“尝试停止”，无法保证任务一定立即结束。

如果任务代码完全忽略中断，shutdownNow() 也不能强制安全终止它。因此，任务本身要有合理的中断响应策略。

三、execute 与 submit

线程池常见的任务提交方式有两种。

1. execute(Runnable)

execute() 定义在 Executor 接口中，只接收 Runnable，没有任务结果返回。

pool.execute(() -> System.out.println("执行无返回值任务"));

2. submit(...)

submit() 定义在 ExecutorService 中，可以提交 Runnable 或 Callable，并返回 Future。

import java.util.concurrent.Future;

Future<Integer> future = pool.submit(() -> {
    int sum = 0;
    for (int i = 1; i <= 100; i++) {
        sum += i;
    }
    return sum;
});

System.out.println(future.get());

Future 可以查询任务是否完成、等待并获取结果、尝试取消任务。get() 仍然可能阻塞，所以主线程应在确实需要结果的位置调用，而不是任务刚提交就立刻逐个 get()，否则可能把本可并行的任务写成近似串行。

异常表现也有差异：通过 submit() 提交的任务如果抛出异常，异常会被保存到 Future 中，调用 get() 时以 ExecutionException 形式暴露。如果完全不读取 Future，异常可能没有预期中的控制台表现。因此，任务异常监控需要主动设计。

四、Executors 工厂方法的特点

Executors 提供多个便捷工厂：

• newFixedThreadPool(n)：固定数量工作线程，额外任务进入共享队列。
• newSingleThreadExecutor()：只有一个工作线程，任务按顺序执行。
• newCachedThreadPool()：根据需要创建和复用线程，空闲线程一定时间后回收。

它们很适合教学和简单工具，但隐藏了队列容量、最大线程数、拒绝策略等关键配置。例如固定线程池使用的队列容量非常大，任务生产速度长期超过消费速度时，任务可能持续堆积；缓存线程池允许创建非常多的线程，突发负载下可能造成资源压力。

因此，生产系统更常直接创建 ThreadPoolExecutor，显式写出容量和策略，使资源上限可见、可评估。不能机械地说所有 Executors 工厂都“绝对不能用”，关键是理解它们的内部配置是否符合业务边界。

五、ThreadPoolExecutor 的七个核心参数

最完整的常用构造方法如下：

public ThreadPoolExecutor(
        int corePoolSize,
        int maximumPoolSize,
        long keepAliveTime,
        TimeUnit unit,
        BlockingQueue<Runnable> workQueue,
        ThreadFactory threadFactory,
        RejectedExecutionHandler handler
)

1. corePoolSize：核心线程数

线程池希望长期维持的基础工作线程数量。默认情况下，即使核心线程暂时没有任务，也不会因为 keepAliveTime 到期而销毁。

2. maximumPoolSize：最大线程数

线程池能够容纳的工作线程总上限，包括核心线程和临时扩展线程。它必须大于等于核心线程数。

3. keepAliveTime：空闲存活时间

当线程数量超过核心线程数时，多出的线程如果空闲超过该时间，通常会被回收。

4. unit：时间单位

指定 keepAliveTime 的单位，例如 TimeUnit.SECONDS、TimeUnit.MILLISECONDS。

5. workQueue：任务队列

核心线程都忙时，新任务会先尝试进入任务队列。常见选择有 ArrayBlockingQueue 和 LinkedBlockingQueue。生产环境通常需要明确评估队列容量，避免任务无界堆积。

6. threadFactory：线程工厂

负责创建工作线程。可以统一设置线程名称、是否为守护线程、未捕获异常处理器等。清晰的线程名对日志定位和线程转储分析非常有价值。

7. handler：拒绝策略

当线程数已经达到最大值，任务队列也已满时，线程池通过拒绝策略处理新任务。拒绝不是线程池“坏了”，而是资源边界被触发后的必然决策。

六、线程池接收任务的完整流程

理解任务进入顺序非常重要。假设核心线程数为 2，最大线程数为 4，队列容量为 3：

1. 当前工作线程少于 2：优先创建核心线程执行任务。
2. 核心线程都在忙：尝试把新任务放入队列。
3. 队列已满，当前线程数少于 4：再创建非核心线程执行新任务。
4. 线程数达到 4 且队列也满：执行拒绝策略。

很多初学者误以为任务到来后，线程池会先一路创建到最大线程数，再使用队列。实际顺序是：核心线程 → 队列 → 扩展到最大线程数 → 拒绝。

按照上面的配置，如果长任务持续占用线程：

• 前 2 个任务创建核心线程。
• 接下来的 3 个任务进入队列。
• 再来的 2 个任务推动线程数从 2 增长到 4。
• 第 8 个仍无法接纳的任务触发拒绝策略。

任务完成后，工作线程会继续从队列取任务。超过核心数量的线程空闲达到 keepAliveTime 后可以被回收。

七、自定义线程池完整示例

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class CustomPoolDemo {
    public static void main(String[] args) {
        AtomicInteger threadNumber = new AtomicInteger(1);

        ThreadFactory factory = task -> {
            Thread thread = new Thread(task);
            thread.setName(
                    "order-worker-" + threadNumber.getAndIncrement()
            );
            thread.setUncaughtExceptionHandler((t, error) ->
                    System.err.println(t.getName()
                            + " 未捕获异常：" + error.getMessage())
            );
            return thread;
        };

        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                2,
                4,
                30,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(3),
                factory,
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        try {
            for (int i = 1; i <= 10; i++) {
                int taskId = i;
                pool.execute(() -> processOrder(taskId));
            }
        } finally {
            pool.shutdown();
        }
    }

    private static void processOrder(int taskId) {
        System.out.println(
                Thread.currentThread().getName()
                        + " 开始处理订单 " + taskId
        );
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

此处选择有界队列，让任务最大积压量明确；线程工厂统一命名，便于诊断；拒绝策略使用 CallerRunsPolicy，让提交任务的线程自己执行无法接纳的任务。提交方因此会变慢，形成一种简单的反压效果。

但这不代表 CallerRunsPolicy 适合所有业务。如果提交线程是 UI 线程，直接执行耗时任务会卡住界面；如果提交线程是 Web 请求线程，也可能显著增加请求延迟。拒绝策略必须结合业务上下文选择。

八、四种内置拒绝策略

ThreadPoolExecutor 提供四种常见策略：

1. AbortPolicy

默认策略，拒绝时抛出 RejectedExecutionException。优点是失败显式，不会悄悄丢任务；调用方必须决定重试、降级或返回错误。

2. CallerRunsPolicy

由提交任务的线程直接执行任务，前提是线程池还没有关闭。它能降低提交速度，但也会把执行压力传递给调用方。

3. DiscardPolicy

直接丢弃新任务，不抛异常。除非业务明确允许丢失，并且已经有完善监控，否则风险较高。

4. DiscardOldestPolicy

丢弃队列中等待时间最久的任务，再尝试提交当前任务。它可能让旧请求永久失去处理机会，也需要非常明确的业务依据。

实际项目还可以实现自定义 RejectedExecutionHandler，例如记录指标、写入持久化队列、执行有限次数重试或触发降级。需要避免在拒绝处理器中无上限地阻塞或递归重试，否则只是把过载问题转移到了别处。

九、线程池大小怎样设置

不存在对所有程序都适用的固定公式，但可以从任务性质出发估算，再用压测和监控调整。

CPU 密集型任务

任务主要消耗 CPU，例如编码、压缩、复杂计算。线程数通常接近 CPU 核心数，过多线程只会增加切换开销。

I/O 密集型任务

任务大量时间在等待网络、数据库或磁盘，线程等待时 CPU 可以处理其他任务，因此线程数可以比 CPU 核心数更多。

更一般的估算思路是：

线程数 ≈ CPU 核心数 × 目标利用率 × (1 + 等待时间 / 计算时间)

这只是初始估算。最终配置还受下游数据库连接数、接口限流、内存、任务延迟目标和峰值流量影响。线程池不能凭空提高下游容量；如果数据库只有 20 个连接，把线程数调到 200 可能只会制造更多等待。

队列容量同样不能随意取一个很大的数。队列越大，越能吸收短时突发，但过载时任务等待时间也越长，还会占用更多内存。线程数、队列容量、超时和拒绝策略需要作为一个整体设计。

十、线程池的异常与监控

要让线程池可维护，至少应关注：

• 当前线程数与最大线程数。
• 活跃线程数。
• 队列当前长度与剩余容量。
• 已完成任务数。
• 拒绝任务次数。
• 任务等待时间与执行时间。
• 任务异常数量。

ThreadPoolExecutor 提供 getPoolSize()、getActiveCount()、getQueue()、getCompletedTaskCount() 等方法，可用于观测。生产环境更适合接入指标系统，而不是高频打印日志。

还要区分 execute() 和 submit() 的异常路径。自定义线程工厂的未捕获异常处理器有助于捕获 execute() 任务的未处理异常；submit() 会把异常包装进 Future，应通过 get()、任务包装器或重写 afterExecute() 等方式统一记录。

十一、常见误区

误区一：线程池越大，处理速度越快

线程过多会增加内存和上下文切换开销，还可能把数据库、远程接口等下游资源压垮。

误区二：队列越大越安全

超大队列可能把“立即拒绝”变成“长时间排队后超时”，同时增加内存风险。容量应该根据允许的等待时间和峰值任务量评估。

误区三：shutdown 会丢掉所有任务

shutdown() 会停止接收新任务，但继续执行已提交任务。尝试更快停止的是 shutdownNow()，它也不保证正在运行的任务立即结束。

误区四：提交成功就代表业务成功

提交成功只表示线程池接纳了任务。任务可能稍后执行失败、被取消，甚至因进程退出而没来得及完成。需要可靠交付的任务还要结合持久化、重试、幂等和状态记录。

十二、本篇小结

线程池通过复用工作线程、限制线程数量和统一管理任务，避免为每个短任务反复创建线程。ExecutorService 提供任务提交、结果获取和关闭能力，ThreadPoolExecutor 则让核心线程数、最大线程数、空闲时间、队列、线程工厂和拒绝策略都显式可控。

必须记住任务处理顺序：先创建核心线程，核心线程忙时任务进入队列，队列满后才扩展到最大线程数，最后触发拒绝策略。线程数和队列容量没有万能答案，需要结合任务的 CPU/I/O 比例、下游容量、延迟目标，通过压测和指标持续调整。

至此，我们完成了从线程创建、调度与状态，到线程安全、等待唤醒、综合练习和线程池的完整学习路线。真正掌握多线程的标志，不只是记住 API，而是能够识别共享状态、建立清晰的不变式、设计停止方式，并用测试和监控验证程序在各种调度顺序下仍然正确。

参考资料

• 课程视频：BV1LG4y1T7n2
• Oracle：ExecutorService
• Oracle：Executors
• Oracle：ThreadPoolExecutor