HDUO Lab

标签： TCP流控

  • 深入 RabbitMQ 陷阱排查：死信路由死循环引发的 Erlang 调度器满载与 TCP 流控雪崩实战

    排查过程中，我们遭遇了一次经典的微服务全局雪崩。上游业务网关的 99 线延迟飙升至 10 秒以上，大量核心服务的 Tomcat/Tomcat 线程池被耗尽，所有涉及异步消息投递的 API 全部处于超时阻塞状态。最终排查结论：业务侧在利用 RabbitMQ 实现延迟重试机制时，错误地将死信交换机（DLX）与业务队列配置成了路由闭环，且代码中缺失对最大重试次数的校验。几条格式非法的“毒药消息（Poison Pill）”在队列间以每秒数万次的频率疯狂死循环，打满了 RabbitMQ 的 Erlang VM 调度器。底层的 Credit-based 流控机制被触发，导致全局生产者的 TCP 连接被强行 blocked，最终由于 TCP 反压（Backpressure）引发了整个微服务集群的线程池雪崩。

    不要把 MQ 当成无所不能的状态机，更不要在没有熔断机制的情况下滥用死信路由。今天我们把这个看似低级但极具破坏力的 AMQP 路由陷阱拆解清楚。

    案发现场：诡异的 TCP 零窗口与 Blocked 连接

    报警爆发时，监控大屏一片通红。RabbitMQ 集群节点 Load Average 飙升至 80+，CPU 占用率顶满（多核 100%）。

    首先怀疑是消息堆积导致触发了 vm_memory_high_watermark（内存阈值告警），但通过 rabbitmqctl 查看，队列深度（Messages Ready）居然几乎为 0。

    直接上服务器抓取连接状态，发现了致命线索：

    # 查看所有连接状态，过滤出被阻塞的连接
$ rabbitmqctl list_connections pid name state channels | grep -E 'block|flow'
<0.14532.1>  10.0.x.x:54321 -> 10.0.x.y:5672  blocked     4
<0.14555.1>  10.0.x.z:43210 -> 10.0.x.y:5672  blocking    2
... (输出了上千行)

    注意这里的状态：blocking 表示 RabbitMQ 正在试图停止读取该连接的数据，而 blocked 表示该连接已经被彻底禁止写入。

    切到微服务宿主机，用 ss 命令检查 TCP 状态：

    $ ss -tnp | grep 5672
ESTAB  0      409600  10.0.x.x:54321   10.0.x.y:5672   users:(("java",pid=1024,fd=154))

    注意 Send-Q（发送队列）已经堆积了 409600 字节。此时客户端抓包会看到大量的 TCP Zero Window。这说明 RabbitMQ 已经停止从 Socket 接收缓冲区读取数据，内核通过 TCP 窗口向外传递了背压，最终导致 Java 客户端底层的 Socket.write() 发生阻塞，业务线程就此卡死。

    深挖死因：AMQP 路由死循环与流控机制

    既然没有堆积，为什么 Erlang VM 会拒绝服务？

    查看消息的吞吐速率指标，我们看到了令人毛骨悚然的数据：

    $ rabbitmqctl list_queues name messages_ready message_stats.publish_details.rate message_stats.deliver_details.rate
