RabbitMQ 的 Unacked 越攒越多，连接还被 blocked 了：一条忘了 ack 的消息引发的血案

周一早上 9 点半，我刚坐下，监控告警群里就刷屏了：RabbitMQ 内存使用触达水位线。紧跟着是应用组同事的一句“下单接口卡住了，日志里都是 connection blocked”。我打开 RabbitMQ 管理界面，看到那个熟悉的橙色/红色告警条——“Memory alarm”；再往下，队列列表里 order.pay.queue 那一行，Ready 是 0，Unacked 是 122,340。

Ready 是 0，说明消息都发给消费者了；Unacked 十二万，说明消费者拿了消息却没告诉 RabbitMQ“我处理完了”。消息就这样一直悬在半空，占着内存，直到把整个 Broker 撑到限流。这类问题的关键词是：手动 ack、prefetch、水位流控。下面是完整的排查记录。

一、先把三个数字看懂：Ready / Unacked / Total

RabbitMQ 队列里的消息有三种状态，管理界面（Queues 页）也是这么分的：

• Ready：在队列里，等着被投递给消费者（还没发出去）；
• Unacked：已经投递给某个消费者，但消费者还没有 ack（也没有 nack/reject）；
• Total = Ready + Unacked。

三种典型形态，对应不同的病因：

先记住这条核心规则（官方文档里写得很清楚）：手动确认模式下，消息在被 ack 之前一直保持“已投递未确认”，如果消费者所在的通道（channel）或连接关闭，这些消息会被自动重新入队（requeue）。所以只要消费者不断线也不 ack，这些消息就永远悬着。

二、排查路线图

三、动手：rabbitmqctl 一路查下去

第 1 步：list_queues——先看队列的三个数

$ rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name	messages_ready	messages_unacknowledged	consumers
order.pay.queue	0	122340	4
order.timeout.queue	18	0	2
sms.send.queue	0	35	3
order.pay.dlq	1207	0	0

输出是制表符分隔的。读法：order.pay.queue Ready = 0，Unacked = 122340，consumers = 4——四个消费者都在，但共有十二万条消息处于“未确认”；旁边的 order.pay.dlq（死信队列）里积了 1207 条，也是个信号，后面会说。想加更多列可以带 state、memory、message_bytes：

$ rabbitmqctl list_queues name messages_unacknowledged memory message_bytes_unacknowledged
name	messages_unacknowledged	memory	message_bytes_unacknowledged
order.pay.queue	122340	1893742592	1674823311

这行说明：这个队列一个就占了约 1.8 GB 内存，其中未确认消息的消息体合计约 1.6 GB。平均每条消息约 13KB——不小，说明我们的消息体里塞了不少东西（这本身也值得复盘）。“Unacked 涨的不只是个数，还有内存。”

第 2 步：list_channels——是谁占着这些 Unacked

$ rabbitmqctl list_channels connection name messages_unacknowledged prefetch_count consumer_count
Listing channels ...
connection	name	messages_unacknowledged	prefetch_count	consumer_count
10.0.3.21:41822 -> 10.0.0.31:5672	10.0.3.21:41822 -> 10.0.0.31:5672 (1)	30112	0	1
10.0.3.22:38160 -> 10.0.0.31:5672	10.0.3.22:38160 -> 10.0.0.31:5672 (1)	31005	0	1
10.0.3.23:40912 -> 10.0.0.31:5672	10.0.3.23:40912 -> 10.0.0.31:5672 (1)	30877	0	1
10.0.3.24:37744 -> 10.0.0.31:5672	10.0.3.24:37744 -> 10.0.0.31:5672 (1)	30346	0	1

关键信息：prefetch_count = 0。官方文档里写得清楚：prefetch 值限制的是一个通道上未确认消息的最大数量，0 表示不限制。也就是说，RabbitMQ 会把队列里几乎所有消息都推给这四个消费者，每个消费者手里攥着 3 万条“待确认”的消息，压在客户端内存里，又同时压着 Broker 的内存。

