自定义线程池 - ExecutorBuilder

简介

在 JDK 中，Executors 提供了创建自定义线程池对象 ExecutorService 的方法，但对于灵活的线程管理策略，单独使用 Executors 无法满足需求。ExecutorBuilder 通过组合不同的配置项来实现更复杂的线程管理策略，适用于多样化的线程池需求。

1. 使用场景

• 异步任务处理：在高并发场景下，需要通过自定义线程池来控制任务的执行数量和等待队列。
• 微服务调用并发控制：微服务调用中，需要限定并发请求数量和任务的最大等待队列。
• 大规模数据处理：在批量数据处理、任务分发等场景中，通过自定义线程池可以有效提升系统的处理能力。

2. 主要概念与配置

2.1 核心配置

• corePoolSize：初始池大小，表示核心线程数，即在没有任务时线程池维持的最小线程数量。
• maxPoolSize：最大池大小，表示允许同时执行的最大线程数。
• workQueue：任务队列，用于存储未执行的任务。当所有核心线程都在忙碌时，新的任务会被放入此队列等待执行。
• handler：异常处理器（RejectedExecutionHandler），当线程池和等待队列都已满且无法处理新任务时，触发异常策略。

2.2 线程池的工作流程

1. 如果池中任务数 < corePoolSize，则立即创建新线程执行任务。
2. 如果池中任务数 > corePoolSize，任务进入等待队列。
3. 当队列已满且池中任务数 < maxPoolSize 时，创建新线程立即执行任务。
4. 如果执行中的线程数 > maxPoolSize，则触发异常处理策略。

2.3 常见工作队列策略

• SynchronousQueue：不保存任务，任务直接提交给线程。当运行线程小于 maxPoolSize 时会创建新线程，否则触发异常策略。
• LinkedBlockingQueue：默认无界队列，当运行线程数超过 corePoolSize 时，任务放入此队列。设置队列大小后，队列满时创建新线程，否则触发异常策略。
• ArrayBlockingQueue：有界队列，适合控制队列大小。当队列满时，创建新线程，若线程数超过 maxPoolSize，则触发异常策略。

3. 主要方法与示例

3.1 创建默认线程池

默认策略：

• 初始线程数为 corePoolSize 指定的大小。
• 没有最大线程数限制。
• 使用 LinkedBlockingQueue，默认队列大小为 1024。
• 当线程数大于 corePoolSize 时，任务进入队列，队列满后抛出异常。

示例：创建默认线程池

import cn.hutool.core.thread.ExecutorBuilder;

import java.util.concurrent.ExecutorService;

public class DefaultExecutorExample {
    public static void main(String[] args) {
        // 创建默认线程池
        ExecutorService executor = ExecutorBuilder.create().build();

        // 提交任务
        executor.execute(() -> System.out.println("任务执行中..."));

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

3.2 创建单线程线程池

配置策略：

• 初始线程数为 1。
• 最大线程数为 1。
• 使用 LinkedBlockingQueue，默认队列大小为 1024。
• 同时只允许一个线程工作，其他任务进入队列，等待队列满时抛出异常。

示例：创建单线程线程池

import cn.hutool.core.thread.ExecutorBuilder;

import java.util.concurrent.ExecutorService;

public class SingleThreadExecutorExample {
    public static void main(String[] args) {
        // 创建单线程线程池
        ExecutorService executor = ExecutorBuilder.create()
            .setCorePoolSize(1)
            .setMaxPoolSize(1)
            .setKeepAliveTime(0)
            .build();

        // 提交任务
        for (int i = 0; i < 5; i++) {
            int taskId = i;
            executor.execute(() -> System.out.println("执行任务: " + taskId));
        }

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

3.3 创建自定义线程池

配置策略：

• 初始 5 个线程。
• 最大 10 个线程。
• 使用有界等待队列，最大等待数为 100。

示例：创建自定义线程池

import cn.hutool.core.thread.ExecutorBuilder;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;

public class CustomExecutorExample {
    public static void main(String[] args) {
        // 创建自定义线程池
        ExecutorService executor = ExecutorBuilder.create()
            .setCorePoolSize(5)
            .setMaxPoolSize(10)
            .setWorkQueue(new LinkedBlockingQueue<>(100))
            .build();

        // 提交任务
        for (int i = 0; i < 20; i++) {
            int taskId = i;
            executor.execute(() -> System.out.println("执行任务: " + taskId));
        }

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

3.4 创建特殊策略线程池

配置策略：

• 初始 5 个线程。
• 最大 10 个线程。
• 使用 SynchronousQueue，任务直接提交给线程，当线程数小于 maxPoolSize 时创建新线程，否则触发异常策略。

示例：创建特殊策略线程池

import cn.hutool.core.thread.ExecutorBuilder;

import java.util.concurrent.ExecutorService;

public class SpecialStrategyExecutorExample {
    public static void main(String[] args) {
        // 创建特殊策略线程池
        ExecutorService executor = ExecutorBuilder.create()
            .setCorePoolSize(5)
            .setMaxPoolSize(10)
            .useSynchronousQueue()
            .build();

        // 提交任务
        for (int i = 0; i < 15; i++) {
            int taskId = i;
            executor.execute(() -> System.out.println("执行任务: " + taskId));
        }

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

4. 实际应用场景

• 并发任务处理：在需要并行处理大量任务时，通过自定义线程池控制并发数量和任务队列。
• 异步服务调用：在微服务架构中，通过自定义线程池优化异步服务调用的并发和等待队列。
• 资源受限场景：在系统资源有限的场景下，通过合理配置线程池大小和队列策略来提高处理能力。

5. 注意事项

• 在线程池中设置合适的 corePoolSize 和 maxPoolSize，避免因资源占用过多导致系统性能下降。
• 使用有界队列时，需要考虑队列满后的处理策略，确保系统的健壮性。
• 在配置 handler 时，可以根据业务需求选择合适的异常处理策略，如抛出异常、丢弃任务或重试。