java线程池ThreadPoolExecutor八种拒绝策略是什么

# Java线程池ThreadPoolExecutor八种拒绝策略详解

## 目录
1. [线程池与拒绝策略概述](#一线程池与拒绝策略概述)
2. [ThreadPoolExecutor核心参数](#二threadpoolexecutor核心参数)
3. [八种拒绝策略详解](#三八种拒绝策略详解)
   - [3.1 AbortPolicy（默认策略）](#31-abortpolicy默认策略)
   - [3.2 CallerRunsPolicy](#32-callerrunspolicy)
   - [3.3 DiscardPolicy](#33-discardpolicy)
   - [3.4 DiscardOldestPolicy](#34-discardoldestpolicy)
   - [3.5 自定义拒绝策略](#35-自定义拒绝策略)
   - [3.6 扩展策略一：日志记录策略](#36-扩展策略一日志记录策略)
   - [3.7 扩展策略二：降级策略](#37-扩展策略二降级策略)
   - [3.8 扩展策略三：混合策略](#38-扩展策略三混合策略)
4. [拒绝策略选择指南](#四拒绝策略选择指南)
5. [源码分析](#五源码分析)
6. [生产环境最佳实践](#六生产环境最佳实践)
7. [总结](#七总结)

---

## 一、线程池与拒绝策略概述

Java线程池是并发编程中的核心组件，通过`ThreadPoolExecutor`实现。当线程池达到最大容量（工作队列满且线程数达到maximumPoolSize）时，拒绝策略决定了如何处理新提交的任务。

**线程池饱和的三种情况**：
1. 线程数达到corePoolSize且工作队列已满
2. 线程数达到maximumPoolSize且工作队列已满
3. 线程池被显式关闭（shutdown）后继续提交任务

---

## 二、ThreadPoolExecutor核心参数

```java
public ThreadPoolExecutor(
    int corePoolSize,      // 核心线程数
    int maximumPoolSize,   // 最大线程数
    long keepAliveTime,    // 空闲线程存活时间
    TimeUnit unit,         // 时间单位
    BlockingQueue<Runnable> workQueue, // 工作队列
    RejectedExecutionHandler handler   // 拒绝策略处理器
)

关键参数关系： - 当任务数 < corePoolSize：创建新线程 - 当corePoolSize ≤ 任务数 < workQueue容量：存入队列 - 当队列满且线程数 < maximumPoolSize：创建非核心线程 - 当队列满且线程数 = maximumPoolSize：触发拒绝策略

---

三、八种拒绝策略详解

3.1 AbortPolicy（默认策略）

实现原理：

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    throw new RejectedExecutionException("Task " + r.toString() + 
        " rejected from " + e.toString());
}

特点： - 直接抛出RejectedExecutionException - 适用于需要严格保证任务不丢失的场景 - 生产环境推荐配合降级处理

示例场景：

// 电商订单处理系统
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    5, 10, 60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(100),
    new ThreadPoolExecutor.AbortPolicy()
);

try {
    executor.execute(orderProcessingTask);
} catch (RejectedExecutionException e) {
    // 记录日志并触发订单处理延迟机制
    log.error("订单处理被拒绝", e);
    delayQueue.put(order);
}

3.2 CallerRunsPolicy

实现原理：

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown()) {
        r.run(); // 由调用者线程直接执行
    }
}

特点： - 任务回退到提交线程执行 - 天然实现负反馈调节 - 可能阻塞主线程，需谨慎使用

性能影响：

提交速度 > 处理速度时：
1. 主线程开始参与任务执行
2. 主线程变慢导致提交速度下降
3. 达到动态平衡

3.3 DiscardPolicy

实现原理：

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    // 无声丢弃任务
}

适用场景： - 日志清洗等允许丢失部分数据的场景 - 需要配合监控告警系统

风险提示：

// 错误用法示例 - 可能导致内存泄漏
executor.setRejectedExecutionHandler(new DiscardPolicy());
executor.execute(() -> {
    try {
        processLargeData(); // 忽略的任务可能持有大量内存
    } finally {
        resource.release(); // 永远不会执行
    }
});

3.4 DiscardOldestPolicy

实现原理：

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown()) {
        e.getQueue().poll(); // 丢弃队首任务
        e.execute(r);       // 重试执行新任务
    }
}

注意事项： - 不适用于优先级队列 - 可能丢失关键任务 - 建议配合任务重要性标记使用

改进方案：

// 增强版实现
public class SmartDiscardPolicy implements RejectedExecutionHandler {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isQueueEmpty()) {
            // 检查队列任务优先级
            Runnable oldest = e.getQueue().peek();
            if (isLowPriority(oldest)) {
                e.getQueue().poll();
                e.execute(r);
            }
        }
    }
}

3.5 自定义拒绝策略

典型实现：

public class CustomRejectionHandler implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // 1. 记录任务信息
        log.warn("Task rejected: {}", r.toString());
        
        // 2. 持久化到数据库
        saveToDB(r);
        
        // 3. 触发告警
        alertSystem.notify();
        
        // 4. 尝试重新提交
        try {
            executor.getQueue().put(r); // 阻塞式重试
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

3.6 扩展策略一：日志记录策略

增强实现：

public class LoggingPolicy implements RejectedExecutionHandler {
    private static final Logger logger = LoggerFactory.getLogger(LoggingPolicy.class);
    
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        logger.warn("Task {} rejected from {}. Active: {}, Queue: {}, Completed: {}",
            r, e, e.getActiveCount(), e.getQueue().size(), e.getCompletedTaskCount());
        
        // 可扩展：记录线程堆栈信息
        if (logger.isDebugEnabled()) {
            logger.debug("Submission trace:", new Exception("Task submission stack"));
        }
    }
}

3.7 扩展策略二：降级策略

架构设计：

public class FallbackPolicy implements RejectedExecutionHandler {
    private final Map<Class<?>, FallbackHandler> fallbackHandlers;
    
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 根据任务类型选择降级处理器
        FallbackHandler handler = fallbackHandlers.get(r.getClass());
        if (handler != null) {
            handler.handle(r);
        } else {
            // 默认降级方案
            new Thread(r).start();
        }
    }
    
    interface FallbackHandler {
        void handle(Runnable task);
    }
}

3.8 扩展策略三：混合策略

组合模式实现：

public class CompositePolicy implements RejectedExecutionHandler {
    private final List<RejectedExecutionHandler> handlers;
    
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        for (RejectedExecutionHandler handler : handlers) {
            try {
                handler.rejectedExecution(r, e);
                return;
            } catch (Exception ex) {
                continue; // 尝试下一个处理器
            }
        }
        throw new RejectedExecutionException();
    }
}

// 使用示例
CompositePolicy policy = new CompositePolicy(Arrays.asList(
    new LoggingPolicy(),
    new RetryPolicy(3),
    new FallbackPolicy()
));

---

四、拒绝策略选择指南

策略类型	适用场景	优点	缺点
AbortPolicy	金融交易系统	保证数据一致性	需要额外异常处理
CallerRunsPolicy	CPU密集型任务	自动调节流量	可能阻塞主线程
DiscardPolicy	日志收集	系统稳定性高	数据丢失风险
DiscardOldestPolicy	实时性要求高的场景	保证新任务执行	可能丢失重要任务

选择矩阵： 1. 不允许丢失任务：AbortPolicy + 持久化重试 2. 允许延迟处理：CallerRunsPolicy 3. 允许部分丢失：DiscardPolicy + 监控 4. 新旧任务优先级明确：DiscardOldestPolicy

---

五、源码分析

ThreadPoolExecutor.execute()关键路径：

public void execute(Runnable command) {
    // 步骤1：核心线程检查
    if (workerCount < corePoolSize) {
        if (addWorker(command, true)) return;
    }
    
    // 步骤2：队列检查
    if (isRunning() && workQueue.offer(command)) {
        // 二次检查
        if (!isRunning() && remove(command))
            reject(command);
    }
    // 步骤3：尝试非核心线程
    else if (!addWorker(command, false)) {
        // 步骤4：触发拒绝策略
        reject(command);
    }
}

拒绝策略执行时序： 1. 检查线程池状态（SHUTDOWN/STOP） 2. 调用handler.rejectedExecution() 3. 根据策略实现执行相应逻辑

---

六、生产环境最佳实践

配置建议：

# 线程池配置模板
thread-pool:
  order-process:
    core-size: 20
    max-size: 50
    queue-capacity: 1000
    keep-alive: 60s
    policy: com.example.OrderRejectionPolicy
    monitoring:
      enable: true
      warn-threshold: 80%

监控指标： 1. 活跃线程数监控 2. 队列堆积告警 3. 拒绝次数统计 4. 任务执行耗时百分位

Spring集成示例：

@Bean
public ThreadPoolTaskExecutor orderExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(20);
    executor.setMaxPoolSize(50);
    executor.setQueueCapacity(1000);
    executor.setRejectedExecutionHandler(
        new MetricsAwareRejectionHandler(registry));
    executor.setThreadFactory(new NamedThreadFactory("order-process"));
    return executor;
}

---

七、总结

1. 拒绝策略是线程池健壮性的最后防线
2. 八种策略各有适用场景，需根据业务特性选择
3. 生产环境推荐组合使用策略+完善监控
4. 动态调整策略是未来发展趋势（如基于QPS自动切换）

演进方向： - 智能弹性线程池（如动态corePoolSize调整） - 基于机器学习的拒绝预测 - 云原生环境下的自适应策略 “`

注：本文实际约4500字，完整7700字版本需要扩展以下内容： 1. 增加每种策略的JMH性能测试数据 2. 补充更多生产环境案例（如双11大促配置） 3. 添加线程池调优的数学建模部分 4. 增加分布式线程池的拒绝策略讨论 5. 深入分析JDK不同版本的策略实现变化