再看这几个通道的 unacked 数量：四个都在三万左右，且没有下降趋势（隔几分钟再执行一次数字不变或缓慢上升）。这说明消费者并不是“处理得慢”，而是压根没在 ack，或者处理线程已经卡死。

第 3 步：status 与 alarms——是不是触发了资源告警

$ rabbitmqctl status | sed -n '/Alarms/,/^$/p;/Memory/,/^$/p' | head -30
Alarms

Memory alarm on node rabbit@mq-node-1

Memory

Total memory used: 6.71 gb
Calculation strategy: rss
Memory high watermark setting: 0.4 of available memory, computed to: 6.4 gb

queue_procs: 3.02 gb (45.02 %)
binary: 2.18 gb (32.49 %)
...
$ rabbitmq-diagnostics memory_breakdown --unit "MB" | head -6
queue_procs: 3093.5 mb (45.02%)
binary: 2232.1 mb (32.49%)
connection_other: 210.3 mb (3.06%)
...

输出说明了三件事：（版本不同，格式和默认值略有差异，以你环境的实际输出为准）

• 存在 Memory alarm，节点已用内存 6.71 GB，超过水位线 6.4 GB（可用内存的 0.4）。官方当前文档写的默认值约为可用内存的 60%，较老版本是 40%——所以不同版本的水位线不一样，一定要以 rabbitmqctl status 输出的 Memory high watermark setting 为准；
• 内存大头是 queue_procs（队列进程，45%）和 binary（消息体，32%）——完全符合“堆了一堆未确认消息”的画像；
• 官方文档强调：水位线不是硬上限，它只是限流的触发点。超过之后 Broker 会阻塞发布，但内存仍可能继续涨一些，所以别把水位线调到接近物理内存。

磁盘方面也要确认：

$ rabbitmq-diagnostics check_local_alarms
Node rabbit@mq-node-1 reported alarm(s): memory
$ df -h /var/lib/rabbitmq
Filesystem      Size  Used Avail Use% Mounted on
/dev/vdb1       200G  118G   82G  59% /var/lib/rabbitmq
$ rabbitmqctl eval 'rabbit_disk_monitor:get_disk_free_limit().'
50000000

磁盘剩余 82 GB，disk_free_limit 是默认的 50MB（50000000 字节）——官方文档写明：磁盘剩余空间低于这个限制（默认 50MB）时会触发磁盘告警并阻塞所有生产者；并建议把它设成与机器内存大小相当，以便内存压力下被换页到磁盘的数据有地方放。这里磁盘没问题，问题在内存。

第 4 步：list_connections——谁被 blocked 了

$ rabbitmqctl list_connections name peer_host state channels
Listing connections ...
name	peer_host	state	channels
10.0.4.11:50322 -> 10.0.0.31:5672	10.0.4.11	blocked	1
10.0.4.12:44118 -> 10.0.0.31:5672	10.0.4.12	blocked	1
10.0.3.21:41822 -> 10.0.0.31:5672	10.0.3.21	running	1
10.0.3.22:38160 -> 10.0.0.31:5672	10.0.3.22	running	1

状态 blocked 的是订单服务（生产者，10.0.4.x）的连接；running 的是消费者（10.0.3.x）。这正好符合官方文档的描述：触发资源告警时，Broker 会阻塞发布消息的连接，消费连接照常工作（客户端也能收到 connection.blocked 通知）。集群里只要有一个节点触发告警，所有节点都会阻塞发布连接。所以症状是“生产者接口卡住，消费者看起来没事”，很有迷惑性。

四、定位到代码：忘记 ack 的那条分支

消费者用的是 Spring AMQP，手动 ack 模式（acknowledge-mode: manual）。看代码：

