RabbitMQ 消息队列积压告警：我用消费者并发 + 死信队列把积压从 10 万条清空到 0-平芜编程栈

RabbitMQ 消息队列积压告警：我用消费者并发 + 死信队列把积压从 10 万条清空到 0

凌晨两点，手机震了。

Prometheus 告警：RabbitMQorder_queue消息积压超过 10 万条，消费速率只有 120 msg/s，生产速率 850 msg/s。再不管，队列内存撑爆只是时间问题。

我之前写过 Kafka 消费延迟的排查，但 RabbitMQ 的积压治理完全是另一套逻辑。这次用了 40 分钟把积压清空到 0，核心就两件事：消费者并发提上去，死信队列把烂消息隔离出去。

问题背景：为什么 Kafka 那套不顶用了

我们的订单系统用 RabbitMQ 做异步解耦，上游订单服务生产消息，下游库存、物流、财务三个服务消费。平时队列深度稳定在 500 条以内，但那天晚上突然炸了。

先说一个认知误区：很多人觉得消息队列积压了，加消费者就行了。这在 Kafka 里确实好使，因为分区机制天然支持横向扩容。但 RabbitMQ 的队列是单线程的，一个队列只能被一个消费者串行处理，加机器没用，得加并发通道。

另外，RabbitMQ 的默认prefetch_count是无限，消费者一次拉走所有消息，如果处理一条消息卡住，整个通道都堵死。这和 Kafka 的拉取模式完全不同。

排查：5 分钟定位瓶颈

Ready：10.2 万条消息待消费
Unacked：187 条（消费者已拉取但未确认）
Consumer utilisation：12.3%（消费者利用率极低）

Unack 只有 187，但 Ready 有 10 万，说明消费者没卡死，就是处理速度太慢。

SSH 到消费服务节点，先看消费者进程：

rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers# order_queue 102347 187 3

只有 3 个消费者，但每个消费者是单线程串行处理。进一步看消费者配置：

rabbitmqctl list_consumers|greporder_queue# 3 个 consumer，channel 都是同一个连接

问题找到了：应用虽然启动了 3 个消费者实例，但每个实例都是单线程，而且共用同一个连接，实际并发度是 3，根本扛不住 850 msg/s 的生产速率。

还有一个隐藏问题：积压的消息里混了一批异常消息。下游物流系统调用第三方 API 超时，消息处理失败进入重试，每次重试 30 秒，直接把消费者拖死了。

解决方案：三层并发 + 死信队列

第一层：提高消费者并发度

把每个消费者实例从单线程改成多线程，并且每个线程独占一个 channel。Spring AMQP 里的配置是这样的：

@BeanpublicSimpleRabbitListenerContainerFactoryrabbitListenerContainerFactory(ConnectionFactoryconnectionFactory){SimpleRabbitListenerContainerFactoryfactory=newSimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);// 每个消费者实例并发 10 个线程factory.setConcurrentConsumers(10);factory.setMaxConcurrentConsumers(20);// 每次预取 50 条，避免单个消费者堆积factory.setPrefetchCount(50);// 手动 ACK，确保消息不丢失factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);returnfactory;}

关键参数解释：

concurrentConsumers=10：每个消费者容器启动 10 个并发线程
maxConcurrentConsumers=20：负载高时自动扩容到 20 个
prefetchCount=50：每个消费者一次只拉 50 条，处理完再拉，避免某个消费者饿死

改完重启，消费者从 3 个变成 30 个（3 个实例 × 10 线程），并发度直接翻 10 倍。

第二层：死信队列隔离异常消息

光加并发不够，那批超时的异常消息还在反复重试，只要有一条卡住，整个线程就废了。需要把处理失败的消息踢出去，别让一颗老鼠屎坏了一锅粥。

RabbitMQ 的死信交换机（DLX）机制正好干这个。思路是：主队列消费失败的消息，超过最大重试次数后，自动路由到死信队列，避免阻塞正常消费。

@BeanpublicQueueorderQueue(){Map<String,Object>args=newHashMap<>();// 消息在队列存活超过 60 秒仍未消费，进入死信args.put("x-message-ttl",60000);// 最大重试 3 次args.put("x-dead-letter-exchange","order.dlx.exchange");args.put("x-dead-letter-routing-key","order.dlx.routing.key");returnQueueBuilder.durable("order_queue").withArguments(args).build();}@BeanpublicQueueorderDlxQueue(){returnQueueBuilder.durable("order_dlx_queue").build();}@BeanpublicDirectExchangeorderDlxExchange(){returnnewDirectExchange("order.dlx.exchange");}@BeanpublicBindingorderDlxBinding(){returnBindingBuilder.bind(orderDlxQueue()).to(orderDlxExchange()).with("order.dlx.routing.key");}

消费端加上重试和死信逻辑：

@RabbitListener(queues="order_queue",containerFactory="rabbitListenerContainerFactory")publicvoidhandleOrder(Messagemessage,Channelchannel)throwsIOException{longdeliveryTag=message.getMessageProperties().getDeliveryTag();try{// 业务处理processOrder(newString(message.getBody()));channel.basicAck(deliveryTag,false);}catch(Exceptione){// 重试次数从消息头里取IntegerretryCount=(Integer)message.getMessageProperties().getHeaders().getOrDefault("x-retry-count",0);if(retryCount>=3){// 超过 3 次，拒绝消息并送入死信队列channel.basicReject(deliveryTag,false);}else{// 重新发布，retry count +1message.getMessageProperties().setHeader("x-retry-count",retryCount+1);channel.basicAck(deliveryTag,false);rabbitTemplate.send("order_queue",message);}}}

这样异常消息最多重试 3 次，之后就进死信队列，主队列的消费通道彻底解放。

第三层：死信队列的补偿消费

死信队列里的消息不能不管，需要人工或定时任务补偿处理。我加了一个简单的补偿消费者，专门处理死信：

@RabbitListener(queues="order_dlx_queue",concurrency="3-5")publicvoidhandleDlx(Messagemessage,Channelchannel)throwsIOException{longdeliveryTag=message.getMessageProperties().getDeliveryTag();try{// 死信补偿：记录日志 + 通知人工介入log.warn("死信消息补偿: {}",newString(message.getBody()));// 落库到补偿表，后续人工处理或重试dlxCompensationService.save(message);channel.basicAck(deliveryTag,false);}catch(Exceptione){// 补偿失败，直接丢弃，避免死信队列也积压channel.basicReject(deliveryTag,false);}}