RabbitMQ 如何实现消费端限流？

结论：RabbitMQ 的消费端限流只有一个原生开关——basicQos 的预取计数（prefetch count），它限制"每个消费者最多有多少条未确认消息"，配合手动 ack 就能把 Broker 的推送速度压到业务处理能力之内。业务层的限流（线程池、令牌桶、熔断）是第二道闸，不能替代 QoS。

一、预取窗口怎么工作

Broker                          Consumer
  │  prefetch=3，最多推 3 条未 ack  │
  ├── msg1 ──────────────────────>│ 处理中
  ├── msg2 ──────────────────────>│ 处理中
  ├── msg3 ──────────────────────>│ 处理中
  │   （窗口已满，暂停推送）          │
  │<──────── basicAck(msg1) ───────┤ 处理完成
  ├── msg4 ──────────────────────>│ 窗口腾出一个位置
  │                                │
  └── 若消费者崩溃，未 ack 的 3 条重新入队

窗口大小直接决定了三件事：吞吐上限、内存占用、故障时的重投量。prefetch 越大吞吐越高，但消费者崩溃时一次性重投的消息也越多，且容易把内存打满。

二、QoS 参数

经验公式：prefetch ≈ 目标吞吐 × 单条处理耗时。比如单条 50 毫秒、想做到 200 TPS，窗口取 10 左右即可；取 1000 只会让消息堆在客户端内存里。

三、代码实现

Connection conn = factory.newConnection();
Channel ch = conn.createChannel();
ch.queueDeclare("order.pay.q", true, false, false, Map.of("x-queue-type", "quorum"));

// 关键：先设 QoS，再开始消费
ch.basicQos(20);                       // 每个消费者最多 20 条未确认

ch.basicConsume("order.pay.q", false, (tag, msg) -> {
    try {
        handler.handle(msg);           // 同步处理，慢一点也没关系
        ch.basicAck(tag, false);
    } catch (Exception e) {
        ch.basicNack(tag, false, false); // 进死信，避免无限重投
    }
});

业务层再加一道闸：把处理逻辑丢进固定大小的线程池，池满时阻塞或拒绝，避免"QoS 放进来 20 条，线程池瞬间排满 200 个任务"。

ThreadPoolExecutor pool = new ThreadPoolExecutor(
        8, 8, 0L, TimeUnit.MILLISECONDS,
        new ArrayBlockingQueue<>(200),
        new ThreadPoolExecutor.CallerRunsPolicy());   // 满了让消费线程自己处理，形成反压

四、限流与相关机制的区别

注意 Broker 的流控（blocked 连接）是"整体刹车"，会连累同节点所有生产者；消费端限流是"精准节流"，两者要分开配置。

五、常见追问

• 为什么开了 autoAck 限流就无效？ autoAck 在消息发出时立刻确认，未确认计数永远是 0，QoS 窗口永远有空位，Broker 会全速推送直到消费者内存爆掉。
• prefetch 设 1 是不是最稳？ 最稳但吞吐最低，每条消息都要等一次网络往返；网络 RTT 高时（跨机房）损耗非常明显。
• prefetch 能动态调整吗？ 可以，重新调用 basicQos 只影响后续投递，已经发出的未确认消息不受影响。
• 多个消费者怎么分配窗口？ 每个消费者独立计数（global=false），所以 3 个消费者、prefetch=20，稳态下最多有 60 条未确认消息。

小结

• 消费端限流的原生手段就是 basicQos(prefetch) + 手动 ack，别指望 autoAck。
• prefetch 按"吞吐 × 处理耗时"估算，过小浪费往返、过大放大重投与内存。
• 业务线程池是第二道闸，池满时的拒绝策略决定了反压能否传回消费线程。
• Broker 级流控（内存/磁盘告警）与被消费端限流是两回事，需要分别监控。