Java ThreadPoolExecutor的拒绝策略与自定义

在多线程编程中，ThreadPoolExecutor 是Java线程池的核心实现。当提交给线程池的任务数量超过其最大容量（即队列已满且线程数达到 maximumPoolSize）时，线程池会采取一个拒绝策略来处理新提交的任务。理解并正确配置这个策略，是构建健壮并发应用的关键。

1. 为什么需要拒绝策略

线程池的资源（线程数、工作队列容量）是有限的。在高并发场景下，如果任务提交速度持续超过线程池的处理能力，且不采取任何措施，会导致任务无限堆积，最终耗尽内存，引发 OutOfMemoryError。

拒绝策略的本质是一种过载保护机制。它明确告知调用者：“当前系统已满负荷，无法接纳更多任务。” 这为系统提供了明确的行为预期和稳定性保障。

2. 四种内置拒绝策略详解

Java在 java.util.concurrent 包中内置了四种标准的拒绝策略。它们是 RejectedExecutionHandler 接口的实现。

1. AbortPolicy (默认策略)

   • 行为：直接抛出 RejectedExecutionException 运行时异常。
   • 适用场景：这是默认行为，适用于必须明确知道任务被拒绝的场景。它迫使调用者处理异常，通常用于那些对任务丢失零容忍的业务逻辑。
   • 效果：任务提交失败，程序可能中断（如果不捕获异常）。

2. CallerRunsPolicy

   • 行为：不抛弃任务，也不抛出异常。而是将任务回退给调用者线程（即提交任务的那个线程）来执行。
   • 适用场景：这是一种有效的限流手段。它既能保证所有提交的任务最终都会被执行，又能通过占用调用者线程的时间，自动降低任务提交速度，使系统负载趋于平稳。
   • 效果：任务被提交线程执行，起到了背压（back-pressure）作用。

3. DiscardPolicy

   • 行为：静默丢弃被拒绝的任务，不抛出任何异常。
   • 适用场景：适用于那些允许丢失的、非关键性的后台任务。例如，某些日志收集或监控上报任务。
   • 效果：任务被默默丢弃，提交方完全无感知。

4. DiscardOldestPolicy

   • 行为：丢弃队列中最早（即将被执行）的一个任务，然后重新尝试提交当前任务。
   • 适用场景：适用于希望保留最新任务的场景，例如实时数据更新，旧的数据更新可能不如新数据重要。
   • 效果：总是尝试执行最新的任务，但可能造成重要任务的丢失。

3. 如何配置与使用内置策略

在创建 ThreadPoolExecutor 实例时，通过构造函数的最后一个参数指定拒绝策略。

import java.util.concurrent.*;

public class RejectionPolicyDemo {
    public static void main(String[] args) {
        // 创建一个核心线程为1，最大线程为1，队列容量为2的线程池
        // 使用 CallerRunsPolicy 作为拒绝策略
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1,  // corePoolSize
                1,  // maximumPoolSize
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(2), // 工作队列，容量为2
                new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
        );

