RocketMQ 消费者模型与负载均衡

一句话：MessageQueue 是 RocketMQ 的并行度单位和顺序单位；消费者再多，也只能按 Queue 分配，Rebalance 决定哪个实例拿到哪几条 Queue。

适用版本：4.9 / 5.3 / 5.5。Push 消费、Rebalance 周期、分配策略在三个版本间基本一致。

为什么需要

消费能力要能水平扩展：流量涨了加实例，流量降了减实例。但「加实例」不等于「自动变快」，因为消息不是以单条为单位分发的，而是以 Queue 为单位分配的。

这带来两个直接后果：Queue 数决定并行度上限；Queue 的归属变化会让某些队列短暂停止消费。理解 Rebalance，本质是理解这两个后果，而不是记住某个算法名字。

核心机制

Push 是长轮询，不是 Broker 主动推

DefaultMQPushConsumer 对开发者表现为「有消息就回调」，底层仍是 Consumer 发 Pull 请求：

长轮询的价值是兼顾实时性和空转成本：短轮询会高频打 Broker，纯 Push 又需要 Broker 维护连接状态。Broker 侧 longPollingEnable=true，短轮询兜底为 1s。

Pull 与消费的批量参数：

参数	默认	含义
pullBatchSize	32	一次网络拉取最多多少条
consumeMessageBatchMaxSize	1	一次回调最多多少条
pullThresholdForQueue	1000	单 Queue 本地缓存消息数上限，超过则暂停拉取
pullThresholdSizeForQueue	100 MiB	单 Queue 本地缓存大小上限

Rebalance：20 秒一次的全组重算

触发条件：

• 组内 Consumer 实例增减（上线、下线、心跳超时）。
• Topic 的 Queue 数变化（扩队列、Broker 上下线）。
• 定时触发：默认每 20s 一次（rebalanceInterval）。

每次 Rebalance，组内每个 Consumer 都从 Broker 拿到「当前存活的 Consumer ID 列表 + Topic 的 Queue 列表」，用同一个分配策略各自算出自己该拿哪些 Queue。注意：不是 Broker 集中分配，而是每个客户端独立计算；只要输入一致，结果就一致。

Rebalance 期间，被调整的 Queue 会先停止拉取、提交位点，再按新归属恢复。这个窗口通常是秒级，但对低延迟业务是可感知的。

四种分配策略的真实影响面

以 8 个 Queue、3 个 Consumer 为例，初始分配（Averagely）：

A: Q0 Q1 Q2
B: Q3 Q4 Q5
C: Q6 Q7

Averagely（默认）：按 Consumer ID 排序后连续切分。算法是「前 8 % 3 = 2 个 Consumer 各多拿 1 个」。

当 D 加入，4 个 Consumer 分 8 个 Queue，变成每人 2 个：

A: Q0 Q1        （失去 Q2）
B: Q2 Q3        （失去 Q4 Q5，拿到 Q2）
C: Q4 Q5        （失去 Q6 Q7，拿到 Q4 Q5）
D: Q6 Q7

A、B、C 三个消费者都受影响，不是「只把新 Consumer 插进去、其他人不动」。这是 Averagely 最重要的事实：它每次都按新的 Consumer 列表全局重算，成员一变，多个 Consumer 的 Queue 归属都可能变。反过来，Consumer 下线也是一次全局重算，不只是把它原来的 Queue 平均分给剩下的人。

AveragelyByCircle：轮流发牌，Q0→A、Q1→B、Q2→C、Q3→A……8 个 Queue 在 3 个 Consumer 下结果与 Averagely 相同；成员变化时同样全组重算，差异只在某些数量组合下更均匀。生产上很少单独选它。

ConsistentHash：把 Consumer 和 Queue 都哈希到环上，Queue 顺时针找最近的 Consumer。D 加入后只影响环上相邻的一段，通常只有 1 个 Consumer 失去少量 Queue，其余不动。代价是分布不一定均匀，且需要虚拟节点（默认 10 个）来缓解倾斜。适合 Pod 频繁扩缩的场景。

ByConfig：手工指定 Queue → Consumer 的映射，不做自动分配。Consumer 增减不会改变现有映射；某个 Consumer 挂了，它负责的 Queue 就成了孤儿，直到人工改配置。适合灰度或把大 Queue 固定给高性能机器。

