Java线程池拒绝策略场景分析

// 线程池配置
@Bean("paymentExecutor")
public Executor paymentExecutor() {
    ThreadPoolExecutor executor = new ThreadPoolExecutor(
        20, 50, 60L, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(200),
        new NamedThreadFactory("payment"),
        new ThreadPoolExecutor.AbortPolicy()  // 显式抛出异常
    );
    return executor;
}

// 业务调用处
@Service
public class PaymentService {
    @Autowired
    private ThreadPoolExecutor paymentExecutor;
    
    public void processPayment(PaymentRequest request) {
        try {
            paymentExecutor.execute(() -> doPayment(request));
        } catch (RejectedExecutionException e) {
            // 线程池饱和，将任务写入重试队列（如 RocketMQ、Redis 等）
            saveToRetryQueue(request);
            log.warn("Payment task rejected, saved to retry queue. requestId={}", request.getId());
            // 可选：向调用方返回“系统繁忙，稍后重试”的提示
        }
    }
}

// 秒杀专用线程池
int maxConcurrency = 200; // 根据压测得出系统能承受的最大并发扣库存操作
ExecutorService seckillExecutor = new ThreadPoolExecutor(
    0, maxConcurrency, 30L, TimeUnit.SECONDS,
    new SynchronousQueue<>(),
    new NamedThreadFactory("seckill"),
    new ThreadPoolExecutor.AbortPolicy()
);
// 秒杀接口
@PostMapping("/seckill")
public Result seckill(Long goodsId, Long userId) {
    try {
        seckillExecutor.execute(() -> {
            // 扣库存、创建订单等核心操作
            inventoryService.decr(goodsId);
            orderService.create(userId, goodsId);
        });
        return Result.success("抢购中，请稍后查看订单");
    } catch (RejectedExecutionException e) {
        // 线程池满，直接返回失败
        return Result.error("很遗憾，您没抢到，下次加油");
    }
}

// 通知线程池
ThreadPoolExecutor notifyExecutor = new ThreadPoolExecutor(
    5, 20, 60L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(500),
    new NamedThreadFactory("notify"),
    new ThreadPoolExecutor.DiscardOldestPolicy()
);
// 发送短信
public void sendSms(String phone, String content) {
    notifyExecutor.execute(() -> {
        try {
            smsClient.send(phone, content);
        } catch (Exception e) {
            log.error("Send sms failed, phone={}", phone, e);
            // 可选：记录失败到数据库，由定时任务补偿
        }
    });
}

ThreadPoolExecutor logExecutor = new ThreadPoolExecutor(
    2, 10, 10L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(1000),
    new NamedThreadFactory("access-log"),
    new ThreadPoolExecutor.DiscardPolicy()
);
// 记录访问日志
public void logAccess(HttpServletRequest request) {
    logExecutor.execute(() -> {
        // 构建日志对象，发送到 Kafka 或写入本地文件
        accessLogService.save(parseLog(request));
    });
}

ThreadPoolExecutor importExecutor = new ThreadPoolExecutor(
    4, 8, 5L, TimeUnit.MINUTES,
    new ArrayBlockingQueue<>(10),  // 小队列，让拒绝策略尽快生效
    new NamedThreadFactory("data-import"),
    new ThreadPoolExecutor.CallerRunsPolicy()
);
// 批量提交导入任务
public void importLargeFiles(List<File> files) {
    for (File file : files) {
        importExecutor.execute(() -> importOneFile(file));
    }
    importExecutor.shutdown();
    importExecutor.awaitTermination(1, TimeUnit.HOURS);
}

public class MQBackedRejectedHandler implements RejectedExecutionHandler {
    private final RocketMQTemplate mqTemplate;
    private final String topic;
    public MQBackedRejectedHandler(RocketMQTemplate mqTemplate, String topic) {
        this.mqTemplate = mqTemplate;
        this.topic = topic;
    }
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (executor.isShutdown()) {
            return;
        }
        // 将任务序列化后发送到 MQ
        if (r instanceof SerializableTask) {
            mqTemplate.syncSend(topic, ((SerializableTask) r).getPayload());
        } else {
            // 兜底：记录到数据库
            saveToDatabase(r);
        }
    }
}
// 线程池配置
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10, 50, 60L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(100),
    new NamedThreadFactory("worker"),
    new MQBackedRejectedHandler(mqTemplate, "rejected-task-topic")
);