消息队列排查 列表全部文章
消息队列排查

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

管理界面里 Unacked 涨到 12 万,内存水位告警、生产者连接被 blocked。用 rabbitmqctl list_queues / list_channels / list_connections 定位忘记 ack 的消费者,讲清 prefetch、内存与磁盘水位流控、死信与 TTL,并给出止血与根治配置。

更多

周一早上 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。

三种典型形态,对应不同的病因:

形态含义常见原因
Ready 高,Unacked 低消息发不出去或消费者太少没有消费者 / 消费者掉线 / 消费能力不足 / prefetch 限制吃满
Ready 低,Unacked 高(本次)消费者拿走了,但不 ack忘记 ack、处理卡死、prefetch 设得过大、autoAck 关闭却没确认
两者都高整体积压生产速度远大于消费速度,且消费者也慢

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

二、排查路线图

生产者持续 publish 队列 order.payReady 0 / Unacked 12 万 消费者(手动 ack)处理后【忘了 ack】 Broker 内存持续上涨未确认消息全在内存/磁盘 内存超过 vm_memory_high_watermark(或磁盘低于 disk_free_limit) → 触发资源告警(alarm),Broker 阻塞所有发布连接 生产者连接状态:blocked 发布阻塞,接口超时(消费连接不受影响) 排查顺序 ① list_queues看 Ready/Unacked与 consumers 数 ② list_channels哪个通道占着Unacked、prefetch ③ status / alarms是否触发内存/磁盘告警、水位多少 ④ list_connections哪些连接 blocked(生产者) ⑤ 看代码ack / prefetch异常路径

三、动手: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 提供更安全的至少一次死信语义。死信队列上要有消费者或补偿任务,别把它当垃圾桶——死信队列不处理,就是把问题从“堆积”换成“隐藏”。

七、复盘

做法为什么
看队列先分清 Ready / Unacked:list_queues name messages_ready messages_unacknowledged consumersReady 高是“发不出去/消费慢”,Unacked 高是“拿了没确认”,病因完全不同
手动 ack 时,每条代码路径都以 ack / nack / reject 结尾漏掉任何一个分支,消息就悬在 Unacked,直到通道关闭
设置 prefetch(basic.qos),常规 100~300,慢任务调小,绝不留 0限制单个通道囤积的未确认消息数,保护 Broker 和消费者内存
失败消息用 requeue=false + 死信队列,重试要有上限避免“毒消息”无限重入队形成死循环
内存/磁盘水位用绝对值配置,水位线预警在 70%水位线只是限流触发点,不是硬上限;触发后所有节点会阻塞发布
生产和消费用不同连接;生产端处理 connection.blocked 并配 publisher confirms被阻塞时消费者仍能消费释放压力,生产者也能快速失败
用 policy 配置 DLX 与 TTL;给 DLQ 加告警和补偿policy 可热更新,不必改代码;DLQ 无人管等于把问题藏起来
消费逻辑保持幂等关闭通道后 Unacked 会被重新入队,重复投递是必然

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

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

参考资料

— 感谢读到这里 —发现错误,或想补充?告诉我 →
TURN KNOWLEDGE INTO SOMETHING REAL

有想法,也想把它做出来?

从技术方案到上线交付,聊聊你的下一个项目。

聊聊需求

一起聊聊

搜索文章

输入关键词,按标题和摘要即时筛选。

浏览分类文章归档

阅读设置

字号
行宽
专注模式

专注模式会隐藏页头与侧栏,只留正文。设置保存在当前浏览器。

稍后阅读

保存在当前浏览器,留给下一次安静阅读。

总浏览 -- 次 · 总访客 -- 人 · 今日浏览 -- 次 · 今日访客 -- 人