线程池四种拒绝策略压测对比：AbortPolicy vs CallerRunsPolicy，数据说了算

一、引言

线程池是高并发场景下的核心组件，而拒绝策略则是线程池满载时的"安全阀"。很多开发者对四种拒绝策略的理解停留在理论层面，今天我们用真实数据说话，对比四种策略在相同压力下的表现。

测试条件：

• 核心线程：4
• 最大线程：8
• 队列容量：100
• 并发请求：200
• 任务耗时：模拟 100ms IO 操作

二、四种拒绝策略原理剖析

2.1 AbortPolicy（默认策略）

直接抛出 RejectedExecutionException 异常。

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

特点：

• 快速失败，吞吐量最高
• 异常必须被上游捕获，否则任务丢失
• 适合对任务可靠性要求高的场景

2.2 CallerRunsPolicy

由调用者线程直接执行被拒绝的任务。

public static class CallerRunsPolicy implements RejectedExecutionHandler {
    public CallerRunsPolicy() {}
    
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

特点：

• 天然背压机制，不会丢失任务
• 可能拖慢调用者线程（如 Tomcat 主线程）
• 适合对任务可靠性要求高但允许一定延迟的场景

2.3 DiscardPolicy

静默丢弃被拒绝的任务，不做任何处理。

public static class DiscardPolicy implements RejectedExecutionHandler {
    public DiscardPolicy() {}
    
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 什么都不做
    }
}

特点：

• 完全静默，最危险的策略
• 任务无声无息丢失，难以排查
• 仅适合非核心任务

2.4 DiscardOldestPolicy

丢弃队列中最老的任务，然后尝试重新提交。

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    public DiscardOldestPolicy() {}
    
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll();  // 丢弃最老的任务
            e.execute(r);         // 重新提交新任务
        }
    }
}

特点：

• 比 DiscardPolicy 稍好，但仍会丢失任务
• 可能导致"饥饿"现象：旧任务永远无法执行
• 适合对实时性要求高的场景

三、压测数据对比

3.1 测试环境

// 统一线程池配置
int corePoolSize = 4;
int maxPoolSize = 8;
int queueCapacity = 100;
int concurrentRequests = 200;
long taskDurationMs = 100;  // 模拟 IO 耗时

3.2 测试结果

3.3 数据解读

AbortPolicy：

• 吞吐量最高，但代价是接近一半的任务被拒绝
• 如果上游没有正确处理异常，这些任务就永久丢失了
• 异常率高达 46%，生产环境必须做好异常处理

CallerRunsPolicy：

• 任务丢失率为 0，所有任务都能执行
• 平均延迟和 P99 延迟显著高于其他策略（约 2 倍）
• 吞吐量下降约 15%，但换取了任务可靠性
• 调用者线程会被占用执行任务，可能影响请求响应速度

DiscardPolicy / DiscardOldestPolicy：

• 吞吐量与 AbortPolicy 相当
• 异常率为 0（静默丢弃），但任务丢失率同样高达 46%
• 最危险的策略：丢失任务却没有任何告警

四、自定义拒绝策略：死信队列方案

4.1 问题分析

四种默认策略都有局限性：

• AbortPolicy：可能丢任务
• CallerRunsPolicy：拖慢主线程
• DiscardPolicy：静默丢任务
• DiscardOldestPolicy：丢老任务

最佳方案：自定义拒绝策略 + 死信队列 + 异步重试

4.2 实现方案

public class DeadLetterQueuePolicy implements RejectedExecutionHandler {
    
    private final BlockingQueue<Runnable> deadLetterQueue;
    private final ExecutorService retryExecutor;
    private final AtomicInteger rejectCount = new AtomicInteger(0);
    private static final int MAX_RETRY_COUNT = 3;
    
    public DeadLetterQueuePolicy(int queueSize) {
        this.deadLetterQueue = new LinkedBlockingQueue<>(queueSize);
        this.retryExecutor = Executors.newSingleThreadExecutor(r -> {
            Thread t = new Thread(r, "dead-letter-retry");
            t.setDaemon(true);
            return t;
        });
        startRetryLoop();
    }
    
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        int count = rejectCount.incrementAndGet();
        if (!deadLetterQueue.offer(r)) {
            log.error("Dead letter queue is full! Task rejected, total rejected: {}", count);
            return;
        }
        log.warn("Task rejected, added to dead letter queue, total rejected: {}", count);
    }
    
    private void startRetryLoop() {
        retryExecutor.submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    Runnable task = deadLetterQueue.take();
                    retryTask(task, 0);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });
    }
    
    private void retryTask(Runnable task, int retryCount) {
        if (retryCount >= MAX_RETRY_COUNT) {
            log.error("Task failed after {} retries, dropping: {}", MAX_RETRY_COUNT, task);
            return;
        }
        try {
            task.run();
            log.info("Task recovered successfully, retry count: {}", retryCount);
        } catch (Exception e) {
            log.warn("Task retry failed, attempt {}/{}, error: {}", 
                retryCount + 1, MAX_RETRY_COUNT, e.getMessage());
            try {
                Thread.sleep((long) Math.pow(2, retryCount) * 1000);
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
            }
            retryTask(task, retryCount + 1);
        }
    }
    
    public int getRejectCount() {
        return rejectCount.get();
    }
}

4.3 监控告警集成

public class ThreadPoolMonitor {
    
