读完这篇，你能做什么
能为订单、积分和通知画出订阅关系，并解释同组消费者的作用。
学习理由、准备要求与知识点
为什么学它
理解主题、队列与组之后，才能正确表达一份业务事件由哪些独立服务处理。
先读这些内容
知识点清单
- Topic
- Message Queue
- Producer
- Consumer Group
- 组内分担
- 组间独立
- Tag 过滤
- 订阅一致性
运行前先确认主线为 RocketMQ 5.x 领域模型。源码对照 rocketmq-all-5.3.1 仓库中的经典 remoting Java 客户端 AVG 队列分配策略；新 5.x gRPC SDK 的消息粒度负载均衡不能套用同一队列独占结论。算法使用 Java 17。
01先弄懂它是什么
读 RocketMQ 配置时先问三个问题：事件是哪类业务、存到哪里、由哪种职责处理。Topic 负责逻辑分类，MessageQueue 负责存储与扩展，ConsumerGroup 表示共同承担同一消费职责的组；生产者和消费者才是实际运行的进程。把 Topic、组和消息键混成一种名字，会导致订阅和扩容都不符合预期。
例如订单事件进入 OrderEvents，Audit 审计组和 Points 积分组各自处理一份。给 Points 增加实例，是让更多实例分担积分工作，不是再增加一份独立订阅。不同业务要独立组，并各自记录成功、重试和进度。同组实例的 Topic 与过滤表达式应一致，否则部分实例要处理的消息可能被其他实例过滤掉。
消息本身也分多种身份：Topic 选择业务类别，Tag 常作细分类，Keys 帮助按业务线索检索，消息 ID 关联 Broker 接收或 SDK 行为，而 eventId 由业务定义同一事实。
不要把 Keys 当成数据库唯一索引，也不要把每次发送尝试的新消息 ID 当成跨重试稳定的幂等键。可以在负载中携带 eventId、orderId、schemaVersion，但服务端过滤使用的元数据必须通过 SDK 正确设置，仅写进 JSON 字段不会自动产生过滤能力。
文中这几个词，先认一认
- Topic
- 按业务对消息分类和管理的名称，下面可以包含一个或多个队列。
- ConsumerGroup
- 共同承担一种消费职责的逻辑组，不是某一个运行进程的名字。
- 订阅关系
- 一组消费者约定读取哪个 Topic、按什么表达式过滤，同组实例应保持一致。
- 消息粒度分担
- 把同一队列中的不同消息分给不同消费者处理，和经典客户端按整个队列分配的方式不同。
02看看它怎么工作
发布路径先取得 Topic 的可用路由，再按客户端与消息类型选择承载位置，发送消息到服务端。消费路径根据组订阅获取可处理消息，执行业务并反馈结果。这里需要区分数据的存储位置和执行位置：队列是存储分片，线程或消费者是执行资源，一条消息被哪个线程处理并不构成它的永久身份。
在经典客户端的队列粒度负载均衡中，组成员依据一致的队列清单与消费者清单计算分配。若队列数为 8、消费者数为 3，平均分配可得到 3、3、2。超过队列数的实例可能拿不到队列，所以“加实例就必然线性扩容”不成立。
但 RocketMQ 5.x 新 SDK 的 PushConsumer、SimpleConsumer 常采用消息粒度负载均衡，其并发关系不同。先确认实际客户端和消费类型，再决定容量上限，不能只根据服务端版本做推断。
组成员加入、离开或路由变化会触发工作重新分配。此时旧持有者可能仍有在途业务，新持有者又开始接手，业务去重与顺序要求不能依赖理想的瞬时切换。顺序消息还要求按消息组维持特定序列，不能把普通消息的最大并行方式直接用于所有事件。模型图用两个组显示职责隔离；具体实例分配规则要放回所选 SDK 的协议中解释。
正在加载流程图…
查看流程图源码（Mermaid）
flowchart TD
P["订单生产者"] --> T["Topic：OrderEvents"]
T --> Q["存储队列集合"]
Q --> A["审计组 Audit 的订阅"]
Q --> B["积分组 Points 的订阅"]
A --> A1["审计实例协作"]
B --> B1["积分实例一"]
B --> B2["积分实例二"]
B1 --> D["积分业务结果与幂等记录"]
B2 --> D按顺序看，每一步做了什么
- 确定事件业务边界及消息类型，再规划 Topic。
- 为独立接收职责创建独立
ConsumerGroup。 - 把 Topic、Tag 或过滤表达式作为统一组配置发布。
- 确认所用 SDK 的消费类型与负载均衡粒度。
- 通过稳定
eventId核对每个职责的最终处理结果，并观察重分配期间重复。
03对照实现代码
rocketmq-all-5.3.1，经典 remoting 客户端代码 · AllocateMessageQueueAveragely#allocate
原创数学化局部模型，复述均分与余数分配思想，不是原样复制源码。只演示已排序清单下某个消费者的范围计算，不代表新 gRPC SDK 的消息粒度分配实现。
final class AverageRangeModel {
record Range(int from, int length) {}
static Range locate(int queueCount, int consumerCount, int index) {
if (queueCount < 0 || consumerCount <= 0
|| index < 0 || index >= consumerCount)
throw new IllegalArgumentException("invalid counts");
int base = queueCount / consumerCount; // 每个实例至少获得的份数。
int extras = queueCount % consumerCount; // 前 extras 个实例各多一份。
int length = base + (index < extras ? 1 : 0);
int from = index * base + Math.min(index, extras); // 前面实例共占的数量。
return new Range(from, length); // 半开区间 [from, from+length)。
}
}allocate 先检查组、当前消费者和全部清单，再按当前消费者在清单里的位置算出负责范围。清单顺序也是输入：两个实例看到的消费者顺序不同，即使运行同一段方法，也可能都认为某个队列归自己。相同输入得到相同输出，并不保证不同机器拿到的输入一致；完整的重平衡流程还要让成员使用一致清单，并完成队列交接。
源码把不能整除的余数优先分给靠前的消费者，保证分配数量最多相差一。教学实现用 base 与 extras 将条件归一化：q=cm+r，前 r 个区间长度 m+1，其余长度 m。起点是之前所有区间长度之和。这一推导比记忆多层三元表达式更容易检查队列数小于实例数时的行为。
清单分配能够保证队列数量均衡，却不能保证工作量均衡。一个队列若承载热商户事件，单条处理又特别慢，它的一个分片可能比另外十个更重。把数量均衡直接写成吞吐均衡，是从数据结构模型越界到了未测量的业务假设。后面的高级算法会给任务附权重，进一步看不同目标的代价。
返回一段队列列表只是计算阶段。真实客户端还要对比旧分配、停止不再属于自己的任务、更新拉取进度并启动新工作。这个交接阶段会受到网络延迟、在途消息和失败重试影响。源码阅读应沿着“输入视图、分配计算、交接执行”三层进行，不能用一个返回列表的方法解释全部故障恢复。
代码先调用谁，再调用谁
- 客户端取得 Topic 队列与组成员视图
- 重平衡逻辑按策略调用 allocate
- AVG 策略计算实例索引、基数与余数
- 返回该实例对应的队列范围
- 完整重平衡流程对比旧新分配并管理消费任务交接
04动手写一个例子
均匀连续分片：证明无重叠、无遗漏与数量平衡
给定 n 个有序任务和 c 个工作者，将任务按连续区间尽可能均匀分配。前面的工作者优先多分一个，返回全部半开区间。n 可以为零，c 必须为正。
- 输入 1
n=8，c=3- 输出 1
[0,3), [3,6), [6,8)
- 输入 2
n=2，c=4- 输出 2
[0,1), [1,2), [2,2), [2,2)
欧几里得除法给出 n=c×base+extras，其中 0≤extras<c。若每个工作者先拿 base 项，仍有 extras 项未分。把它们逐个发给前 extras 个工作者，所有长度便只可能是 base 或 base+1，因此最大长度差不超过一。这直接给出了数量平衡的构造。
第 i 个区间之前有 i×base 个基础任务，另有 min(i,extras) 个额外任务，所以起点为两者之和。长度为 base 加上 i<extras 的指示值。相邻起点正好等于前一个终点，因此不会重叠，也不会留下空洞。最后一个终点由总数公式可知等于 n，从而覆盖全部任务。
代码不真的创建 n 个任务对象，只返回 c 个区间，因此即使 n 很大仍能检验数学边界。测试除常见 8 分 3，还验证零任务、工作者多于任务和 Integer.MAX_VALUE 规模。校验函数计算所有长度之和并检查端点相接，验证的是问题性质，而不是把同一公式复制一遍当成测试预期。
为什么这样写是对的前 i 个区间恰好覆盖 [0,start_i)，互不重叠；每个已分配长度属于 base 或 base+1，最后累计长度等于 n。
每个工作者只做常数次整数运算并产生一个区间。n 不决定循环次数，因为程序没有展开具体任务清单；若需要逐个复制任务到列表，时间和输出空间还要加 O(n)。
保存为 Main.java，运行 javac --release 17 -encoding UTF-8 Main.java，再运行 java Main。程序包含固定输入自测，断言失败时会抛出错误。
展开收起代码 · Main.java · 完整可运行程序
长行可在代码区域内左右滚动，按 Tab 可把焦点移入代码区。
import java.util.*;
public class Main {
record Range(int from, int to) {}
static List<Range> split(int n, int workers) {
if (n < 0 || workers <= 0) throw new IllegalArgumentException("counts");
List<Range> result = new ArrayList<>();
int base = n / workers, extra = n % workers;
int start = 0; // 已分配前缀的终点。
for (int i = 0; i < workers; i++) {
int length = base + (i < extra ? 1 : 0);
int end = Math.addExact(start, length); // 意外越界立即失败。
result.add(new Range(start, end));
start = end;
}
return result;
}
static void verify(int n, int workers) {
List<Range> parts = split(n, workers);
long total = 0;
int previousEnd = 0, min = Integer.MAX_VALUE, max = 0;
for (Range part : parts) {
if (part.from != previousEnd || part.to < part.from)
throw new AssertionError("gap or overlap");
int length = part.to - part.from;
total += length; // 使用 long 累计，独立检查总数。
min = Math.min(min, length); max = Math.max(max, length);
previousEnd = part.to;
}
if (total != n || previousEnd != n || max - min > 1)
throw new AssertionError("coverage or balance");
}
public static void main(String[] args) {
if (!split(8, 3).equals(List.of(new Range(0, 3),
new Range(3, 6), new Range(6, 8))))
throw new AssertionError("known partition");
verify(8, 3);
verify(0, 3); // 全空区间。
verify(2, 4); // 工作者多于任务。
verify(Integer.MAX_VALUE, 2); // 不创建二十亿个对象。
boolean bad = false;
try { split(3, 0); } catch (IllegalArgumentException e) { bad = true; }
if (!bad) throw new AssertionError("zero workers");
System.out.println("PASS 6");
}
}这些特殊情况也要试一试
- 零任务应产生 c 个空区间，不是除零。
- 工作者多于任务时，后面工作者拿到空区间。
- 数量平衡依赖任务代价近似相同，热分片会破坏吞吐平衡。
- 各参与者对清单顺序的理解必须一致。
进阶选读 · 带权任务调度与最小化最大负载
场景任务耗时为 [9,8,7,2,1]，均分数量不能平衡三个工作者的处理时间。
做法将任务按估计耗时降序，把当前最大任务放到总负载最小的工作者，使用最小堆维护各工作者负载。这是常见 LPT 启发式；若任务可任意分配，求最优最小最大负载的一般问题是 NP-hard，不能宣称贪心总是最优。若必须保持连续分片且权重非负，则可二分答案并贪心检验可行性。
复杂度任意分配的 LPT 为 O(n log n+n log c)，空间 O(n+c)。连续非负分片的二分可行性为 O(n log S)，S 是整数总权重范围；两者解决的约束不同。
好处与代价权重平衡可能增加任务迁移，破坏缓存局部性或顺序约束；耗时估计也可能过期。调度要同时考虑顺序、迁移成本和尾延迟。
05项目里怎么用
Topic 粒度应与业务隔离、权限和运维职责一致。所有业务塞进一个 Topic 会增加过滤与故障定位成本；每个用户创建一个 Topic 又可能产生资源管理压力。普通、顺序、延迟与事务消息还涉及不同类型约束，部署时应核对实际启用的类型检查策略。不要仅凭文档页面标题就假定服务器所有默认值与旧版本相同。
扩容前同时看生产速率、成功处理速率、积压年龄、重试和下游资源。若消费瓶颈是数据库连接数，增加消费者可能只是把更多线程送去等待。若问题是热消息组的顺序约束，则不能靠破坏组内顺序来换表面吞吐。先确定负载均衡粒度，再做受控增量扩容并验证业务正确性，才能解释扩容为什么有效或无效。
订阅配置要和代码版本一起审查。改变 Tag 表达式，会改变这个组应该收到哪些事件，不只是改变处理速度。逐批发布时，要确认新旧实例的订阅配置能配合工作，消费者也能读懂消息格式的变化。按 eventId 记录成功结果和重试原因，才能分别核对同一事件在每个组里处理到了哪一步，不能拿一个组的进度代表全部业务已经完成。
容易踩的坑
- 把 Keys 当作全局去重约束。
- 同组实例订阅不一致，导致事件覆盖范围不清。
- 混用经典客户端与新 SDK 的并发上限结论。
- 数量平均就断言业务负载平均。
- 为增加一份独立业务消费而只增加同组实例。
几种做法，怎么选
表格有四列，窄屏可左右滑动查看代价和适用条件。
| 方案 | 优势 | 代价与限制 | 适用条件 |
|---|---|---|---|
| 独立业务使用独立组 | 进度和失败策略各自管理。 | 额外消费成本与结果对账。 | 审计、积分等职责需要各自处理。 |
| 同组横向扩容 | 协作分担相同职责的工作。 | 受分配粒度、热点和下游容量约束。 | 处理逻辑一致且已验证可并行。 |
| 拆分 Topic | 权限、消息类型与运维隔离清晰。 | Topic 数量与配置管理成本增加。 | 业务容量、时效或消息类型差异明显。 |
06检验是否读懂了
先自己想、自己写，再展开答案核对。场景题是假设的练习情境；遇到还没学过的内容，可以做完基础后再回来。
先用自己的话解释
01Topic 和 ConsumerGroup 为什么不能互相替代？
展开参考答案与推导
Topic 组织业务消息与存储，组组织独立消费职责；同一个 Topic 可被多个组各自处理。
02n=2、c=4 的经典队列分配意味着什么？
展开参考答案与推导
队列粒度模式下最多两个实例获得非空范围，其余空闲；不能据此断言新消息粒度 SDK 也有同样关系。
03为何同组过滤规则必须一致？
展开参考答案与推导
组代表统一职责，规则不一致会让可见事件集合依赖具体接收实例，破坏可解释的消费与故障恢复。
换个情况，还会处理吗
01增加一倍消费者后积压未改善，如何定位？
展开参考答案与推导
核对 SDK 分配粒度和活跃任务数，再查热队列、顺序消息组、连接池等待与业务成功速率；避免只看进程数量。
02积分想各自收到订单组的事件，应如何接入？
展开参考答案与推导
使用独立积分组订阅对应 Topic 与过滤规则，维护自己的幂等结果，评估起始进度及历史补偿，而非加入订单服务组抢任务。
03两个实例算出同一队列归自己，先看哪类证据？
展开参考答案与推导
比较组成员视图、清单排序、版本与重平衡时间线；本地算法相同不代表输入视图相同，并检查旧任务是否尚未退出。
写代码试一试
01手工推导 10 个任务分给 4 个工作者。
- 输入 / 条件
n=10,c=4- 输出 / 预期
[0,3),[3,6),[6,8),[8,10)
提示 先算商 2 和余数 2。
展开参考答案与推导
前两个长度 3，后两个长度 2，起点逐段累加即可；检查最后端点为 10。
02只返回第 i 个区间，避免创建全部列表。
- 输入 / 条件
n=8,c=3,i=2- 输出 / 预期
[6,8)
提示 起点等于 i×base+min(i,extra)。
展开参考答案与推导
校验 i 范围后用公式计算，时间与额外空间 O(1)，并与完整分配结果逐项比较。
03为带权连续分片编写可行性判断。
- 输入 / 条件
weights=[7,2,5,10,8],c=2,limit=18- 输出 / 预期
true，可分 [7,2,5] 与 [10,8]
提示 非负权重让“尽量装满当前段”成立。
展开参考答案与推导
累计超过 limit 才新开段，任何单项大于 limit 立即失败，最后检查段数不超过 c；用于二分最小最大段和。
读完之后，独立完成
课后作业：自己做一遍
本篇共 9 道作业，先完成 7 道基础与综合任务，再按前置知识挑战进阶题。其中包含 1 道思维拓展，试着换个条件，看看原来的办法是否还管用。不附参考答案。
能为订单、积分和通知画出订阅关系，并解释同组消费者的作用。
本篇重点加练
组内分担依赖一致订阅，需同时检验分配身份和动态配置变更。
- 组内分担
- 订阅一致性
重点内容会多练几次。先说清楚怎么做，再试试空数据、出错或条件改变时会发生什么。
先自己做，卡住时记下问题，再回来查。问答题写清“为什么、什么时候适用”；动手题留下输入、输出和运行步骤，方便下次重做。
基础必做
- 作业 01原理问答
商城有订单状态事件和商品价格事件，二者保留期和消费延迟要求不同。讨论拆 Topic 与共用 Topic 的影响，并界定队列作为并行单位的意义。
- 提交内容
- 主题规划与取舍表
- 怎样算完成
- 注明业务需求和运维假设
- 区分逻辑主题与实际队列
练到的知识点Topic · Message Queue
- 作业 02代码推演
同一 Producer 产生六条带事件 ID 的订单消息，会员与财务分别独立处理。画出生产实例变化和消费者组重启时仍需保持的身份关系。
- 提交内容
- 身份关系图与事件账本字段
- 怎样算完成
- 区分 Producer 身份与业务 ID
- 标出每个 Consumer Group 的进度归属
练到的知识点Producer · Consumer Group
综合应用
- 作业 03面试场景
一个处理组有三个实例，另一组只有一个实例且停机十分钟。分析恢复前后两组的任务分担与进度核验，列出不能仅由实例数量推出的结论。
- 提交内容
- 分担示意与排查清单
- 怎样算完成
- 分别观察组内和组间
- 明确队列模型及版本前提
练到的知识点组内分担 · 组间独立
- 作业 04设计评审
同组实例 A 订阅 PAID，实例 B 订阅 PAID 或 REFUND，部署分批进行。评审这段过渡期并为订阅配置建立一致性检查。
- 提交内容
- 配置矩阵与发布条件
- 怎样算完成
- 覆盖实例加入及重启
- 说明过滤规则变化的生效范围
练到的知识点Tag 过滤 · 订阅一致性
- 作业 05编程实践
写本地 Java 消费分配模拟器，输入队列、实例和订阅声明，输出分配与配置冲突报告；模拟增减实例及两组同时存在。
- 提交内容
- 源码、样本和性质检查
- 怎样算完成
- 验证每组内任务归属
- 注明模拟器不验证真实 Broker 再均衡协议
练到的知识点Message Queue · Consumer Group · 组内分担 · 组间独立 · Tag 过滤 · 订阅一致性
- 作业 06设计评审
服务实例在启动后根据本机配置选择订阅标签，配置可能热更新。设计订阅声明的审计协议，分析同组成员的配置如何参与分配资格判断。
- 提交内容
- 订阅审计与分配资格表
- 怎样算完成
- 包含实例加入和退出
- 明确采用的客户端版本与消息模型
练到的知识点组内分担 · 订阅一致性
- 作业 07编程实践
写本地配置检查器，输入多个组的实例清单、标签表达式和队列订阅，找出同组差异与无人有资格处理的队列；仅核查声明。
- 提交内容
- 源码和差异样本
- 怎样算完成
- 把语法等价和文本相等的范围说清
- 不宣称验证真实 Broker 分配
练到的知识点组内分担 · 订阅一致性
进阶挑战 · 掌握前置后选做
- 作业 08设计评审
审计部门新增全量事件订阅，要求不影响现有积分进度。列出接入所需配置、历史读取边界和观察指标，并设计小规模本地验收。
- 提交内容
- 接入方案与验收步骤
- 怎样算完成
- 确认起始消费位置口径
- 区分新组隔离和 Broker 共享资源
练到的知识点Topic · Message Queue · Consumer Group · 组间独立 · Tag 过滤
思维拓展
先写出你的猜想，再用例子或实验验证。答案可以不同，但要说清依据、适用情况，以及还有什么没弄懂。
- 作业 09思维拓展 · 跨域迁移
将消费组迁移为多语言工作池，部分任务只有特定语言运行时能执行。订阅一致性与组内分担的原假设如何变化？
- 提交内容
- 能力模型与分组评审
- 怎样算完成
- 区分任务资格和负载均衡
- 包含某能力实例全部退出
练到的知识点组内分担 · 订阅一致性
完成后，再检验一次
给每题标记“独立完成 / 查阅后完成 / 尚未完成”。查过资料或没做出的题，记下卡在哪里。隔一天换一组输入再试，能解释结果和出错原因，再把这个知识点标为掌握。
继续阅读
先看能帮你完成当前练习的部分。例子跑通后，再读更深入的解释。
- 必读教程英文
RocketMQ：Topic （新标签页）↗
阅读 Topic、队列和消息类型，设计业务事件的命名与边界。
- 实践教程英文
RocketMQ：ConsumerGroup （新标签页）↗
阅读组内消费与组间订阅关系，规划独立业务的消费进度。
- 深入参考文档英文
RocketMQ：Subscription （新标签页）↗
阅读订阅过滤与组内一致性要求，避免不同消费者收到不一致的消息集合。
- 深入参考文档英文
RocketMQ：Consumer Load Balancing （新标签页）↗
对比队列粒度与消息粒度的分担模式。
- 深入参考文档英文
RocketMQ AVG 分配源码 （新标签页）↗
用真实固定标签对照区间均分公式。