策略	成员变化时谁受影响	分布均匀性	适用
Averagely	全组重算，多个 Consumer 可能换 Queue	高	绝大多数场景（默认）
AveragelyByCircle	全组重算	高	历史兼容，少用
ConsistentHash	环上相邻的 1 个左右	中，需虚拟节点	Consumer 频繁上下线
ByConfig	无自动调整	由配置决定	灰度、固定映射

消费模式与并发

• 集群消费（CLUSTERING，默认）：同组内一个 Queue 同时只给一个 Consumer，位点存 Broker。
• 广播消费（BROADCASTING）：每个 Consumer 都消费全量，位点存本地 ~/.rocketmq_offsets，不做 Rebalance 分摊。
• 并发消费：同一 Queue 的消息可被线程池并发处理，不保证顺序。线程数由 consumeThreadMin / consumeThreadMax 控制（4.x 默认 20/64；5.x 以版本为准）。
• 顺序消费：同一 Queue 加锁串行，失败原地重试，会阻塞后续消息。

实战场景

一、扩容 4 个消费者，一半订单被重复处理

现象：积压告警触发自动扩容，消费组从 3 个实例扩到 4 个。扩容后 3 分钟内日志出现大量重复处理（同一 orderId 被处理两次），部分实例报 ConsumeMessageService 线程池繁忙；积压反而短时上升后才下降。

排查：Averagely 每次按新的 Consumer 列表全局重算，不是把新成员插进去就完事。

定位层	取证据	判定
归属变化	消费者日志 rebalance / MessageQueue changed，对比前后各实例的 Queue 列表	3 个老实例的归属都变了即命中
重算时点	默认定时 20s 一次（rebalanceInterval），成员变化后立即触发	毛刺时间点与扩容时间对齐
位点是否提交	被移走的 Queue 在停止拉取前是否已提交位点	处理完未提交 → 新归属者重放
重复量	同一 keys/msgId 的处理次数分布	集中在被移动的 Queue 上即确认
消费线程	consumeThreadMin/Max 与 pullBatchSize=32、consumeMessageBatchMaxSize=1	重算窗口内并发抖动放大延迟

8 个 Queue、3 个实例：A=Q0-Q2 B=Q3-Q5 C=Q6-Q7
加到 4 个实例后：A=Q0,Q1  B=Q2,Q3  C=Q4,Q5  D=Q6,Q7
A、B、C 三人归属全变，不是“D 拿一点、别人不动”

对症做法：

1. 接受“重算必然发生”，把处理逻辑改成幂等：
   以业务唯一键（orderId + eventType）做唯一索引或 UPSERT，
   重复投递只生效一次，这才是消费端的第一道防线。
2. 降低归属抖动：用一致性哈希分配策略，成员增减只影响相邻区间。

consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueConsistentHash());
// 或者用机房就近策略（跨机房部署时避免远程拉取）
consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueByConfig());   // 固定映射，适合极稳定拓扑

3. 缩容要走优雅下线：先停止拉取并提交位点，再退出进程，
   避免“进程已退出、位点未提交”造成新归属者重放整段。
4. 扩容优先于“自动扩容”：在积压持续 N 分钟后手动加实例并观察，
   自动扩缩在低延迟业务上会制造持续的重算。

复盘要点：判据是“重复处理不影响业务结果”——用幂等率（重复消息被识别并跳过的比例）验证，而不是要求“没有重复”。同时要明确 Rebalance 期间的秒级暂停对低延迟业务可感知，这类业务应考虑固定映射或减少实例数变化，见 消费者模型 里的 20s 周期这一事实。

二、消费“不动”但不报错：本地流控阈值触发暂停拉取

现象：报表消费组处理变慢后，Broker 侧积压持续上涨，而消费者 CPU 只有 20%、日志没有任何异常；控制台显示这些实例“在线”。重启消费者后短暂恢复，很快又停住。

排查：客户端有本地缓存上限，超过就暂停拉取，这是保护而不是故障。

定位层	取证据	判定
本地缓存是否打满	ProcessQueue 消息数与 pullThresholdForQueue（默认 1000）	打满即暂停拉取，Broker 侧积压自然涨
缓存大小上限	pullThresholdSizeForQueue（默认 100 MiB）	大报文时先撞这条线
是否真的暂停	消费者日志 the cached message count exceeds the threshold	命中即确认流控
拉取与消费速率	pullBatchSize=32 与实际消费 TPS	消费慢时拉得再快也无意义
GC 与堆	老年代占用、GC 频率	本地缓存大报文会压垮堆
订阅关系	是否同组订阅不同 Topic/Tag	排除下一节的经典故障

