RabbitMQ消息积压排查问题分析之从指标定位到消费者扩容

比赛高峰时，判题结果从几秒瞬间变成几分钟，RabbitMQ控制台里的消息数一路飙升。这时候，最直觉的反应是什么？加消费者。但可别急着动手——如果瓶颈在数据库、容器启动、外部API或宿主机CPU上，那扩容只会让更多任务同时压向下游，最后把队列积压演变成一场数据库雪崩。

其实，消息积压从来不是单一故障。它表示的是：进入速率在一段时间内大于有效完成速率。要排查，得先确认消息是处于Ready状态还是Unacked状态，再把每条消息的处理阶段拆开，建立容量模型，最后才能决定是扩容、限流、降级，还是修复毒消息。下文以Spring Boot + RabbitMQ的异步判题任务为例，整理出一套可执行的定位与恢复方法。文中的公式用于估算，实际结论必须基于本系统指标和压测验证。

一、先判断是正常削峰还是持续性积压

首先，队列本来就是用来吸收短暂峰值的。峰值结束后，消费速率高于生产速率，消息数在业务允许时间内归零，这就是正常削峰。真正需要告警的是：

• messages_ready持续往上走，而且生产速率长期高于ACK速率；
• messages_unacknowledged一直处于高位，消费者已经取走消息但完成得很慢；
• 消费者数量在下降、频繁重连，或者不断重启；
• 最老消息的年龄超过了业务SLA，即使队列总量暂时不大；
• 积压下降的速度，不足以在下一次峰值到来前清空；
• 大量消息在主队列、重试队列与死信队列之间来回循环。

关键是，不要看见“队列有十万条”就急着判断严重程度。每条任务耗时5毫秒和30秒，含义完全不同；消息体200字节和2MB，对磁盘和网络的影响也天差地别。最老消息的年龄，通常比纯粹的队列深度更能反映用户体验。

事故一开始，先冻结关键现场：队列名、vhost、时间窗口、Ready、Unacked、publish/deliver/ack rate、消费者数、最老消息年龄、节点磁盘与内存告警、应用发布记录、下游健康状态。别一边随意改prefetch，一边把基线给丢了。

二、读懂Ready、Unacked与速率组合

RabbitMQ里常用的三个数量是：

• Ready：消息在队列中等待投递，还没分配给消费者；
• Unacked：已经投递给消费者，但Broker还没收到确认；
• Total：通常可以理解为Ready与Unacked的合计视图。

但光看数量还不够，得结合速率才能定位：

可以看到，ACK rate才接近有效完成速率。Deliver rate高不代表处理快，它可能只是把Ready搬到了Unacked。如果用的是自动确认或错误确认策略，还得核对一下，ACK是否真的发生在业务事务完成之后。过早ACK会让队列看起来健康，却把失败消息给丢了。

管理页面的瞬时速率会抖动，所以建议看至少数分钟窗口的趋势。发布端批量发送也可能造成锯齿，不能单凭一个点就下结论。

三、先排除Broker与拓扑层问题

应用耗时不是唯一原因。RabbitMQ节点如果触发了内存或磁盘水位，会对发布连接实施流控；集群节点之间网络异常，或者队列leader所在的节点资源紧张，也会影响吞吐。排查时得检查：

• 节点是否出现了memory alarm、disk alarm或连接blocked；
• 文件描述符、Erlang进程数、磁盘IOPS与网络是否耗尽；
• 队列类型、leader分布、镜像或quorum副本是否符合预期；
• 是否有大消息、过多队列、频繁自动删除或拓扑反复声明；
• 发布确认延迟是否同时上升；
• 是否在近期修改过TTL、死信路由、优先级或最大长度。

注意，不要在事故中直接purge队列。消息可能仍是需要处理的业务事实，删除属于破坏性操作，必须经过业务授权，并有备份或重放方案。如果确认某类消息无效，也应该通过受控脚本按条件迁移或标记，而不是清空整个队列。

状态可以通过管理API、监控系统或运维命令来采集。例如，下面这个命令只是一个示意，生产环境中的凭证与vhost应遵循权限管理：

rabbitmqctl list_queues -p app   name messages_ready messages_unacknowledged consumers   message_bytes_ready message_bytes_unacknowledged
rabbitmqctl list_connections name state channels send_pend

命令输出只是一个快照，还需要与时序指标结合起来看。如果只有一个队列异常，而节点整体健康，那优先查消费者和消息内容；如果所有队列、发布确认和连接都变慢了，那就先处理Broker或基础设施。

四、用容量模型判断“需要多少消费者”