name            messages_ready  publish_rate    deliver_rate
main_queue      0               45321.5         45321.5
retry_queue     0               45321.5         45321.5

    队列里没有现成消息，但每秒的入队/出队速率高达 4.5 万次。

    追查业务拓扑，发现开发人员为了实现“延迟重试”，搞了这样一个骚操作：

    1. 消息进入 main_queue。

    2. 消费者遇到 JSON 解析异常，直接抛错并执行了 basic.reject(requeue=false)。

    3. main_queue 绑定了死信交换机 DLX_Retry，消息被路由至 retry_queue。

    4. retry_queue 没有消费者，但设置了 x-message-ttl: 1000（1秒过期），同时它绑定的死信交换机居然是 main_queue 前端的 DLX_Main！

    这就形成了一个完美的无阻尼振荡回路： main_queue -> 消费失败 -> DLX -> retry_queue -> 1秒过期 -> DLX -> main_queue…

    如果只是一条消息，每秒循环一次也就罢了。但业务侧在几分钟内连续抛入了数百条非法数据，且 RabbitMQ 在消息多次死信转发时，实际上忽略了部分 TTL 开销，导致在消费者不停 Reject 的情况下，这几百条消息在 Erlang Schedulers 中疯狂流转。

    底层的 Credit-based Flow Control 是如何掐死生产者的？ RabbitMQ 内部的进程通信基于信用点（Credit）机制。当消息高速流转时，队列进程（Queue Process）的 CPU 开销极大，处理速度跟不上，它就会停止向通道进程（Channel Process）发放 Credit；通道进程没有了 Credit，就会停止向连接进程（Connection Process）发放；连接进程拿不到 Credit，直接停止调用 gen_tcp:recv() 从 Socket 读取数据。 这就是为什么没有触发内存高水位告警，却依然引发了全连接 blocked。

    破局与修复：阻断循环与防御性编程

    第一步：紧急止血 不要试图去 Purge 队列，因为消息在高速流动，Purge 命令大概率扑空。最直接的办法是打破拓扑闭环，解除 retry_queue 的死信绑定，或者直接停掉 main_queue 的消费者，让消息在 main_queue 沉淀下来后再一并 Purge。

    # 紧急解绑 retry_queue 的死信路由
$ rabbitmqadmin delete binding source="DLX_Main" destination_type="queue" destination="main_queue" routing_key="retry_routing_key"

    第二步：代码层面的防御性修复 开发人员最大的错觉是：认为异常扔给死信队列就万事大吉。 任何基于 DLX 的重试机制，必须校验 x-death Header 来限制最大重试次数。AMQP 协议在消息每次经过死信路由时，都会在 Header 中追加/更新 x-death 属性。

    在 Spring AMQP 中，正确的防御性消费代码应该长这样：

    @RabbitListener(queues = "main_queue")
public void processMessage(Message message, Channel channel) throws IOException {
    try {
        // 执行业务逻辑
        doBusinessLogic(message.getBody());
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    } catch (Exception e) {
        // 核心防御：检查重试次数
        List<Map<String, ?>> xDeath = message.getMessageProperties().getXDeathHeader();
        long retryCount = getRetryCount(xDeath);

        if (retryCount >= 3) {
            log.error("Poison Pill detected, discard message. ID: {}", message.getMessageProperties().getMessageId());
            // 超过阈值，直接 ACK 丢弃，或者路由到人工干预的 failed_queue，严禁再次 Reject
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } else {
            // 未达阈值，丢入死信重试回路
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        }
    }
}

private long getRetryCount(List<Map<String, ?>> xDeath) {
    if (xDeath != null && !xDeath.isEmpty()) {
        Long count = (Long) xDeath.get(0).get("count");
        return count != null ? count : 0;
    }
    return 0;
}

    架构师的忠告

    不要用 MQ 去做复杂的定时任务或状态机。如果需要高级的延迟机制，请老老实实使用 rabbitmq-delayed-message-exchange 插件，它基于 Erlang 计时器实现，而不是依赖 TTL+DLX 这种极易产生死循环的 Hack 手段。 系统设计必须要有“边界感”：一个无法被处理的毒药消息，如果没有重试次数上限的约束，它就是一颗悬在集群上的定时炸弹。

    同类问题速查清单 (Troubleshooting Checklist)

    1. 连接 Blocked 检查： 通过 rabbitmqctl list_connections name state | grep -E 'block|flow' 快速确认是否存在 TCP 级的流控背压。如果存在，大概率是 Erlang VM 资源吃紧（内存触顶或队列进程 CPU 满载）。

    2. 流量幻影现象： 如果队列 messages_ready 为 0，但节点 CPU 满载，立刻检查 publish_rate 和 deliver_rate。高吞吐+零堆积通常意味着两种情况：存在死信路由闭环，或者消费者存在无条件 requeue=true 的异常死循环。

    3. 毒药消息排查： 从发生异常的业务模块着手，查看 Error 日志中是否有固定几条 Message ID 频繁出现。若有，立刻切断该队列的死信回路或下线对应 Consumer 集群。

    4. 流控阈值审查： 检查 rabbitmq.conf 中的 vm_memory_high_watermark.relative（默认 0.4）配置。确认当前 blocked 是由于内存阈值触发，还是调度器 Credit 耗尽触发。前者需扩容或优化堆积，后者必须从业务逻辑层面阻断死循环。