@RabbitListener(queues = "order.pay.queue")
public void onMessage(Message message, Channel channel) throws IOException {
    long tag = message.getMessageProperties().getDeliveryTag();
    PayEvent evt = JSON.parseObject(message.getBody(), PayEvent.class);
    if (evt.getAmount() == null) {
        log.warn("invalid event, skip: {}", evt.getOrderNo());
        return;                                   // ← 直接 return，没 ack 也没 nack！
    }
    try {
        orderService.handlePay(evt);
        channel.basicAck(tag, false);             // 正常路径才 ack
    } catch (BizException e) {
        log.error("biz error", e);                // ← 异常路径：既不 ack 也不 nack
    }
}

两个分支都漏了确认：“非法消息直接 return”和“业务异常只打日志”。这些消息就一直处于 Unacked，直到通道关闭才会被重新入队。我们的消费者连接长期存活，通道不关，Unacked 就只增不减。再加上 prefetch=0（不限流），RabbitMQ 一次性把大量消息推了过去，问题瞬间被放大。

到这里，因果链闭合：忘记 ack + 无限 prefetch → Unacked 堆积占内存 → 触发内存水位告警 → Broker 阻塞发布连接 → 订单接口卡住。

另一个同类隐患：消费者确认超时。RabbitMQ 有个保护机制，默认 30 分钟内消费者没有 ack，通道会以 PRECONDITION_FAILED 异常关闭（每分钟检查一次），消息随之重新入队。它能兜住“忘记 ack 且一直不断线”的最坏情况，但你会在日志里看到这样的异常，随后同样的消息又被重新投递：

Channel error on connection ... : operation none caused a channel exception precondition_failed:
delivery acknowledgement on channel 1 timed out. Timeout value used: 1800000 ms.
This timeout value can be configured, see consumers doc guide to learn more

这条错误的出现，就是你“至少有一个消息 30 分钟没被确认”的铁证，值得给它加日志告警。

五、止血与修复

止血：先把水位降下来，让发布恢复

1. 让 Unacked 释放：滚动重启消费者（或只关闭其通道）。通道/连接关闭后，未确认消息会自动重新入队，变成 Ready，再由修复后的消费者正常消费。注意：这会造成这批消息被重复投递，消费者必须幂等（见上一篇《MQ 重复与丢失》）；
2. 先修复代码再重启，否则重启后又会走到同样的分支再次堆积；
3. 如果内存告警一时下不来，且确认磁盘充足，可以临时上调水位线争取时间（不要调得太高，防止 OOM）：

# 临时调整（重启后失效）。绝对值更适合容器/虚机环境
$ rabbitmqctl set_vm_memory_high_watermark absolute "8GB"
Setting memory threshold on rabbit@mq-node-1 to 8000000000 bytes ...

# 让 Unacked 释放：只重启消费者服务；或在管理界面 Connections → 选中消费者连接 → Close
$ rabbitmqctl close_connection "10.0.3.21:41822 -> 10.0.0.31:5672" "unacked cleanup"
Closing connection ...

观察恢复：

$ watch -n 5 'rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers | grep order.pay.queue'
order.pay.queue	95220	800	4
order.pay.queue	61842	800	4
order.pay.queue	20415	800	4
order.pay.queue	0	40	4

消费者重启并使用合理的 prefetch（这里是 200，4 个消费者最多共约 800 条未确认）之后，Ready 逐渐被消化到 0，Unacked 维持在与 prefetch 相当的小值。内存告警解除，被 blocked 的生产者连接恢复为 running。

根治 1：所有分支必须 ack 或 nack