假设平均生产速率为λ条/秒，单个有效消费单元完成速率为μ条/秒，并行消费单元数为c，那么稳定条件近似为：

c × μ > λ

这里的“消费单元”可能是一个进程、一个listener并发线程，也可能受CPU核、容器槽位和数据库连接限制，不能简单等同于实例数。如果单任务平均耗时为S秒且能完全并行，理论μ约为1 / S；实际还要考虑长尾、I/O等待、失败重试和共享资源竞争。

如果已有积压B条，峰值后生产仍为λ，当前总完成速率为C，那么理想清空时间约为：

drain_time = B / (C - λ)    其中C必须大于λ

如果C小于等于λ，那队列永远清不完。计算时要用ACK速率的稳定窗口，而不是用配置的线程数来推测。任务耗时分布有长尾时，平均数会过于乐观，需要结合P95/P99和不同任务类型分别建模。

Little定律可以帮助检查一致性：稳定状态下，系统内平均任务数L约等于到达率λ乘平均停留时间W。如果消息年龄快速上升，但吞吐不变，说明队列等待已成为用户延迟的主导因素。

容量评估还要为节点故障、发布波动和长任务留出余量。不能把系统长期运行在100% CPU、100%连接池占用的理论极限上，否则任何一次重试都可能触发排队雪崩。

五、把消费者处理链路拆成阶段计时

判题任务可能包含：反序列化、拉取代码、创建沙箱、编译、运行测试、上传日志、写数据库、发送结果事件。如果只记录listener总耗时，根本就不知道慢在哪里。需要为每个阶段创建指标或Trace Span：

@RabbitListener(queues = "judge.task")
public void consume(
        JudgeTask task,
        Channel channel,
        @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
    Timer.Sample total = Timer.start(registry);
    try {
        SourceBundle source = timed("source.fetch",
                () -> sourceService.fetch(task.submissionId()));
        Sandbox sandbox = timed("sandbox.create",
                () -> sandboxService.create(task.language()));
        CompileResult compiled = timed("compile",
                () -> compiler.compile(sandbox, source));
        RunResult result = timed("execute",
                () -> runner.run(sandbox, compiled, task.limits()));
        timed("result.persist",
                () -> resultService.finish(task, result));
    } catch (Exception ex) {
        // router必须先把消息可靠送入延迟重试或隔离流程。
        // route失败时异常上抛，连接关闭后原消息会重新投递。
        failureHandler.route(task, ex);
        channel.basicNack(deliveryTag, false, false);
        return;
    } finally {
        total.stop(registry.timer(
                "judge.consume.total",
                "language", safeLanguage(task.language())));
    }
    // 业务结果和幂等记录均已持久化后再确认消息。
    // ACK自身失败时不另发重试消息，由Broker重投原消息。
    channel.basicAck(deliveryTag, false);
}

标签要使用有限枚举，不能把submissionId放进指标，否则会造成高基数。请求ID应放在Trace或日志里。

这里与后文的acknowledge-mode: manual是配套的：成功路径显式basicAck，失败路径只有在错误已经可靠进入延迟重试或隔离流程后，才basicNack(requeue=false)。不能只记录异常后吞掉，也不能对所有失败立即requeue=true，否则毒消息会形成高速循环。如果项目不需要逐条控制确认，应改用AUTO，让容器在监听方法正常返回后确认，而不是保留MANUAL配置却不调用ACK。

阶段耗时与资源指标要交叉观察：如果sandbox.create变慢，同时宿主机磁盘IOPS饱和，那优先处理镜像与磁盘；如果result.persist变慢，同时数据库连接等待增加，那扩容消费者只会更糟；只有应用CPU有余量、下游稳定且Ready增长时，增加并发才更合理。

还要看任务大小分布。少量超长用例占住消费者，会造成队头阻塞。可以按语言、预估用例数或资源等级拆队列，让轻任务不被超长任务拖住；但分队列后要重新规划公平性与总容量，不能无限细分。

六、Prefetch与消费者并发要配套

Prefetch控制Broker允许消费者持有多少条未确认消息。如果设置过大，一个消费者会预取大量任务，其他新实例即使启动也拿不到消息；进程崩溃后，大量Unacked消息重新入队，恢复时的抖动会很明显。如果设置过小，消费者每完成一条都要等待网络投递，可能无法充分利用I/O并发。

Spring AMQP的配置示意如下：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual
        concurrency: 4
        max-concurrency: 8
        prefetch: 8
        default-requeue-rejected: false