Push 的表象（“有消息就回调”）掩盖了底层是 Consumer 主动 Pull：拉取被暂停时，Broker 那个 15s 的长轮询 hold 也不会把消息送过来。

对症做法：

// 1. 先解决“消费慢”的根因，再谈阈值
consumer.setConsumeThreadMin(20);
consumer.setConsumeThreadMax(20);
consumer.setPullBatchSize(16);            // 单批更小，降低本地驻留与尾延迟
consumer.setConsumeMessageBatchMaxSize(1);

// 2. 大报文场景按大小而不是条数控制，并把阈值显式写出来
consumer.setPullThresholdForQueue(200);          // 报文大时调小条数上限
consumer.setPullThresholdSizeForQueue(32);        // 32 MiB，配合堆大小校准

// 3. 处理逻辑本身批量化，减少单位时间的回调次数
consumer.setConsumeMessageBatchMaxSize(32);       // 一次回调 32 条，内部做批量写库

4. 提高并行度的前提是 Queue 数够：单 Queue 同一时刻只被组内一个消费者拉取，
   热点 Queue 加线程无效，只能扩 Queue 或在业务侧拆分发。
5. 阈值不是越大约好：本地缓存变大只会把压力从 Broker 搬到客户端堆里，
   OOM 后的重放规模更大。

复盘要点：判据是“本地缓存水位与 Broker 积压要同时出现在监控里”——只看积压会误判成“没在消费”。消费速率、本地缓存条数/字节数、暂停拉取次数三条曲线放在一起，5 分钟内就能定性是消费慢还是拉取被限。

三、同组订阅关系不一致，消息静默丢失

现象：两个服务共用 order-consumer 这个消费组，一个订阅 OrderTopic 的 Tag create，另一个订阅同一 Topic 的 Tag cancel。表现是：一部分 cancel 消息永远没人消费，另一部分被错误地投给了 create 的处理逻辑并抛异常；把消费者拆成两台机器后现象“随机变化”。

排查：RocketMQ 的队列分配按消费组计算，订阅关系必须在组内一致。