    private final ThreadPoolExecutor executor;
    private final ScheduledExecutorService monitorExecutor;
    private static final long MONITOR_INTERVAL_MS = 5000;
    private static final double REJECT_RATE_THRESHOLD = 0.1;  // 10%
    
    public ThreadPoolMonitor(ThreadPoolExecutor executor, String poolName) {
        this.executor = executor;
        this.monitorExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread t = new Thread(r, "thread-pool-monitor-" + poolName);
            t.setDaemon(true);
            return t;
        });
        startMonitoring();
    }
    
    private void startMonitoring() {
        monitorExecutor.scheduleAtFixedRate(() -> {
            int activeCount = executor.getActiveCount();
            int poolSize = executor.getPoolSize();
            int queueSize = executor.getQueue().size();
            long completedTaskCount = executor.getCompletedTaskCount();
            long rejectedTaskCount = executor.getRejectedExecutionHandler() 
                instanceof DeadLetterQueuePolicy 
                    ? ((DeadLetterQueuePolicy) executor.getRejectedExecutionHandler()).getRejectCount() 
                    : 0;
            
            double utilization = (double) activeCount / poolSize;
            double rejectRate = completedTaskCount > 0 
                ? (double) rejectedTaskCount / (completedTaskCount + rejectedTaskCount) 
                : 0;
            
            log.info("ThreadPoolMonitor - Active: {}, PoolSize: {}, Queue: {}, " +
                "Completed: {}, Rejected: {}, Utilization: {:.2f}, RejectRate: {:.2f}",
                activeCount, poolSize, queueSize, completedTaskCount, 
                rejectedTaskCount, utilization, rejectRate);
            
            if (rejectRate > REJECT_RATE_THRESHOLD) {
                log.error("ALERT: Reject rate exceeded threshold! Current: {:.2f}%, Threshold: {}%",
                    rejectRate * 100, REJECT_RATE_THRESHOLD * 100);
                // 发送告警：钉钉/微信/邮件
            }
            
            if (utilization > 0.9 && queueSize > executor.getQueue().remainingCapacity() * 0.9) {
                log.warn("ALERT: Thread pool is near capacity! Consider scaling up.");
            }
        }, 0, MONITOR_INTERVAL_MS, TimeUnit.MILLISECONDS);
    }
}

五、生产环境最佳实践

5.1 策略选择建议

5.2 90% 场景的选择

生产环境 90% 的场景推荐使用：CallerRunsPolicy + 监控告警

理由：

1. 任务不丢失：所有任务都能执行，不会出现数据丢失
2. 天然背压：当线程池满载时，调用者线程会被阻塞，自动降低请求速率
3. 实现简单：无需额外开发，Spring Boot 可直接配置
4. 配合监控：通过监控告警及时发现线程池满载问题

5.3 关键业务的选择

关键业务（支付、订单等）推荐使用：自定义拒绝策略 + 死信队列

理由：

1. 任务不丢失：被拒绝的任务进入死信队列
2. 异步重试：后台线程自动重试，不影响主线程
3. 可追溯：有完整的日志记录，便于排查问题
4. 告警机制：拒绝率超过阈值时及时告警

六、实战配置示例

6.1 Spring Boot 配置 CallerRunsPolicy

@Configuration
public class ThreadPoolConfig {
    
    @Bean(name = "businessThreadPool")
    public Executor businessThreadPool() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(8);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("business-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        executor.initialize();
        
        new ThreadPoolMonitor((ThreadPoolExecutor) executor.getThreadPoolExecutor(), "business");
        return executor;
    }
    
    @Bean(name = "criticalThreadPool")
    public Executor criticalThreadPool() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(8);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("critical-");
        
        DeadLetterQueuePolicy deadLetterQueuePolicy = new DeadLetterQueuePolicy(1000);
        executor.setRejectedExecutionHandler(deadLetterQueuePolicy);
        
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        executor.initialize();
        
        new ThreadPoolMonitor((ThreadPoolExecutor) executor.getThreadPoolExecutor(), "critical");
        return executor;
    }
}

6.2 使用示例

@Service
public class OrderService {
    
    @Autowired
    @Qualifier("businessThreadPool")
    private Executor businessExecutor;
    
    @Autowired
    @Qualifier("criticalThreadPool")
    private Executor criticalExecutor;
    
    public void processNormalOrder(Order order) {
        businessExecutor.execute(() -> {
            // 普通订单处理逻辑
        });
    }
    
    public void processPaymentOrder(PaymentOrder order) {
        criticalExecutor.execute(() -> {
            // 支付订单处理逻辑
        });
    }
}

七、总结

核心结论：

1. AbortPolicy：吞吐量最高，但任务可能丢失，需要上游处理异常
2. CallerRunsPolicy：任务不丢失，天然背压，但会拖慢主线程
3. DiscardPolicy / DiscardOldestPolicy：静默丢弃任务，最危险，不建议生产使用
4. 自定义死信队列：任务可靠性最高，适合关键业务

最佳实践：

• 90% 普通业务：使用 CallerRunsPolicy + 监控告警
• 10% 关键业务：使用自定义拒绝策略 + 死信队列 + 异步重试

监控指标：

• 线程池活跃线程数
• 队列使用率
• 任务拒绝率
• 线程池利用率

💡 互动话题：你在实际项目中遇到过线程池拒绝策略导致的问题吗？你是如何处理的？欢迎在评论区分享你的经验！