这些配置值不能照抄。CPU密集型的任务，prefetch通常接近每个消费线程的在途能力；I/O密集且能并行时，可以稍高一些。需要确认prefetch是按consumer还是channel生效，以及容器实际创建了多少consumer。

一个实用的检查方法是比较Unacked / consumer_count。如果这个值远高于每个实例可并行处理的数量，说明消息被提前占用了。调整后要观察吞吐、公平性、Unacked、内存和重投恢复时间，不能只看Ready是否下降。

使用手动ACK时，只有业务事务和必要的幂等记录成功后才确认。不要把异步任务提交到本地线程池后立即ACK，除非任务已经可靠地落入另一个持久化队列；否则进程崩溃会丢任务。

七、消费者扩容前先检查下游容量

扩容决策至少要满足三个条件：Ready持续上升；现有消费者单元接近自身容量上限；数据库、缓存、第三方API和宿主机依然有余量。否则，应该先修瓶颈或降低入口速率。

举个例子，每个判题Worker同时运行四个容器，宿主机安全上限是二十个容器，那么单机最多大约五个Worker单元；再增加listener并发，只会争抢CPU和内存。数据库连接池也要算进去：每个实例上限20，如果扩到30个实例，理论占用600个连接，数据库未必承受得住。

扩容时要分批增加，观察每批后的ACK rate与下游P95。如果消费者数量翻倍但ACK rate几乎不变，说明瓶颈不在消费者数量；如果错误率和数据库延迟同步上升，应立即停止扩容并回退。

自动扩缩容不要只绑定队列长度。更稳妥的信号是最老消息年龄、净积压增长率、每实例吞吐、CPU/内存和下游保护状态。要设置冷却时间，避免短峰触发实例反复启动；本地模型、沙箱或大镜像启动时，还要计入预热时间。

八、从生产端控制进入速率与任务优先级

当有效消费能力短期无法提升时，必须控制生产。网关可以对非关键请求限流，批量任务可以暂停或延后，重复任务用业务幂等合并。发布端在Broker流控或确认延迟升高时，应主动退避，不能无限地把消息缓存在JVM内存中。

比赛提交可以按用户和题目限速，防止脚本反复提交占满资源。系统任务与在线判题要分开队列，或者保留容量，避免后台重算影响用户请求。高优先级队列只适合明确的少量紧急流量；所有消息都标高优先级，等于没有优先级，还会增加Broker开销。

拆分队列时，要使用业务可解释的维度，比如在线/离线、轻/重任务、不同资源池。不要按每个租户创建大量动态队列，队列与consumer本身也消耗Broker资源。如果公平调度要求复杂，可以在业务调度层先排队，再向有限的RabbitMQ队列发布。

生产端要使用publisher confirm来识别消息是否被Broker接收，并为失败发布保留Outbox或重试记录。没有确认就无限重发，会制造重复消息，消费者仍需幂等。

九、重试、死信与毒消息治理

一个永远失败的消息，如果立即requeue，会形成高速循环，占满消费者和日志。错误必须分类：

退避不能用listener线程sleep数分钟，也不能立即requeue。可以使用带TTL的重试队列、延迟机制或调度服务，重试消息要携带attempt、首次时间和稳定的messageId。达到上限后进入死信队列，由人工或自动修复流程处理。

死信队列不是垃圾桶。要监控它的进入速率、消息年龄和原因，提供脱敏查看、修复、单条重放与审计功能。重放前要确认代码已修复且消费者幂等，否则一次批量回放可能制造第二次事故。

十、积压恢复要防止二次冲击

根因修复后，直接把消费者开到最大并不安全。积压中的老任务可能已被用户取消、超时或由其他流程补偿了；先做业务有效性检查，可以丢弃已处于终态的任务，但这里的“丢弃”要有可审计状态，而不是静默ACK。

恢复计划通常分阶段进行：

1. 保持入口限流，确认新消息不再高速增长；
2. 修复毒消息与下游故障，验证单实例能正确完成任务；
3. 小批量增加消费者，观察ACK rate、错误率和下游P95；
4. 对积压按业务优先级清理，限制总并发；
5. 估算清空时间，定期更新状态；
6. 队列接近正常水位后，逐步解除入口限制；
7. 保留故障后的监控窗口，并完成对账。

恢复过程中，重试流量与正常流量要分开计数。如果大量历史失败消息同时到期，可能产生“重试风暴”。需要为重试设置全局速率和随机抖动，让它只占用一部分容量。

