【线程池饱和策略】线程池饱和策略及自定义方法

常见的线程池饱和策略及使用场景

当线程池的任务队列已满且线程数达到最大值时，新的任务会触发饱和策略（拒绝策略）。Java 提供了四种默认策略：

• AbortPolicy（默认策略）
  行为：直接抛出 RejectedExecutionException 异常。
  适用场景：
  
  
  
  • 需要严格保证任务不丢失的场景（如支付交易）。
    
  • 调用方需明确感知任务提交失败，以便回滚或重试。
    示例：金融系统中订单处理，失败需立即告警。
    
  
  
• CallerRunsPolicy
  行为：由提交任务的线程（调用者线程）直接执行该任务。
  适用场景：
  
  
  
  • 需要避免任务丢失，且能接受任务执行速度下降。
    
  • 利用调用线程执行任务，变相限制新任务提交速率（负反馈）。
    示例：Web 服务器处理请求，高峰期用用户线程执行任务，防止雪崩。
    
  
  
• DiscardPolicy
  行为：直接丢弃新任务，无任何通知。
  适用场景：
  
  
  
  • 允许静默丢弃非关键任务（如日志收集）。
    
  • 对实时性要求低且可容忍数据丢失的场景。
    示例：监控数据上报，丢弃部分数据不影响整体统计。
    
  
  
• DiscardOldestPolicy
  行为：丢弃队列中最旧的任务（队首），然后重试提交新任务。
  适用场景：
  
  
  
  • 新任务优先级高于旧任务（如实时消息推送）。
    
  • 可容忍历史任务延迟或丢失。
    注意：优先队列中“最旧”不一定是提交时间最早，而是优先级最低。
    示例：股票价格更新，最新价格比历史价格更重要。
    
  
  

如何自定义饱和策略？

通过实现 RejectedExecutionHandler 接口并重写 rejectedExecution 方法：

1. import java.util.concurrent.RejectedExecutionHandler;
2. import java.util.concurrent.ThreadPoolExecutor;
4. public class CustomRejectionPolicy implements RejectedExecutionHandler {
5.     @Override
6.     public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) {
7.         // 自定义处理逻辑（示例：记录日志 + 重试一次）
8.         System.err.println("任务被拒绝: " + task);
9.         
10.         if (!executor.isShutdown()) {
11.             // 重试一次
12.             executor.execute(task);
13.             System.out.println("任务重试提交成功");
14.         }
15.     }
16. }

复制代码

使用自定义策略

1. import java.util.concurrent.*;
3. public class ThreadPoolDemo {
4.     public static void main(String[] args) {
5.         // 创建线程池并指定自定义策略
6.         ThreadPoolExecutor executor = new ThreadPoolExecutor(
7.             2, // 核心线程数
8.             4, // 最大线程数
9.             60, TimeUnit.SECONDS,
10.             new ArrayBlockingQueue<>(2), // 容量为2的队列
11.             new CustomRejectionPolicy()   // 自定义饱和策略
12.         );
14.         // 提交任务（模拟饱和）
15.         for (int i = 0; i < 10; i++) {
16.             executor.execute(() -> {
17.                 try {
18.                     Thread.sleep(1000);
19.                 } catch (InterruptedException e) {
20.                     Thread.currentThread().interrupt();
21.                 }
22.             });
23.         }
24.         executor.shutdown();
25.     }
26. }

复制代码

自定义策略的常见实践

• 记录日志/告警：
  1. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  2.     log.error("任务拒绝: {}", r, new RejectedException());
  3. }

  复制代码
• 暂存到数据库/Redis：
  1. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  2.     taskRedisQueue.save(r); // 保存到外部存储，后续恢复
  3. }

  复制代码
• 有限次重试：
  1. private int maxRetries = 3;
  2. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  3.     if (retryCount.getAndIncrement() < maxRetries) {
  4.         e.execute(r); // 重试提交
  5.     }
  6. }

  复制代码
• 降级处理：
  1. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  2.     runFallbackService((Task) r); // 执行降级逻辑
  3. }

  复制代码

总结

策略行为适用场景AbortPolicy抛出异常需严格保证任务不丢失（如交易系统）CallerRunsPolicy调用者线程执行任务限制提交速率，避免雪崩（如Web服务）DiscardPolicy静默丢弃新任务容忍数据丢失（如日志收集）DiscardOldestPolicy丢弃队首任务并重试提交新任务新任务优先级更高（如实时推送）自定义策略：通过实现 RejectedExecutionHandler 接口，可结合业务需求实现降级、重试、持久化等灵活逻辑。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！