消息积压了怎么办：从止损到根治

凌晨被告警叫醒，队列积压几十万条。慌是没用的，按顺序处理。

第一步：确认是不是真积压

# RabbitMQ
rabbitmqctl list_queues name messages messages_ready messages_unacknowledged

# Kafka
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group

看 messages_ready（待消费）和 unacknowledged（已投递未确认）。如果 unack 很大而 ready 很小，说明消费者拿到了但不确认——可能是消费逻辑卡住了，不是量的问题。

Kafka 看 LAG 列，每小时的增量比总量更能说明趋势。

第二步：找原因

按概率排序：

1. 消费者挂了。看进程、看日志、看是不是 OOM 被杀
2. 消费变慢。下游接口变慢、数据库锁等待、外部 API 超时
3. 生产突增。上游批量任务、重试风暴、有人手动灌数据
4. 分区/队列不够。消费者数量超过分区数，多的在空转

第三步：止损

先扩消费者（如果是消费能力不足）：

docker compose up -d --scale worker=6

注意不能超过分区数，超过了也没用。

再降级非核心逻辑。比如消费时要发通知、写统计，这时候可以先跳过，只处理主流程，积压消化完再补。

必要时临时加速消费：把批量大小从 1 调到 100，批量入库代替逐条。这一步能提升 5~10 倍吞吐，但要小心失败重试的代价变大。

第四步：处理毒消息

如果某条消息一直失败、反复重投，会堵住整个队列。找出来扔进死信：

if retry_count > 3:
    channel.basic_publish(exchange="dlx", routing_key="failed", body=body)
    channel.basic_ack(delivery_tag=method.delivery_tag)
    return

死信队列要有人看，不然只是把问题藏起来了。

第五步：根治

积压只是症状，根因通常是：

• 消费逻辑里有同步阻塞调用（HTTP 请求、大文件处理）→ 改成异步或加超时
• 数据库是瓶颈 → 批量写、加索引、分表
• 单条消息太大 → 消息里只放 id，消费者回查
• 没有背压 → 生产端限流，或者监控 LAG 自动扩容

事后

加两个监控：LAG 持续 5 分钟超过阈值告警；消费耗时 P99 超过 1 秒告警。这两个能提前发现大部分问题，不用等积压到几十万才知道。

还有一条：消费必须幂等。积压处理过程中必然会有重复消费，如果不幂等，修完积压会发现数据全乱了。