用户可见的任务，要提供状态查询和超时语义。超过有效期的判题，即使最终执行，也可能没有价值；业务层应定义截止时间，消费者在执行昂贵步骤前先检查一下。这样可以减少无效工作，也让清空时间更可预测。

十一、监控与告警应覆盖原因和影响

建议至少采集以下指标：

• Ready、Unacked、总消息、最老消息年龄；
• publish、deliver、ack、redeliver与reject rate；
• consumer数量、连接/Channel状态和重启次数；
• 每个处理阶段的P50/P95/P99与错误分类；
• Prefetch、线程池活跃数、队列长度、拒绝数；
• 数据库连接等待、慢SQL、缓存与外部API延迟；
• 重试队列、死信队列深度和最老年龄；
• Publisher confirm延迟、失败与应用本地待发送数；
• Broker内存、磁盘、文件描述符、网络与alarm。

告警不要只设置固定的队列长度。结合“最老年龄超过SLA”、“净增长持续N分钟”、“消费者为零”、“redeliver比例突升”等条件，会更准确。告警消息应附带vhost、队列、当前速率、消费者数、最近发布版本和Runbook链接，这样值班人员才能快速行动。

日志要用messageId、业务ID、attempt和TraceId串联起来，载荷要脱敏。单条毒消息反复失败时，应使用采样或聚合，避免日志洪水占满磁盘。

十二、测试与性能验证方法

容量测试要使用与生产环境相近的消息大小和任务耗时分布，不能全是固定10毫秒的空任务。需要分别构造稳定流量、阶梯增压、短峰、长任务混入和下游变慢等场景，采集进入率、ACK rate、最老年龄、P95处理时间与资源利用率。

故障注入包括：关闭一个消费者实例、让数据库延迟上升、让容器创建失败、阻断ACK连接、产生毒消息、触发重试队列同时回流。要验证消息没有丢失，重复投递能被幂等处理，死信可定位，恢复阶段不会冲垮下游。

Prefetch实验要固定消费者与任务分布，逐档调整，比较吞吐、Unacked、公平性和崩溃重投时间。扩容实验每次增加少量实例，绘制消费者数与有效ACK rate的关系曲线；曲线进入平台区，说明共享瓶颈出现了。

如果没有实测环境，应报告上述方法和需要采集的指标，不要编造“扩容后提升了几倍”。真实的报告应注明队列类型、RabbitMQ版本、节点规格、消息大小、持久化、publisher confirm、消费者并发、prefetch、下游配置和预热过程。

十三、常见误区与延伸

常见误区包括：只看Ready不看Unacked；把deliver rate当成完成率；一有积压就无限加消费者；认为prefetch越大吞吐越高；业务提交前就ACK；失败立即requeue；死信只进不管；恢复时一次放开全部流量；用队列深度这一个指标驱动自动扩容。

面试回答时，可以按“现象、定位、容量、恢复”这个框架展开。先看Ready/Unacked、生产与ACK rate、消费者数和最老消息年龄；再拆消费者阶段耗时并检查Broker与下游；用λ、μ、并发和清空时间估算容量；确认下游有余量后分批扩容，同时治理prefetch、重试和毒消息；恢复期限流，防止二次冲击。

常见追问有：Ready低但Unacked高怎么办？检查prefetch、线程与下游；消费者翻倍但吞吐不变说明什么？说明存在共享瓶颈或资源上限；如何避免重复消费？使用稳定messageId、Inbox/业务唯一键，事务完成后ACK；怎样估算多久能清空？使用当前有效ACK rate减去持续生产rate，计算净清理速率，并考虑长尾和失败。

十四、总结

RabbitMQ消息积压的本质，是进入速率长期超过有效完成速率。排查要从Ready、Unacked、publish和ACK rate开始，再检查消费者、Broker、处理阶段和下游资源。容量模型能帮助判断系统是否稳定以及理论清空时间，但最终决策必须由真实的ACK rate、任务长尾和资源指标来验证。

扩容只是工具之一。正确的方案还包括合理配置prefetch、业务幂等、生产限流、任务分级、有限退避重试、死信治理和分阶段恢复。能解释每一条消息在哪里等待、为什么变慢、增加并发会压到谁，并通过故障注入证明恢复路径，才算真正把队列从“黑盒缓冲区”变成了可治理的异步系统。

最后，事故复盘时，别忘了把“触发积压的第一项变化”和“放大积压的后续因素”分开来看。一次发布造成单任务耗时上升可能是根因，而立即重试、过大prefetch与盲目扩容则是放大器。把时间线、配置变化和指标拐点对齐，才能制定真正有效的预防措施，而不是只给集群永久加机器。