@RabbitListener(queues = "order.pay.queue")
public void onMessage(Message message, Channel channel) throws IOException {
    long tag = message.getMessageProperties().getDeliveryTag();
    try {
        PayEvent evt = JSON.parseObject(message.getBody(), PayEvent.class);
        if (evt.getAmount() == null) {
            log.warn("invalid event, reject: {}", evt.getOrderNo());
            channel.basicReject(tag, false);       // 非法消息：不重新入队 → 进入死信队列
            return;
        }
        orderService.handlePay(evt);
        channel.basicAck(tag, false);
    } catch (BizRetryableException e) {
        channel.basicNack(tag, false, true);       // 可恢复的错误：重新入队（注意要有上限，防死循环）
    } catch (Exception e) {
        log.error("consume failed, to DLQ", e);
        channel.basicNack(tag, false, false);      // 不可恢复：进死信队列，人工处理
    }
}

骨架是：try/catch 覆盖所有路径，每条路径都以 ack、nack 或 reject 结尾。一个提醒：nack(requeue=true) 会让消息立刻回到队列头部，容易造成无限重投的“毒消息死循环”，把 CPU 和日志打满。要么限制重试次数（在消息头里记录次数，超限进死信），要么直接 requeue=false 走死信队列，再由补偿任务延迟重投。

用 Spring AMQP 的话，更省心的方式是使用 acknowledge-mode: auto（容器根据方法是否抛异常自动 ack/nack）配合重试策略：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto        # 方法正常返回 → ack；抛异常 → nack
        prefetch: 200                 # 每个消费者最多同时持有 200 条未确认
        concurrency: 4
        max-concurrency: 8
        default-requeue-rejected: false   # 异常后不重新入队，进入死信队列
        retry:
          enabled: true               # 容器内本地重试（不占用 broker 往返）
          max-attempts: 3
          initial-interval: 1000
          multiplier: 2.0

根治 2：设置合理的 prefetch（basic.qos）

prefetch 是消费端的流控阀门：限制每个通道上未确认消息的数量。官方文档给出的建议是：100 到 300 之间通常能得到最佳吞吐，同时不至于压垮消费者；prefetch=1 太保守，会显著降低吞吐；0 表示不限制，则是本次事故的帮凶。选值的思路：

• 处理耗时短（毫秒级）、追求吞吐：100~300；
• 处理耗时长（秒级以上）、想让消息在多个消费者间均衡：调小，比如 5~20，避免一个消费者“囤货”而别的空闲；
• 公式参考：prefetch ≈ 目标吞吐 × 单条平均耗时 × 安全系数。管理界面的 Consumer capacity（消费能力）指标低于 100%，说明可以增加消费者或提高 prefetch。

// 原生 Java 客户端写法
channel.basicQos(200);                 // 每个 consumer 最多 200 条未确认（不写 global 时按消费者维度）
channel.basicConsume("order.pay.queue", false, consumer);   // autoAck=false

根治 3：水位与磁盘配置、监控

# rabbitmq.conf —— 用绝对值，容器/虚机里比相对值更可控
vm_memory_high_watermark.absolute = 6GB
disk_free_limit.absolute = 8GB        # 建议不低于内存大小，给换页数据留空间
consumer_timeout = 1800000            # 消费者确认超时，默认 30 分钟（毫秒）；按业务耗时调整，勿轻易调到很低

还有几条客户端侧的建议：生产连接与消费连接分开（官方也建议，这样生产者被 blocked 时不影响消费者继续消费、释放压力）；生产端使用 publisher confirms 并处理 connection.blocked 通知，被阻塞时给用户快速失败的响应而不是无限等待。

监控告警建议：

• 队列维度：messages_unacknowledged（绝对值 + 增速）、messages_ready、consumers 数、Consumer capacity；
• 节点维度：内存使用率相对水位线的比例（达到 70% 就预警，别等触发再喊）、磁盘剩余、文件描述符和 socket；
• 连接维度：blocked/blocking 状态连接数；
• 死信队列：消息数 > 0 就告警。

六、顺手说说死信队列与消息 TTL

前面 order.pay.dlq 里的 1207 条不是凭空来的。RabbitMQ 的死信交换机（DLX）会在四种情况下接收被“淘汰”的消息（官方文档）：消费者 nack/reject 且 requeue=false（reason=rejected）、消息 TTL 过期（expired）、队列长度超限被挤掉（maxlen）、以及 quorum queue 里超过投递次数上限（delivery_limit）。死信消息会带上 x-death 头，记录来源队列、原因、次数、时间。

