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 分钟定位瓶颈登录 RabbitMQ Management UI先看三个指标Ready10.2 万条消息待消费Unacked187 条消费者已拉取但未确认Consumer utilisation12.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 个 consumerchannel 都是同一个连接问题找到了应用虽然启动了 3 个消费者实例但每个实例都是单线程而且共用同一个连接实际并发度是 3根本扛不住 850 msg/s 的生产速率。还有一个隐藏问题积压的消息里混了一批异常消息。下游物流系统调用第三方 API 超时消息处理失败进入重试每次重试 30 秒直接把消费者拖死了。解决方案三层并发 死信队列第一层提高消费者并发度把每个消费者实例从单线程改成多线程并且每个线程独占一个 channel。Spring AMQP 里的配置是这样的BeanpublicSimpleRabbitListenerContainerFactoryrabbitListenerContainerFactory(ConnectionFactoryconnectionFactory){SimpleRabbitListenerContainerFactoryfactorynewSimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);// 每个消费者实例并发 10 个线程factory.setConcurrentConsumers(10);factory.setMaxConcurrentConsumers(20);// 每次预取 50 条避免单个消费者堆积factory.setPrefetchCount(50);// 手动 ACK确保消息不丢失factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);returnfactory;}关键参数解释concurrentConsumers10每个消费者容器启动 10 个并发线程maxConcurrentConsumers20负载高时自动扩容到 20 个prefetchCount50每个消费者一次只拉 50 条处理完再拉避免某个消费者饿死改完重启消费者从 3 个变成 30 个3 个实例 × 10 线程并发度直接翻 10 倍。第二层死信队列隔离异常消息光加并发不够那批超时的异常消息还在反复重试只要有一条卡住整个线程就废了。需要把处理失败的消息踢出去别让一颗老鼠屎坏了一锅粥。RabbitMQ 的死信交换机DLX机制正好干这个。思路是主队列消费失败的消息超过最大重试次数后自动路由到死信队列避免阻塞正常消费。BeanpublicQueueorderQueue(){MapString,ObjectargsnewHashMap();// 消息在队列存活超过 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(queuesorder_queue,containerFactoryrabbitListenerContainerFactory)publicvoidhandleOrder(Messagemessage,Channelchannel)throwsIOException{longdeliveryTagmessage.getMessageProperties().getDeliveryTag();try{// 业务处理processOrder(newString(message.getBody()));channel.basicAck(deliveryTag,false);}catch(Exceptione){// 重试次数从消息头里取IntegerretryCount(Integer)message.getMessageProperties().getHeaders().getOrDefault(x-retry-count,0);if(retryCount3){// 超过 3 次拒绝消息并送入死信队列channel.basicReject(deliveryTag,false);}else{// 重新发布retry count 1message.getMessageProperties().setHeader(x-retry-count,retryCount1);channel.basicAck(deliveryTag,false);rabbitTemplate.send(order_queue,message);}}}这样异常消息最多重试 3 次之后就进死信队列主队列的消费通道彻底解放。第三层死信队列的补偿消费死信队列里的消息不能不管需要人工或定时任务补偿处理。我加了一个简单的补偿消费者专门处理死信RabbitListener(queuesorder_dlx_queue,concurrency3-5)publicvoidhandleDlx(Messagemessage,Channelchannel)throwsIOException{longdeliveryTagmessage.getMessageProperties().getDeliveryTag();try{// 死信补偿记录日志 通知人工介入log.warn(死信消息补偿: {},newString(message.getBody()));// 落库到补偿表后续人工处理或重试dlxCompensationService.save(message);channel.basicAck(deliveryTag,false);}catch(Exceptione){// 补偿失败直接丢弃避免死信队列也积压channel.basicReject(deliveryTag,false);}}死信消息的处理原则是不能再重试回主队列否则又会堵死。要么人工处理要么走降级流程。效果40 分钟从 10 万条清空到 0改完配置重启后RabbitMQ 面板的变化时间点Ready 消息数消费速率消费者数02:00102,347120 msg/s302:1587,2101,200 msg/s3002:3034,5001,850 msg/s3002:400850 msg/s30积压清空的那一刻消费速率和生产速率持平队列深度稳定为 0。更关键的是死信队列把 23 条异常消息隔离了出去主队列再也不被卡住。写在最后这次排查和 Kafka 那次最大的不同是 RabbitMQ 的队列模型决定了你不能靠堆机器解决积压得从并发通道和消息隔离两个维度动手。三个经验prefetchCount 必须设上限。默认无限的话一个消费者可能把所有消息都拉走其他消费者干看着。死信队列不是可选项是必选项。任何生产环境的 RabbitMQ 都必须配死信否则异常消息就是定时炸弹。监控不要只看队列深度。消费者利用率Consumer utilisation和 Unacked 数比 Ready 数更能反映瓶颈在哪。如果这篇文章对你有帮助欢迎收藏。下次遇到 RabbitMQ 积压至少知道该改哪几个参数了。