定位层	取证据	判定
组内订阅是否一致	控制台“消费组-订阅关系”对比各实例的 Topic + Tag 表达式	不一致即命中，且会报订阅关系不一致
Tag 表达式差异	是否一个用 create、另一个用 `create
实例归属	同一 Queue 在不同时刻被不同订阅表达的实例持有	解释“随机丢失/错投”
消费组命名	是否为了省事复用同一个 group 名	跨业务复用 group 是根因
位点互相覆盖	两个实例提交同一 <group, topic, queue> 位点	位点被来回推进，出现跳过

订阅关系不一致的后果不是报错，而是消息被过滤掉或位点被互相覆盖，这让它比大多数配置错误更难发现。

对症做法：

// 1. 一个消费组只做一件事：同一 group 内所有实例订阅完全相同的 Topic + Tag
consumer.setConsumerGroup("order-create-consumer");
consumer.subscribe("OrderTopic", "create");

// 2. 需要处理多种 Tag 时用表达式在一个消费者里分发，而不是拆实例共用 group
consumer.subscribe("OrderTopic", "create || cancel");
// 回调内按 Tag 分流
String tag = msg.getTags();
if ("create".equals(tag)) onCreate(msg); else if ("cancel".equals(tag)) onCancel(msg);

3. 命名规则强制：consumerGroup 必须带业务域 + 用途，禁止跨服务复用；
   新增消费者时先申请 group 名，由平台侧唯一性校验。
4. 上线检查：订阅关系一致性校验失败即拒绝启动，
   而不是等控制台飘红。

// 4. 启动自检：拉取本组订阅关系并与期望比对
@PostConstruct void assertSubscriptionConsistent() {
    Set<String> actual = admin.querySubscriptionSet(consumerGroup);      // 组内所有实例的订阅表达式
    if (actual.size() > 1) throw new IllegalStateException("inconsistent subscription: " + actual);
}

复盘要点：判据是“每个消费组内订阅关系集合大小为 1”，这条可以中心化扫描所有 group 自动检查，应该纳入巡检。跨业务复用 group 属于命名治理问题，技术校验只能兜住一部分；顺序消费场景下这类混用还会破坏“同 Queue 单线程”的前提，见 消息类型与使用场景。

最小上手

# 1. 建一个 8 队列的 Topic
sh bin/mqadmin updateTopic -n 127.0.0.1:9876 -c DefaultCluster -t demo -r 8 -w 8

# 2. 起第 1 个 Consumer（同一个 group），观察它拿到 8 条 Queue
# 3. 再起第 2、第 3 个，观察分配变化

# 4. 看每个队列的位点与积压
sh bin/mqadmin consumerProgress -n 127.0.0.1:9876 -g cg-demo

# 部分版本加 -s 会显示队列当前归属的客户端 IP
sh bin/mqadmin consumerProgress -n 127.0.0.1:9876 -g cg-demo -s

Consumer 侧打开 Rebalance 日志（logback 里把 org.apache.rocketmq.client.impl.consumer.RebalanceImpl 调到 INFO/DEBUG），能看到类似：

rebalance result: mqSet=[demo-0, demo-1], cid=consumer-1

预期结果：

• 1 个 Consumer：8 条 Queue 全归它。
• 2 个 Consumer：各 4 条。
• 3 个 Consumer：A 拿 3 条，B 拿 3 条，C 拿 2 条。
• 第 4 个加入：重新变成每人 2 条，前三个都会换掉部分 Queue。

生产动作

• Queue 数按「目标 Consumer 数 × 2」规划，给扩容留余量。Queue 只能增不能减，事后调整成本高。
• Consumer 数不要超过 Queue 数，多出的实例拿不到 Queue，只会空转。
• 避免频繁扩缩容。每次成员变化都触发全组 Rebalance，Averagely 下影响面是整组；K8s 场景把 rebalanceInterval 与 Pod 稳定时间一起考虑，或改用 ConsistentHash。
• 监控 Rebalance 频率与消费暂停时间。频繁 Rebalance 通常意味着 Consumer 心跳超时、GC 停顿或网络抖动，而不是分配算法有问题。
• 顺序消费单独建组、单独评估 Queue 数。它不能靠加 Consumer 提升单 Queue 吞吐，失败还会阻塞。
• 调 pullThresholdForQueue 要成对看内存：调大能扛住短时消费抖动，但单 Queue 缓存过多会推高堆内存和重复消费范围。

边界

• 没有真正的 Push。Broker 不会主动建立连接推消息，所有「推送」都是客户端长轮询实现的。
• Queue 不能减少。要减少并行度只能建新 Topic 迁移，或让多余 Consumer 空转。
• Rebalance 不是即时的。Consumer 变化后最长约 20s 才被全组感知，期间可能出现短暂重复或暂停。
• 广播模式不做队列分摊，每个实例消费全量，也没有 Broker 位点，换机器就会丢进度。
• Averagely 不是「只重分新增/下线者的 Queue」。它每次按最新成员列表全量重算，多个 Consumer 都可能换 Queue；把它描述成「影响面最小」只在「不需要改算法」的意义上成立。
• 分配策略不解决消费慢。消费慢要看下游 RT、线程池、批量大小和 Queue 数，不是换个策略就能好。

讲清楚

RocketMQ 消费端的并行单位是 Queue，不是消息。一个 Topic 有多少 Queue，最多就有多少个 Consumer 真正干活；多出来的实例拿不到 Queue，只会空转。Push 消费底层是长轮询：Consumer 发拉取请求，Broker 没消息时最多 hold 15 秒，有消息立即返回。

Rebalance 默认 20 秒一次，Consumer 增减或 Queue 变化都会触发。触发后不是 Broker 集中分配，而是组内每个 Consumer 拿到同一份成员和 Queue 列表，各自用同一套算法算出自己该拿哪些 Queue。默认的 Averagely 每次都全局重算：8 个 Queue、3 个 Consumer 是 3/3/2，第 4 个加入后变成每人 2 个，前三个都会换掉部分 Queue，不是只动新来的那一个。ConsistentHash 只影响环上相邻的一段，适合 Pod 频繁扩缩；ByConfig 不自动调整，挂了就成孤儿。

工程上的结论很直接：Queue 数按目标 Consumer 数的两倍规划，Consumer 不要超过 Queue 数，避免频繁扩缩容。顺序消费还要接受失败时整条 Queue 停住。