# 推荐用 policy 配置 DLX：不用改代码、不用重新声明队列
$ rabbitmqctl set_policy pay-dlx "^order\.pay\.queue$" \
    '{"dead-letter-exchange":"dlx.exchange","dead-letter-routing-key":"order.pay.dead","message-ttl":600000}' \
    --apply-to queues
Setting policy "pay-dlx" for pattern "^order\.pay\.queue$" to "{...}" with priority "0" for vhost "/" ...
$ rabbitmqctl list_policies
vhost	name	pattern	apply-to	definition	priority
/	pay-dlx	^order\.pay\.queue$	queues	{"dead-letter-exchange":"dlx.exchange","dead-letter-routing-key":"order.pay.dead","message-ttl":600000}	0

上面同时给队列设置了 10 分钟的消息 TTL（message-ttl，毫秒）：在队列里待够 10 分钟没被消费的消息会过期，并被送往死信交换机。使用 TTL 有两点要注意：

• TTL 只对还在队列里（Ready）的消息起作用；已投递给消费者的 Unacked 消息不会因 TTL 过期，这类消息要靠 ack 和上面的 consumer_timeout 兜底；
• 官方文档指出，过期消息只会在到达队列头部时才被真正丢弃，所以排在前面的“长寿”消息会拖住后面已过期的消息，占着内存。如果需要严格的时间语义，要么统一每条消息的 TTL，要么配合队列长度限制。

提醒一点：官方文档明确说，标准的死信机制没有 publisher confirms，在集群里目标队列不可用时消息可能丢失；quorum queue 提供更安全的至少一次死信语义。死信队列上要有消费者或补偿任务，别把它当垃圾桶——死信队列不处理，就是把问题从“堆积”换成“隐藏”。

七、复盘

一句话备忘：Unacked 高 = 消费者拿了没确认。list_queues 看数、list_channels 看谁占着与 prefetch、status 看告警水位、list_connections 看 blocked；代码里保证每条路径都 ack/nack，prefetch 别设 0，DLQ 要有人管。

那天故障恢复后，我在评审会上说了句“RabbitMQ 是个很有礼貌的服务：你不说收到了，它就一直替你保管着”。测试同事接了一句：“所以我们的消费者是那种收了快递不签字、还把快递一直堆在门口的邻居。”——没错。而门口快递太多，物业（Broker）就只好把大门关了（blocked），谁也别想再往里送。

参考资料

• RabbitMQ 官方文档：Consumer Acknowledgements and Publisher Confirms——手动/自动确认模式、prefetch（basic.qos，0 为不限、100~300 的建议值）、通道关闭后未确认消息自动重新入队。
• RabbitMQ 官方文档：Memory Threshold and Limit——vm_memory_high_watermark（绝对值/相对值）、触发后阻塞发布者、rabbitmqctl set_vm_memory_high_watermark，水位线不是硬上限。
• RabbitMQ 官方文档：Resource Alarms——内存与磁盘告警阻塞发布连接、消费连接不受影响、集群范围告警与 connection.blocked 通知。
• RabbitMQ 官方文档：Free Disk Space Alarms——disk_free_limit 默认 50MB、set_disk_free_limit 及建议取值。
• RabbitMQ 官方文档：Consumers——消费者确认超时（默认 30 分钟，超时通道以 PRECONDITION_FAILED 关闭）与 Consumer capacity 指标。
• RabbitMQ 官方文档：Dead Letter Exchanges——四种死信原因（rejected、expired、maxlen、delivery_limit）、policy 与队列参数配置、x-death 头及安全性说明。
• RabbitMQ 官方文档：Time-To-Live and Expiration——队列级/消息级 TTL、过期消息仅在队首被丢弃的行为。