深入理解 `ThreadPoolExecutor`：自定义线程池的艺术

奋斗的软件工程师 / 2024-09-18 / 原文

深入理解 `ThreadPoolExecutor`：自定义线程池的艺术

在现代软件开发中，处理并发任务是提升性能、优化资源利用的关键策略之一。而线程池是这一策略中的重要工具。特别是在 Java 中，ThreadPoolExecutor 是一个高度灵活且功能强大的类，它允许开发者根据应用场景定制线程池的行为。本篇文章将从自定义线程池的角度，详细介绍 ThreadPoolExecutor 的参数、使用场景和实际意义。

1. 线程池的基本概念与动机

我们可以想象一个普通的餐厅，当客人点餐时，餐厅老板会指派服务员处理订单。然而，当客人数量突然增加时，如果每个客人都要求一个新服务员，那么很快就会有太多服务员，餐厅可能因为人手不足或空间限制而乱成一团。

在计算机系统中，线程就像服务员，每个任务就是客人的订单。如果每次有任务就创建一个线程，会浪费大量资源（例如内存、CPU 时间），且线程的创建和销毁也是昂贵的操作。线程池就像是餐厅中的一个管理策略：只维持一定数量的服务员（线程），超出的任务可以排队等候处理，避免资源的浪费与系统的崩溃。

2. 认识 `ThreadPoolExecutor`

Java 提供的 ThreadPoolExecutor 类，类似于一个“线程池管理者”，可以灵活定制线程池的行为，保证资源利用效率，并控制任务的执行速度。其构造函数接受一系列参数，每个参数都决定了线程池的不同方面。下面我们详细讲解这些参数。

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

每个参数都有其特定的作用，我们通过以下几个问题来形象解释它们的意义。

3. 逐一解析 `ThreadPoolExecutor` 的构造参数

corePoolSize（核心线程数）：

可以想象一个小餐厅的日常运营中有一组“核心”服务员，他们一直待命，即使店里一时没有客人，他们也在准备随时响应订单。corePoolSize 就是这样的核心线程数：这是线程池中保持活跃的最少线程数量。即便这些线程当前没有任务执行，它们也不会被销毁。

举例：假设你有一个系统，它需要持续处理用户的实时请求。你可以将核心线程数设置为根据你服务器的能力（如 CPU 核数）来优化响应速度。
maximumPoolSize（最大线程数）：

当餐厅的客人突然增多时，额外的服务员可能被临时雇佣来处理短时间内增加的需求。但这些临时服务员只在客人多的时候工作，客人少了就不需要他们了。maximumPoolSize 就是线程池允许的最大线程数。如果当前的任务超过了核心线程数，线程池会创建额外的线程（最多创建到 maximumPoolSize）。

举例：在流量高峰期，系统可能需要临时创建更多的线程来处理额外的任务负载，比如电商网站的秒杀活动。
keepAliveTime（线程存活时间）：

对于那些在高峰期临时创建的服务员，当他们闲下来的时候，如果长时间没有客人点单，他们就会被解雇离开。keepAliveTime 就是这样的等待时间：当线程池中超过核心线程数的线程闲置超过这个时间，它们就会被回收掉，释放系统资源。

举例：在非高峰时段，系统不需要大量的线程，可以通过 keepAliveTime 设置线程池回收那些不再需要的线程，从而节省资源。
unit（时间单位）：

这个参数决定了 keepAliveTime 的时间单位。可以是秒、毫秒、甚至微秒等。就像你可以用分钟或小时来衡量服务员的工作时间，unit 确保了 keepAliveTime 的精确度。
workQueue（任务队列）：

当所有核心线程都在忙碌时，新的客人（任务）可以选择等在队列里，直到有服务员可以接单。这就是任务队列的作用。任务队列有不同的类型：
- SynchronousQueue：像一个立即传递的队列，必须有线程可以马上处理任务，否则就会创建新线程。
- LinkedBlockingQueue：像一个无界队列，可以存储无限多的任务，适合任务量大但处理时间短的场景。
- ArrayBlockingQueue：一个固定大小的队列，超过队列容量的任务就会被拒绝或按照拒绝策略处理。
举例：假设你有一个任务处理系统，某些任务需要排队等待处理，你可以使用 LinkedBlockingQueue 或 ArrayBlockingQueue 控制任务的排队逻辑。
threadFactory（线程工厂）：

ThreadFactory 类似于招聘服务员的工厂。你可以通过自定义 ThreadFactory 来定义线程的属性，比如线程的名字、优先级，甚至是否是守护线程。

举例：如果你需要为每个线程指定一个有意义的名字（如 "订单处理线程-1"），你可以自定义 ThreadFactory。
handler（拒绝策略）：

当餐厅满员时，新来的客人无法被接待，怎么办？有几种选择：
- AbortPolicy：像一个保安，把新来的客人直接赶出去——抛出 RejectedExecutionException 异常。
- CallerRunsPolicy：让客人自己动手解决——提交任务的线程自己执行该任务。
- DiscardPolicy：直接忽略新任务，不做任何处理。
- DiscardOldestPolicy：把最老的客人（任务）丢出去，接待新的客人。
举例：在一个消息队列系统中，可能不允许任何任务丢失，因此你可以选择合适的拒绝策略来确保任务的处理。

4. 自定义线程池的应用场景

根据不同的应用场景，我们可以调整 ThreadPoolExecutor 的参数配置，以获得最佳的性能和资源利用率。下面列举几个常见的应用场景：

场景 1：高并发处理
在线购物网站在促销期间经常面临大量并发请求，例如抢购秒杀。当系统无法同时处理所有请求时，必须通过线程池限制并发线程数，避免服务器崩溃。这里可以将 maximumPoolSize 设置得较大，以应对高并发压力，并选择 CallerRunsPolicy 作为拒绝策略，确保任务至少会被执行。
场景 2：后台批量任务处理
某些任务（如图片处理或数据同步）需要长时间运行，但又不能占用过多资源。可以将 corePoolSize 设置为较小值，利用 ArrayBlockingQueue 控制任务队列的长度，避免任务过多导致系统负载过高。
场景 3：CPU 密集型任务
如果任务是 CPU 密集型的（如加密解密、复杂计算），应根据 CPU 核心数限制线程池中的线程数，以免线程之间争夺 CPU 资源导致整体性能下降。可以将 maximumPoolSize 设置为 CPU 核心数 + 1。

5. 实战代码示例

以下是一个使用 ThreadPoolExecutor 创建自定义线程池的代码示例，展示了如何配置线程池并处理多任务的执行。

import java.util.concurrent.*;

public class CustomThreadPoolExample {
    public static void main(String[] args) {
        // 创建自定义线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2, // 核心线程数
            4, // 最大线程数
            60, // 超时时间
            TimeUnit.SECONDS, // 时间单位
            new ArrayBlockingQueue<>(2), // 有界任务队列
            Executors.defaultThreadFactory(), // 默认线程工厂
            new ThreadPoolExecutor.AbortPolicy() // 拒绝策略：抛出异常
        );

        // 提交6个任务
        for (int i = 0

; i < 6; i++) {
            executor.submit(() -> {
                System.out.println(Thread.currentThread().getName() + " 正在执行任务");
                try {
                    Thread.sleep(2000); // 模拟任务执行时间
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        // 关闭线程池
        executor.shutdown();
    }
}