RabbitMQ 消息堆积：未确认(Unacked)消息与队列积压导致内存告警 —— prefetch(QoS) 配置

日期：2026-09-09 ｜环境：Docker Compose ｜技术栈：RabbitMQ 3-management、Python pika 关键词：RabbitMQ、消息堆积、Unacked 未确认消息、prefetch、QoS、basic_qos、memory alarm 内存告警、 flow control 流控、basic_ack、basic_nack、StreamLostError、Connection reset by peer、消息队列

一、问题背景

一个数据处理系统通过 RabbitMQ 做异步削峰。生产者持续往队列 task.queue 投递任务，消费者取消息做耗时处理。某天发现：

管理台里队列的 Ready 与 Unacked 持续上涨，Unacked 尤其巨大；
rabbitmqctl list_queues 显示 messages_ready / messages_unacknowledged 不断膨胀；
一条告警 Memory alarm 被触发，RabbitMQ 进入 flow control 流控，生产者 publish 被阻塞、卡住甚至超时；
生产端报错、任务延迟越来越大，最终疑似把内存顶爆。

典型报错信息：

pika.exceptions.StreamLostError: Stream connection lost: ConnectionResetError(104, 'Connection reset by peer')

二、复现步骤

cd scenario/2026-09-09-0000
docker compose up -d --build

约 30~60 秒后观察队列积压：

docker compose exec rabbitmq rabbitmqctl list_queues name messages_ready messages_unacknowledged messages

复现输出（异常）：

name        messages_ready  messages_unacknowledged  messages
task.queue      367721             2550               370271   （30s）
task.queue      857636               0                857636   （70s，消费者已崩）

消费端日志（docker logs r40-consumer）：

consumer started (no QoS => unlimited prefetch, no ack => unacked piles up)
[consumer] got b'task-00000000-xxxxxx'... (NOT acked, unacked grows)
...
pika.exceptions.StreamLostError: Stream connection lost: ConnectionResetError(104, ...)

三、根本原因

根因是消费端两个缺陷叠加：

未设置 basic_qos(prefetch_count=N)（无 QoS）：RabbitMQ 默认无限预取，一次会把海量消息推给消费者，导致 messages_unacknowledged 瞬间暴涨。
收到消息后不调用 basic_ack（auto_ack=False 但从不确认）：消息永远处于 unacked，不会被移除。

结果：Unacked 巨大 → 内存走高 → 触发 memory alarm / flow control → 生产端 publish 阻塞。同时消费者被海量 unacked 压垮，连接被 broker 重置（StreamLostError），进程崩溃，队列堆积进一步放大。

常见误区：以为是内存不够去调 vm_memory_high_watermark，其实根因是消费端 QoS/ack 配置不当。

四、解决办法

核心：「限制预取 + 正确确认」。

修改 consumer.py：

# 1) 限制每次从 broker 预取的最大未确认消息数
ch.basic_qos(prefetch_count=50)

# 2) 手动确认（auto_ack=False 时处理完必须确认）
def on_message(ch, method, properties, body):
    try:
        # ... 业务处理 ...
        ch.basic_ack(delivery_tag=method.delivery_tag)      # 成功：确认
    except Exception as e:
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)  # 失败：nack / 进死信

ch.basic_consume(queue=QUEUE, on_message_callback=on_message, auto_ack=False)

重建验证：

docker compose up -d --build consumer

五、验证

重新观察队列，messages_unacknowledged 恒定等于 prefetch_count(50)，不再爆炸：

name        messages_ready  messages_unacknowledged  messages
task.queue      469439             50               469489   （40s）
task.queue      766629             50               766679   （70s）

消费端日志持续 ack、无 error：

consumer-fixed started (prefetch=50, manual ack on success)
[consumer-fixed] acked b'task-00002235-xxxxxx'...

消费者存活（修复前约 1 分钟内崩溃）：

r40-consumer  Up About a minute

无内存告警 / 流控：

docker compose logs rabbitmq | grep -i "memory_alarm\|flow control\|blocking"   # 无输出

内存占用：Total memory used: 0.3767 gb，远低于水位 6.87GB。

六、经验与避坑

「消息堆积 + 内存告警」优先查消费端 QoS 和 ack，别急着调内存水位。92% 的场景是 prefetch 无限 + 不确认。
prefetch_count 参数要按业务压测定：太小吞吐低、浪费往返；太大又回到"一次拉太多"。一般从几十到几百开始压测。
保证每条消息要么 ack、要么 nack。丢失确认 = 消息永远 unacked，会重复堆积。
失败消息 nack 策略：requeue=False 丢弃/进死信；requeue=True 重新入队，但要防毒消息死循环。
生产速度 > 消费速度时，即使修好 ack，messages_ready 仍会增长（本用例故意设置的不对称）。此时应水平扩容消费者（多实例）或提高单实例并发，而不是调大 prefetch。
判断 QoS 是否生效：看 list_queues 的 messages_unacknowledged 是否维持在 prefetch 附近恒定值。若一直涨，说明要么没设 prefetch，要么 ack 没调对。
规模化采集可用 rabbitmqctl list_queues name messages_ready messages_unacknowledged 一条命令看全量。

七、可复现命令汇总

# 复现异常
cd scenario/2026-09-09-0000 && docker compose up -d --build
docker compose exec rabbitmq rabbitmqctl list_queues name messages_ready messages_unacknowledged messages

# 看内存告警/流控
docker compose logs rabbitmq | grep -i "memory\|alarm\|flow"

# 修复后重建并验证
docker compose up -d --build consumer
docker compose exec rabbitmq rabbitmqctl list_queues name messages_ready messages_unacknowledged messages

# 清场
docker compose down -v