文章目录
- 一、为什么会出现线程池拒绝任务？
- 二、为什么用 RabbitMQ 做补偿？
- 三、整体架构图（重点）
- 四、Demo 业务场景
- 五、关键代码
- 六、总结
在实际业务中，我们经常需要使用
线程池执行异步任务，例如：
- 生成 PDF 报告
- 发送短信 / 邮件
- 执行大批量数据库操作
- 调外部接口
但线程池始终是有限资源：
- 核心线程数有限
- 最大线程数有限
- 队列容量有限
当系统高峰时流量过大，很容易出现：
线程池满了 → 拒绝策略触发 → 任务直接丢失
对于关键业务任务（如生成报告、订单后处理、补偿扣减）， 不允许丢失。
所以我们需要：
当线程池拒绝任务时，把任务丢到 RabbitMQ 中间缓冲，系统空闲后再补偿执行。
一、为什么会出现线程池拒绝任务？
先看线程池的本质：
线程池 =（固定数量线程 + 任务队列）
设定：
- corePoolSize = 2
- maxPoolSize = 4
- queueSize = 5
当系统瞬间执行 20 个任务时：
- 前面 2 个 → 核心线程执行
- 再来几个 → 放到队列
- 队列满了 → 拿 maxPoolSize 顶上
- maxPoolSize 线程也满了 → 拒绝策略触发
默认拒绝策略是：
- AbortPolicy（抛异常）
- DiscardPolicy（直接丢）
👉 任务丢了 = 数据不一致 = 事故
二、为什么用 RabbitMQ 做补偿？
RabbitMQ 可以承担：
- 缓冲层：线程池执行不了的任务先存着
- 持久化：服务器宕机、重启任务不会丢
- 削峰填谷：系统高峰时 MQ 扛流量，低谷时慢慢消费
- 异步重试：可支持死信队列、延迟重试
一句话：
RabbitMQ 能保证任务“不丢、能补偿、最终执行成功”。
三、整体架构图（重点）
┌──────────────────┐
│ Controller │
│ （创建任务） │
└───────┬──────────┘
│
▼
┌──────────────────┐
│ ThreadPool │
│ 执行 Runnable │
└───────┬──────────┘
│（如果满）
▼
┌───────────────────────────┐
│ RejectedExecutionHandler │
│ （拒绝策略） │
│ ① 任务状态标记 REJECTED │
│ ② 发送消息到 MQ │
└─────────┬─────────────────┘
│
▼
┌──────────────────┐
│ RabbitMQ Queue │
│ （补偿队列） │
└─────────┬────────┘
│
▼
┌───────────────────────────┐
│ MQ Consumer │
│ 收到消息 → 再投入线程池 │
│ → 正常执行 processTask │
└───────────────────────────┘
核心思想：
线程池执行不了 → 先放 MQ MQ 拉出来 → 再交给线程池执行 最终保证“任务必达”
四、Demo 业务场景
我们用“体检系统生成报告”为例：
-
用户发起体检 → 写入 t_exam
-
创建一条异步任务 → 写入 t_report_task
-
任务交给线程池执行 PDF 生成
-
线程池满 → 触发拒绝策略 →
- t_report_task 标记为 REJECTED
- msg 发到 RabbitMQ
-
MQ 消费者把任务拉出来 → 再让线程池执行
-
最终 t_report_task 状态 → SUCCESS
这个场景真实且简单，不依赖外部业务。
五、关键代码
https://gitee.com/donglin-bit/pool-mq
1、定义 RabbitMQ 配置
@Configuration
public class RabbitConfig {
public static final String REPORT_COMP_EXCHANGE = "report.compensation.exchange";
public static final String REPORT_COMP_QUEUE = "report.compensation.queue";
public static final String REPORT_COMP_ROUTING = "report.compensation.routing";
@Bean
public DirectExchange reportCompensationExchange() {
return new DirectExchange(REPORT_COMP_EXCHANGE);
}
@Bean
public Queue reportCompensationQueue() {
return new Queue(REPORT_COMP_QUEUE, true);
}
@Bean
public Binding reportCompensationBinding() {
return BindingBuilder
.bind(reportCompensationQueue())
.to(reportCompensationExchange())
.with(REPORT_COMP_ROUTING);
}
}
2、JSON 消息转换器（必须加）
@Bean
public MessageConverter jacksonMessageConverter() {
return new Jackson2JsonMessageConverter();
}
3、RabbitTemplate & Listener 使用 JSON
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory,
MessageConverter jacksonMessageConverter) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
template.setMessageConverter(jacksonMessageConverter);
return template;
}
4、自定义拒绝策略（关键）
@Slf4j
public class ReportRejectedHandler implements RejectedExecutionHandler {
private final ReportTaskService reportTaskService;
private final RabbitTemplate rabbitTemplate;
public ReportRejectedHandler(ReportTaskService reportTaskService,
RabbitTemplate rabbitTemplate) {
this.reportTaskService = reportTaskService;
this.rabbitTemplate = rabbitTemplate;
}
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
if (r instanceof ReportTaskRunnable runnable) {
// ① 标记任务为 REJECTED（数据库存证）
reportTaskService.markRejected(
runnable.getTaskId(),
"Thread pool rejected, send to MQ"
);
// ② 发 MQ 补偿
ReportTaskMessage msg = new ReportTaskMessage();
msg.setTaskId(runnable.getTaskId());
msg.setPatientId(runnable.getPatientId());
msg.setExamId(runnable.getExamId());
rabbitTemplate.convertAndSend(
RabbitConfig.REPORT_COMP_EXCHANGE,
RabbitConfig.REPORT_COMP_ROUTING,
msg
);
log.warn("任务被线程池拒绝，已发到 MQ taskId={}", runnable.getTaskId());
}
}
}
5、MQ 消费者补偿执行任务
@Component
@RequiredArgsConstructor
@Slf4j
public class ReportCompensationConsumer {
private final ThreadPoolExecutor reportExecutor;
private final ReportTaskService reportTaskService;
/**
* 从补偿队列拿到任务，再交给线程池处理
*/
@RabbitListener(queues = REPORT_COMP_QUEUE)
public void onMessage(ReportTaskMessage msg) {
log.info("收到报告补偿消息: {}", msg);
ReportTaskRunnable runnable = new ReportTaskRunnable(
msg.getTaskId(),
msg.getPatientId(),
msg.getExamId(),
reportTaskService
);
try {
reportExecutor.execute(runnable);
} catch (RejectedExecutionException e) {
// 极端情况下，补偿线程池也满了
log.error("补偿线程池仍然满载，taskId={}", msg.getTaskId(), e);
// 这里简单让异常抛出，Spring AMQP 默认会重试 / requeue
throw e;
}
}
}
6、ThreadPool 配置
@Configuration
public class ThreadPoolConfig {
@Bean
public ThreadPoolExecutor reportExecutor(ReportTaskService reportTaskService,
RabbitTemplate rabbitTemplate) {
int core = 2;
int max = 4;
long keepAlive = 60L;
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(5); // 小一点便于触发拒绝
return new ThreadPoolExecutor(
core,
max,
keepAlive,
TimeUnit.SECONDS,
queue,
Executors.defaultThreadFactory(),
new ReportRejectedHandler(reportTaskService, rabbitTemplate)
);
}
}
7、最终效果
数据库里：
- 有的任务 INIT → SUCCESS
- 有的任务 INIT → REJECTED → SUCCESS（通过 MQ 补偿）
这证明补偿流程完全闭环。
六、总结
线程池拒绝任务 → 自定义拒绝策略存证 + 投递 MQ MQ 消费者重新提交线程池 → 最终任务一定被执行成功。
这是最常用的异步任务高可用方案。
