消息队列消费积压到打爆磁盘：我用Consumer Lag监控+阈值告警在5分钟内止血
凌晨1点，刚躺下没多久，手机就开始嗡嗡震。告警信息写的是"磁盘使用率超过85%"。
爬起来登录机器一看，好家伙，RabbitMQ的/var/lib/rabbitmq目录已经吃了70GB。消息队列的消费 Lag 不知道怎么就炸了，堆积的消息还在往磁盘里写。
说实话，之前我一直觉得消息队列"自带缓冲，不会出问题"。结果这次啪啪打脸。
这篇文章说说我是怎么用 Consumer Lag 监控 + 自动阈值告警，5分钟内把消费积压从临界状态拉回来的，以及我是怎么搭建这套监控体系的。
问题是怎么发生的
先说背景：线上跑着一个订单处理服务，用的 RabbitMQ 做异步下单。消费者是一个 Spring Boot 应用，消费端用了手动 ack。
出问题时我在做什么呢？——给消费端加了个新功能，上线前测试了3轮都正常，就直接发上去了。结果新功能里有个慢查询，把消费线程堵住了。消费速度 < 生产速度，积压就开始涨。
一开始我根本没注意到。告警是磁盘用率触发的，不是队列积压触发的——这是第一个坑。
第一步：先止血，把积压消掉
登录机器第一件事，看队列状态：
# 查看队列信息rabbitmqctl list_queues name messages messages_ready messages_unacknowledged# 查看消费者的 channelrabbitmqctl list_channels connection_name command_name model_name fc当时看到的情况：
messages_ready: 47万+messages_unacknowledged: 3000+- 消费速率
consumer_ms掉到了正常值的1/20
积压确实很严重。先把消费者停掉（不要直接kill，用rabbitmqctl stop_channel），然后开3个消费者实例并行消费，先把水位降下来。
# 批量重置消费 channelrabbitmqctleval'rabbit_channel:close(MyChannelPid, normal).'这步做完，队列里47万条消息开始以正常速度消费了。磁盘写入速度从爆满降到正常。
但这只是临时止血。如果不解决根本问题，下次还会炸。
第二步：Consumer Lag 监控怎么搭
消费积压的本质是：消费速度跟不上生产速度。所以监控的核心指标是Consumer Lag = messages_ready + messages_unacknowledged。
我用的是 Prometheus + RabbitMQ Exporter：
# docker-compose 片段rabbitmq-exporter:image:kbudde/rabbitmq-exporter:latestenvironment:RABBIT_URL:http://rabbitmq:15672RABBIT_USER:guestRABBIT_PASSWORD:guestports:-"9419:9419"然后在 Prometheus 告警规则里加这个：
groups:-name:rabbitmqrules:-alert:RabbitMQConsumerLagHighexpr:rabbitmq_queue_messages{queue_name="order_queue"}>10000for:2mlabels:severity:warningannotations:summary:"队列积压超过1万条"description:"队列 {{ $labels.queue_name }} 积压 {{ $value }} 条，请检查消费者状态"-alert:RabbitMQConsumerLagCriticalexpr:rabbitmq_queue_messages{queue_name="order_queue"}>50000for:1mlabels:severity:criticalannotations:summary:"队列积压超过5万条，接近磁盘上限"第三步：阈值是怎么算出来的
不是随便写1万和5万的。阈值要结合两个维度：
- 消费者消费一条消息的平均时间
- 磁盘剩余空间能扛多少积压
算一下：假设消费一条消息平均耗时 50ms，一个消费者线程每秒能处理 20 条。积压1万条，需要 500 秒（8分钟）才能消费完。如果这段时间内生产速度 > 消费速度，Lag 还会继续涨。
我的经验阈值：
- 消费 Lag > 1万条：发 warning 告警（可能只是临时抖动）
- 消费 Lag > 5万条：发 critical 告警（必须人工介入）
- 消费 Lag 持续增长超过5分钟：即使没到阈值也要告警
第四步：自动恢复脚本
光告警不够，还得有自动止血脚本。我的思路是：当 Lag 持续告警超过 N 分钟，自动扩容消费者。
#!/usr/bin/env python3importpikaimportrequestsimporttimefromdatetimeimportdatetime RABBITMQ_API="http://rabbitmq:15672/api/queues"QUEUE_NAME="order_queue"LAG_THRESHOLD=50000CHECK_INTERVAL=60# 秒defget_queue_messages():resp=requests.get(f"{RABBITMQ_API}/%2f/{QUEUE_NAME}",auth=('guest','guest'))data=resp.json()returndata['messages']defmain():last_alert_time=NonewhileTrue:msgs=get_queue_messages()print(f"[{datetime.now()}] 队列积压:{msgs}条")ifmsgs>LAG_THRESHOLD:iflast_alert_timeisNone:last_alert_time=time.time()eliftime.time()-last_alert_time>300:# 持续5分钟print("🚨 积压持续5分钟，触发自动扩容！")# 这里调用 K8s 或 Docker Swarm 扩容消费者# call_k8s_scale("order-consumer", replicas=5)last_alert_time=time.time()# 重置，避免重复触发else:last_alert_time=Nonetime.sleep(CHECK_INTERVAL)if__name__=="__main__":main()踩坑心得
坑1：消息积压不一定是消费者的问题
有时候积压是因为生产者发得太快（比如突发流量），有时候是 RabbitMQ 本身配置问题（内存限制、磁盘告警阈值）。定位问题时要先确认是哪个环节慢了。
坑2：手动 ack 模式下，消费慢会导致 unacked 堆积
这次我用的手动 ack，结果慢查询导致消息一直处于 unacked 状态。消费者处理完之后没有及时 ack，消息就卡在"消费中"状态。监控时要把messages_unacknowledged也加上。
坑3：不要只看 Lag 数值，要看 Lag 变化趋势
Lag=5000 可能是正常的（消费者刚启动），也可能是异常的（消费突然中断）。结合 Lag 增速一起看：
rate(rabbitmq_queue_messages{queue_name="order_queue"}[5m]) > 100如果5分钟内 Lag 增速超过100条/秒，说明生产速度远大于消费速度，要立刻告警。
写在最后
这次故障让我重新审视了消息队列的可观测性。之前只知道监控"队列长度"，其实 Consumer Lag 和 Lag 增速才是关键指标。
现在我给每个核心队列都配了4个监控：
- Lag 绝对值（阈值告警）
- Lag 增速（变化率告警）
- unacked 数量（消费阻塞告警）
- 磁盘使用率（兜底告警）
上线一周，已经捕获了两次潜在的 Lag 上涨，都是消费者线程池配置有问题。提前处理的感觉，比凌晨1点被炸醒好太多了。
如果你也在用 RabbitMQ，建议先查一下你的监控里有没有 Consumer Lag 这个指标。没有的话，这篇文章值得你花10分钟加上。