        // 模拟提交5个任务，超过了线程池容量 (1线程 + 2队列 = 3)
        for (int i = 1; i <= 5; i++) {
            final int taskNum = i;
            executor.submit(() -> {
                System.out.println("正在执行任务: " + taskNum + "，线程: " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000); // 模拟任务耗时
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        executor.shutdown();
    }
}

运行上述代码，你会观察到第4和第5个任务会由调用 submit 方法的 main 线程执行，而不是被丢弃或抛出异常。

4. 自定义拒绝策略（高级）

当内置策略无法满足复杂业务需求时，你可以实现 RejectedExecutionHandler 接口来创建自定义策略。

1. 定义 实现类。

   • 创建 一个类，实现 java.util.concurrent.RejectedExecutionHandler 接口。
   • 重写 其唯一的方法 rejectedExecution(Runnable r, ThreadPoolExecutor executor)。
     • r：被拒绝的任务。
     • executor：执行拒绝策略的线程池实例，可以从中获取池的状态信息（如当前线程数、队列大小等）。

2. 编写 业务逻辑。

   • 在 rejectedExecution 方法中，根据业务需求处理被拒绝的任务。

3. 应用 自定义策略。

   • 在创建 ThreadPoolExecutor 时，传入 你的自定义策略实例。

自定义策略示例：记录日志并转移任务

假设我们需要一个策略：当任务被拒绝时，将任务信息记录到日志系统，并将任务转移给一个备用的、容量更大的队列进行后续处理。

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

// 1. 定义实现类
public class LoggingAndRedirectRejectionHandler implements RejectedExecutionHandler {
    private static final Logger logger = LoggerFactory.getLogger(LoggingAndRedirectRejectionHandler.class);
    private final BlockingQueue<Runnable> redirectQueue;
    private final AtomicInteger rejectedCount = new AtomicInteger(0);

    // 构造时传入一个备用队列
    public LoggingAndRedirectRejectionHandler(BlockingQueue<Runnable> redirectQueue) {
        this.redirectQueue = redirectQueue;
    }

    // 2. 重写处理方法
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // 记录详细日志
        logger.warn("任务被拒绝！当前池状态 - 活动线程: {}, 核心大小: {}, 最大大小: {}, 队列大小: {}",
                executor.getActiveCount(), executor.getCorePoolSize(),
                executor.getMaximumPoolSize(), executor.getQueue().size());
        rejectedCount.incrementAndGet();

        // 尝试将任务放入备用队列（这里需要处理队列满的情况）
        boolean offered = redirectQueue.offer(r);
        if (!offered) {
            logger.error("备用队列也已满，任务最终丢失！");
            // 可选择记录任务详情到数据库或文件，或进行其他降级处理
        }
    }

    public int getRejectedCount() {
        return rejectedCount.get();
    }
}

应用自定义策略：

// 创建备用队列
BlockingQueue<Runnable> backupQueue = new LinkedBlockingQueue<>(100);
// 创建自定义拒绝策略实例
LoggingAndRedirectRejectionHandler customHandler = new LoggingAndRedirectRejectionHandler(backupQueue);

// 创建线程池并使用自定义策略
ThreadPoolExecutor customExecutor = new ThreadPoolExecutor(
        2, 4, 60L, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(5),
        customHandler // 使用自定义策略
);

// ... 后续提交任务的代码 ...

// 可以启动一个单独的线程或使用另一个线程池，定期从 backupQueue 中取任务执行

5. 最佳实践与选择指南

• 明确业务容忍度：首先明确你的应用能否容忍任务丢失。绝不能在金融交易、订单处理等关键路径使用 DiscardPolicy 或 DiscardOldestPolicy。
• 首选 CallerRunsPolicy：对于大多数场景，CallerRunsPolicy 是一个安全且有效的选择。它通过降速保证了任务的不丢失，且易于理解。
• 避免抛出异常：在生产环境中，应谨慎使用 AbortPolicy（默认策略）。未捕获的 RejectedExecutionException 可能导致任务提交链中断。如果使用，务必在提交任务的代码处做好异常捕获和处理。
• 监控与告警：无论使用哪种策略，都应在拒绝发生时记录日志或上报监控指标。这是发现系统瓶颈和容量规划问题的重要信号。自定义策略是实现此目的的理想方式。
• 理解队列类型的影响：拒绝策略的触发与工作队列的类型和容量紧密相关。使用无界队列（如 LinkedBlockingQueue 不指定容量）理论上永远不会触发拒绝策略，但这可能导致内存溢出，需要极其谨慎。

选择或设计拒绝策略的本质，是在系统吞吐量、任务可靠性、资源安全性之间做出权衡。根据你的应用场景，配置最合适的策略，是构建稳定、可观测的并发系统的必要步骤。