ThreadPoolExecutor线程池使用时触发拒绝策略

1. 问题拆解

(1) 基本参数

• 总数据量：10000 条。
• 每批次处理：100 条。
• 每批次处理时间：2 分钟。
• 线程池配置：
  • corePoolSize = 5（核心线程数）。
  • maxPoolSize = 10（最大线程数）。
  • queueCapacity = 20（队列容量）。

(2) 关键公式

线程池的最大任务处理能力为：

corePoolSize + queueCapacity + (maxPoolSize - corePoolSize)
= 5 + 20 + (10 - 5)
= 30

这意味着线程池最多可以同时处理 30 个批次 的任务。

(3) 任务提交速率

假设每批次处理时间为 2 分钟，那么每个线程每 2 分钟可以完成一个批次的任务。因此，线程池的实际吞吐量取决于线程数量：

• 核心线程数（5）：每 2 分钟可以处理 5 个批次。
• 最大线程数（10）：每 2 分钟可以处理 10 个批次。

如果任务提交速度过快（例如一次性提交所有批次），可能会导致任务堆积甚至触发拒绝策略。

2. 是否会导致 3100-10000 条记录无法处理？

(1) 任务堆积的可能性

• 总共有 10000 条记录，每次处理 100 条，总共需要提交 100 个批次的任务。
• 线程池的最大任务处理能力是 30 个批次：
  • 如果一次性提交所有 100 个批次，前 30 个批次会被处理，剩下的 70 个批次会堆积在队列中或触发拒绝策略。
  • 如果线程池的处理速度（每 2 分钟完成 5-10 个批次）低于任务提交速度，则可能导致任务堆积。

(2) 具体场景分析

• 如果任务是一次性提交的：
  • 前 30 个批次会被线程池接受（包括核心线程处理、队列存储和额外线程处理）。
  • 第 31 及后续批次会触发拒绝策略（如抛出异常或丢弃任务）。
  • 结果：3100 条之后的记录可能无法被处理。
• 如果任务是分批提交的：
  • 按照线程池的处理能力，控制任务提交速度（例如每 2 分钟提交 5-10 个批次），则可以避免任务堆积和拒绝策略触发。
  • 结果：所有 10000 条记录都可以被处理。

3. 解决方案

为了避免任务堆积和拒绝策略触发，可以采取以下措施：

(1) 控制任务提交速度

• 使用生产者-消费者模式，动态控制任务的提交速度。
• 示例代码：

  @Service
  public class DataProcessingService {

      @Autowired
      private TaskExecutor taskExecutor;

      public void processDataInBatches() {
          int batchSize = 100; // 每批次处理的数据量
          int totalRecords = 10000; // 总记录数
          int totalPages = (int) Math.ceil((double) totalRecords / batchSize);

          for (int i = 0; i < totalPages; i++) {
              int offset = i * batchSize;
              List<DataEntity> batchData = fetchDataBatch(offset, batchSize);

              // 提交任务到线程池
              taskExecutor.execute(() -> processBatch(batchData));

              // 控制任务提交速度
              if ((i + 1) % 5 == 0) { // 每提交 5 个批次后等待 2 分钟
                  try {
                      Thread.sleep(2 * 60 * 1000); // 等待 2 分钟
                  } catch (InterruptedException e) {
                      Thread.currentThread().interrupt();
                  }
              }
          }
      }

      private List<DataEntity> fetchDataBatch(int offset, int limit) {
          // 数据库查询逻辑
          return new ArrayList<>();
      }

      private void processBatch(List<DataEntity> batchData) {
          // 数据处理逻辑
      }
  }

(2) 增加队列容量

• 如果任务提交速度较快且无法完全控制，可以适当增加队列容量（如将 queueCapacity 从 20 增加到 100）。
• 注意：无界队列可能导致内存溢出，因此建议使用有界队列并结合拒绝策略。

(3) 动态调整线程池大小

• 根据任务提交速度和系统负载动态调整线程池的核心线程数和最大线程数。
• 示例：executor.setCorePoolSize(Runtime.getRuntime().availableProcessors()); executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 2);

(4) 自定义拒绝策略

• 如果任务被拒绝，可以选择其他策略（如重试或记录日志）而不是直接丢弃任务。
• 示例：executor.setRejectedExecutionHandler((r, executor) -> { System.out.println("Task rejected, retrying..."); try { Thread.sleep(1000); // 等待 1 秒后重新提交 executor.execute(r); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } });

4. 总结

• 如果一次性提交所有 100 个批次的任务，可能会导致任务堆积和拒绝策略触发，从而使得 3100 条之后的记录无法被处理。
• 通过控制任务提交速度、增加队列容量、动态调整线程池大小或自定义拒绝策略，可以确保所有 10000 条记录都能被处理。