rabbitmq消息积压：如何快速排查与处理

线上消息队列积压了，消费者处理不过来，老板在群里@你——这时候怎么办？

今天聊点实际的，从判断积压到快速处理，讲点能直接上手的东西。

怎么判断消息积压了

RabbitMQ

# 查看队列消息数量
rabbitmqctl list_queues name messages messages_ready messages_unacked

# 输出大概长这样
# name messages messages_ready messages_unacked
# order 15234 15200 34

关键看这几个数字：

• messages – 队列里总共有多少消息
• messages_ready – 等待消费的消息（积压的主力）
• messages_unacked – 已投递但还没确认的

一般来说，messages_ready 超过几千就该关注了，上万就是严重积压。

Kafka

# 查看消费 lag（落后多少）
kafka-consumer-groups.sh –bootstrap-server localhost:9092 \\
–group my-consumer-group \\
–describe

# 输出大概长这样
# GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
# my-consumer order 0 52341 67589 15248

LAG 就是积压量。LAG 几百几千问题不大，上万就得处理了。

为什么会积压

常见原因就几种：

1. 消费者挂了
最常见。消费者进程没了，但生产者还在往里扔消息。

2. 消费者变慢了
比如数据库慢查询、对外接口超时。消息在那儿等着，时间一长就堆起来了。

3. 消费者数量不对
分区数固定，但消费者少了或者消费者不干活了。

4. 消费逻辑有死循环或者阻塞
消费端卡在某个地方，不消费也不报错。

快速处理方案

方案一：临时增加消费者（推荐）

如果是 Kafka，把消费者实例扩几个：

# Docker Compose 临时扩容
services:
consumer:
deploy:
replicas: 5 # 从 1 临时改成 5

RabbitMQ 的话，多启动几个消费者实例就行。

注意：Kafka 一个分区只能被一个消费者消费，所以扩容数量别超过分区数。

方案二：提高消费并行度

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
ConsumerFactory<String, String> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.setConcurrency(10); // 默认1，临时调大成10
return factory;
}

RabbitMQ 的话，调整 prefetch 数量：

factory.setPrefetchCount(50); // 默认250，太大了会积压太久，太小了吞吐量上不去

方案三：跳过积压消息，先消费最新的

有时候积压太多了，处理完要很久。这时候可以先跳过旧消息：

Kafka 方案：重置 offset 到最新

# 把消费者组的 offset 重置到最新
kafka-consumer-groups.sh –bootstrap-server localhost:9092 \\
–group my-consumer-group \\
–topic order \\
–reset-offsets –to-latest \\
–execute

注意：这个方案会丢消息，只有在业务能接受丢消息的时候才用。

方案四：临时写个脚本消费掉

直接写个简单脚本，扫队列然后批量处理：

@Service
public class FastConsumerService {

@Autowired
private RabbitTemplate rabbitTemplate;

public void consumeAndDiscard(String queueName, int limit) {
for (int i = 0; i < limit; i++) {
Message message = rabbitTemplate.receive(queueName, 1000);
if (message == null) {
break;
}
// 空消费，直接确认。积压太多可以用这个临时方案
// 但记得后续要补偿处理这些消息！
}
}
}

再次强调：空消费会丢消息，后续一定要有补偿机制。

排查积压的根本原因

处理完积压只是治标，找到根因才是治本。

看消费者日志

消费者有没有报错？有没有超时？有没 BLOCKED？

# RabbitMQ 查看连接状态
rabbitmqctl list_connections pid channels

# 查看消费者是否正常
rabbitmqctl list_consumers queue_name

看监控

RabbitMQ 管理界面：http://localhost:15672

Kafka 使用 Kafka Eagle 或者 JMX 监控：

# 开启 JMX
KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote=true"

常见坑

数据库连接池打满：消费者查询数据库，连接池耗尽，所有消费线程都在等连接。

对外接口超时：消费者调第三方接口，接口响应慢，消费速度断崖式下降。

内存泄漏：消费者 OOM 了但没死透，勉强撑着但几乎不消费。

怎么防止积压

几点经验：

总结

积压不可怕，怕的是没监控、没预案。下次遇到这种情况，希望你能从容应对。