适用范围：RocketMQ4.9.8与5.3.4固定提交；以5.3.4为主线。
本章目标：先把组件、协议、存储和消费方式分开，避免从一个类推断所有部署模式。
0.1 这本手册怎样读
全文以Apache RocketMQ的rocketmq-all-5.3.4标签为基线，固定到提交63d20eb92a4aa685ae0d0696b419d3ffb6ca1738。这是有意固定的源码版本，不是“始终最新”。本仓库里的client模块主要讲经典Java Remoting客户端；5.x多语言gRPC客户端位于另一仓库，本文通过服务端Proxy入口解释协议，不把两个客户端当作同一个实现。
每节先看机制图，再看真实代码，最后看变量和分支。代码是带固定提交链接的连续节选，保留原算法，只统一展示缩进。窗口可能止于方法中间；完整逻辑以链接中的方法和调用方为准。所有推演均为纸面阅读，没有要求运行Java或搭建集群。
源码对照：BrokerController.java · L877–L890。以下为连续节选，可能止于方法中间；仅统一缩进，完整方法与调用方见链接。
public boolean initialize() throws CloneNotSupportedException {
boolean result = this.initializeMetadata();
if (!result) {
return false;
}
result = this.initializeMessageStore();
if (!result) {
return false;
}
return this.recoverAndInitService();
}逐段阅读抓手：initialize是生命周期入口，不是发送入口；先确认依赖加载是否成功，再看服务启动。
0.2 四组容易混淆的路径
四个维度分别回答不同问题：Remoting/gRPC是通信入口；Pull/POP是消费协议；文件CQ/RocksDB CQ是索引后端；传统HA/DLedger/Controller是复制和切换机制。它们不能简单排列成“4.x旧、5.x新”。5.3.4仍包含多条兼容路径。
DefaultMessageStore的索引工厂还包含双写迁移分支。RocksDBMessageStore有自己的实现。看到一个20字节CQ记录，只能说明当前读的是文件ConsumeQueue，不能说所有存储都按这个布局写。
源码对照：DefaultMessageStore.java · L268–L273。以下为连续节选，可能止于方法中间；仅统一缩进，完整方法与调用方见链接。
public ConsumeQueueStoreInterface createConsumeQueueStore() {
if (messageStoreConfig.isRocksdbCQDoubleWriteEnable()) {
return new CombineConsumeQueueStore(this);
}
return new ConsumeQueueStore(this);
}逐段阅读抓手：先辨认实际对象类型；同名接口方法可以落到不同存储实现。
0.3 源码目录对应哪些职责
client负责客户端路由、发送和消费调度；namesrv负责路由注册与查询；remoting负责RPC协议和异步响应；broker负责请求处理、消费元数据及业务消息服务；store负责日志、索引和复制；proxy负责gRPC入口与消息处理编排；controller负责自动切换相关元数据与决策。
目录名就是第一层责任边界。Proxy不会凭空替代CommitLog，NameServer不承载消息体，也不会因为它有“集群信息表”就变成消息一致性的仲裁者。
源码对照：DefaultMessagingProcessor.java · L139–L144。以下为连续节选，可能止于方法中间；仅统一缩进，完整方法与调用方见链接。
protected void init() {
this.appendStartAndShutdown(this.serviceManager);
this.appendStartAndShutdown(this.receiptHandleProcessor);
this.appendShutdown(this.producerProcessorExecutor::shutdown);
this.appendShutdown(this.consumerProcessorExecutor::shutdown);
}逐段阅读抓手：DefaultMessagingProcessor把不同操作交给对应Processor；继续进入具体操作，不要停在接口转发层。
0.4 4.x与5.x对照：两个基线与“升级”究竟比较什么
| 对照维度 | 固定4.9.8 | 固定5.3.4 |
|---|---|---|
| 实现 | 4.9.8保留经典Remoting客户端、队列分配、等级延迟、事务消息与传统HA/DLedger等路径。 | 5.3.4同时保留兼容路径并增加Proxy/gRPC、POP、时间轮定时、Controller及新的存储分支等。 |
变化原因（源码分析）：兼容既有生态，同时提供新的消费和部署模型。这里比较固定发行版差异，不把某特性在5.3.4出现说成必定于5.0.0首次发布。
适用边界：升级Broker版本不会自动把旧Remoting Consumer变成gRPC/POP；源码有某分支，也不表示配置已启用。
4.9.8源码：BrokerController.java · L241–L483，连续节选。
public boolean initialize() throws CloneNotSupportedException {
boolean result = this.topicConfigManager.load();
result = result && this.consumerOffsetManager.load();
result = result && this.subscriptionGroupManager.load();
result = result && this.consumerFilterManager.load();
if (result) {
try {
this.messageStore =
new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,
this.brokerConfig);
if (messageStoreConfig.isEnableDLegerCommitLog()) {
DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, (DefaultMessageStore) messageStore);
((DLedgerCommitLog)((DefaultMessageStore) messageStore).getCommitLog()).getdLedgerServer().getdLedgerLeaderElector().addRoleChangeHandler(roleChangeHandler);
}
this.brokerStats = new BrokerStats((DefaultMessageStore) this.messageStore);
//load plugin
MessageStorePluginContext context = new MessageStorePluginContext(messageStoreConfig, brokerStatsManager, messageArrivingListener, brokerConfig);
this.messageStore = MessageStoreFactory.build(context, this.messageStore);
this.messageStore.getDispatcherList().addFirst(new CommitLogDispatcherCalcBitMap(this.brokerConfig, this.consumerFilterManager));
} catch (IOException e) {
result = false;
log.error("Failed to initialize", e);
}
}
result = result && this.messageStore.load();
if (result) {
this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.clientHousekeepingService);
NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();
fastConfig.setListenPort(nettyServerConfig.getListenPort() - 2);
this.fastRemotingServer = new NettyRemotingServer(fastConfig, this.clientHousekeepingService);
this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(
this.brokerConfig.getSendMessageThreadPoolNums(),
this.brokerConfig.getSendMessageThreadPoolNums(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.sendThreadPoolQueue,
new ThreadFactoryImpl("SendMessageThread_"));
this.putMessageFutureExecutor = new BrokerFixedThreadPoolExecutor(
this.brokerConfig.getPutMessageFutureThreadPoolNums(),
this.brokerConfig.getPutMessageFutureThreadPoolNums(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.putThreadPoolQueue,
new ThreadFactoryImpl("PutMessageThread_"));
this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(
this.brokerConfig.getPullMessageThreadPoolNums(),
this.brokerConfig.getPullMessageThreadPoolNums(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.pullThreadPoolQueue,
new ThreadFactoryImpl("PullMessageThread_"));
this.replyMessageExecutor = new BrokerFixedThreadPoolExecutor(
this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.replyThreadPoolQueue,
new ThreadFactoryImpl("ProcessReplyMessageThread_"));
this.queryMessageExecutor = new BrokerFixedThreadPoolExecutor(
this.brokerConfig.getQueryMessageThreadPoolNums(),
this.brokerConfig.getQueryMessageThreadPoolNums(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.queryThreadPoolQueue,
new ThreadFactoryImpl("QueryMessageThread_"));
this.adminBrokerExecutor =
Executors.newFixedThreadPool(this.brokerConfig.getAdminBrokerThreadPoolNums(), new ThreadFactoryImpl(
"AdminBrokerThread_"));
this.clientManageExecutor = new ThreadPoolExecutor(
this.brokerConfig.getClientManageThreadPoolNums(),
this.brokerConfig.getClientManageThreadPoolNums(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.clientManagerThreadPoolQueue,
new ThreadFactoryImpl("ClientManageThread_"));
this.heartbeatExecutor = new BrokerFixedThreadPoolExecutor(
this.brokerConfig.getHeartbeatThreadPoolNums(),
this.brokerConfig.getHeartbeatThreadPoolNums(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.heartbeatThreadPoolQueue,
new ThreadFactoryImpl("HeartbeatThread_", true));
this.endTransactionExecutor = new BrokerFixedThreadPoolExecutor(
this.brokerConfig.getEndTransactionThreadPoolNums(),
this.brokerConfig.getEndTransactionThreadPoolNums(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.endTransactionThreadPoolQueue,
new ThreadFactoryImpl("EndTransactionThread_"));
this.consumerManageExecutor =
Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(
"ConsumerManageThread_"));
this.registerProcessor();
final long initialDelay = UtilAll.computeNextMorningTimeMillis() - System.currentTimeMillis();
final long period = 1000 * 60 * 60 * 24;
this.scheduledExecutorService.scheduleAtFixedRate(() -> {
try {
BrokerController.this.getBrokerStats().record();
} catch (Throwable e) {
log.error("schedule record error.", e);
}
}, initialDelay, period, TimeUnit.MILLISECONDS);
this.scheduledExecutorService.scheduleAtFixedRate(() -> {
try {
BrokerController.this.consumerOffsetManager.persist();
} catch (Throwable e) {
log.error("schedule persist consumerOffset error.", e);
}
}, 1000 * 10, this.brokerConfig.getFlushConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
this.scheduledExecutorService.scheduleAtFixedRate(() -> {
try {
BrokerController.this.consumerFilterManager.persist();
} catch (Throwable e) {
log.error("schedule persist consumer filter error.", e);
}
}, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);
this.scheduledExecutorService.scheduleAtFixedRate(() -> {
try {
BrokerController.this.protectBroker();
} catch (Throwable e) {
log.error("protectBroker error.", e);
}
}, 3, 3, TimeUnit.MINUTES);
this.scheduledExecutorService.scheduleAtFixedRate(() -> {
try {
BrokerController.this.printWaterMark();
} catch (Throwable e) {
log.error("printWaterMark error.", e);
}
}, 10, 1, TimeUnit.SECONDS);
this.scheduledExecutorService.scheduleAtFixedRate(() -> {
try {
log.info("dispatch behind commit log {} bytes", BrokerController.this.getMessageStore().dispatchBehindBytes());
} catch (Throwable e) {
log.error("schedule dispatchBehindBytes error.", e);
}
}, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
if (this.brokerConfig.getNamesrvAddr() != null) {
this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
log.info("Set user specified name server address: {}", this.brokerConfig.getNamesrvAddr());
this.scheduledExecutorService.scheduleAtFixedRate(() -> {
try {
BrokerController.this.brokerOuterAPI.updateNameServerAddressList(BrokerController.this.brokerConfig.getNamesrvAddr());
} catch (Throwable e) {
log.error("ScheduledTask updateNameServerAddr exception", e);
}
}, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
} else if (this.brokerConfig.isFetchNamesrvAddrByAddressServer()) {
this.scheduledExecutorService.scheduleAtFixedRate(() -> {
try {
BrokerController.this.brokerOuterAPI.fetchNameServerAddr();
} catch (Throwable e) {
log.error("ScheduledTask fetchNameServerAddr exception", e);
}
}, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
}
if (!messageStoreConfig.isEnableDLegerCommitLog()) {
if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
if (this.messageStoreConfig.getHaMasterAddress() != null && this.messageStoreConfig.getHaMasterAddress().length() >= 6) {
this.messageStore.updateHaMasterAddress(this.messageStoreConfig.getHaMasterAddress());
this.updateMasterHAServerAddrPeriodically = false;
} else {
this.updateMasterHAServerAddrPeriodically = true;
}
} else {
this.scheduledExecutorService.scheduleAtFixedRate(() -> {
try {
BrokerController.this.printMasterAndSlaveDiff();
} catch (Throwable e) {
log.error("schedule printMasterAndSlaveDiff error.", e);
}
}, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
}
}
if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {
// Register a listener to reload SslContext
try {
fileWatchService = new FileWatchService(
new String[] {
TlsSystemConfig.tlsServerCertPath,
TlsSystemConfig.tlsServerKeyPath,
TlsSystemConfig.tlsServerTrustCertPath
},
new FileWatchService.Listener() {
boolean certChanged, keyChanged = false;
@Override
public void onChanged(String path) {
if (path.equals(TlsSystemConfig.tlsServerTrustCertPath)) {
log.info("The trust certificate changed, reload the ssl context");
reloadServerSslContext();
}
if (path.equals(TlsSystemConfig.tlsServerCertPath)) {
certChanged = true;
}
if (path.equals(TlsSystemConfig.tlsServerKeyPath)) {
keyChanged = true;
}
if (certChanged && keyChanged) {
log.info("The certificate and private key changed, reload the ssl context");
certChanged = keyChanged = false;
reloadServerSslContext();
}
}
private void reloadServerSslContext() {
((NettyRemotingServer) remotingServer).loadSslContext();
((NettyRemotingServer) fastRemotingServer).loadSslContext();
}
});
} catch (Exception e) {
log.warn("FileWatchService created error, can't load the certificate dynamically");
}
}
initialTransaction();
initialAcl();
initialRpcHooks();
}
return result;
}5.3.4源码：BrokerController.java · L877–L890，连续节选。
public boolean initialize() throws CloneNotSupportedException {
boolean result = this.initializeMetadata();
if (!result) {
return false;
}
result = this.initializeMessageStore();
if (!result) {
return false;
}
return this.recoverAndInitService();
}对照读法：先找输入条件，再标记状态保存在哪个组件，最后比较成功确认和故障恢复的触发点。类名变化不一定表示协议改变；新增分支也不代表旧路径消失。
0.5 先把RocketMQ看成三本账
读源码时最容易走丢，是把“消息存在”“消息可读”和“业务做完”看成同一个状态。经典文件存储可以先看成三本账：CommitLog记录发生过什么，ConsumeQueue记录到哪里找，消费位点记录某个Group下次从哪里继续。IndexFile是按Key查询的辅助入口，不替代按队列消费的CQ。
| 账本 | 谁维护 | 记录的单位 | 回答的问题 | 不能证明什么 |
|---|---|---|---|---|
| CommitLog | Broker的CommitLog追加路径 | 物理字节位置与完整消息记录 | 日志中是否有这条记录 | 不能单凭追加证明已刷盘、已建CQ或业务成功 |
| ConsumeQueue | Reput分发到CQ实现 | 某个Topic/Queue的逻辑序号→物理位置 | 给定queueOffset去哪读消息 | 不能证明任何Group处理完成 |
| consumerOffset | 客户端OffsetStore及Broker位点管理 | Group/Topic/Queue的进度 | 该组重启或重新分配后从哪里继续 | 不能证明数据库副作用恰好发生一次 |
为什么拆开？多个Topic/Queue共享物理追加日志，可把许多写入汇入顺序写；消费又需要按各自Queue读，所以由索引把逻辑序列映射回物理日志；不同Group独立订阅同一批消息，所以进度必须另存。结构的收益与代价是一体的：写入和索引解耦提升处理弹性，也引入追加成功而暂时不可见的窗口。
这只是经典文件主线的模型。RocksDB CQ改变索引后端；POP增加逐次投递的在途记录；DLedger改变日志提交路径。先掌握账本职责，再去看这些分支怎样换实现。
源码接续：第1章的三种offset→第7章的追加→第10章的索引分发→第17章的进度持久化。阅读时在每个类旁边写下“它管哪本账”，再读方法。
0.6 用四条消息手算：为什么三个offset不能互换
假设只讨论同一个Broker的经典文件主线，以下物理位置和长度均为讲解自拟值，不包含文件尾滚动。TopicA有Q0、Q1，TopicB有Q0。表中长度是整条编码记录的长度，包含元数据，并非只有Body。
| 物理追加顺序 | Topic/Queue | queueOffset | commitLogOffset | 记录长度 | 物理结束位置 |
|---|---|---|---|---|---|
| M1 | A/Q0 | 0 | 1000 | 120 | 1120 |
| M2 | B/Q0 | 0 | 1120 | 160 | 1280 |
| M3 | A/Q0 | 1 | 1280 | 140 | 1420 |
| M4 | A/Q1 | 0 | 1420 | 180 | 1600 |
从这张表推出四个结论：A/Q0的逻辑offset连续为0、1，但物理位置隔着B/Q0的记录；A/Q1和B/Q0各自也从0开始；同一条M3对Group甲和Group乙的queueOffset相同，消费进度却可以不同；刷盘等待M3时关注的是完整记录结束位置1420，不是逻辑offset1。
经典文件CQ的一条记录占20字节，所以A/Q0的逻辑offset1对应CQ逻辑字节位置1×20=20，该槽保存物理位置1280、消息长度140和tagsCode。CQ文件自身也会分段；找到槽后再读取CommitLog。乘20定位的是CQ槽，不是消息体，也不是CommitLog偏移。
若Group甲已处理A/Q0的M1，通常把下一读取位置推进到1；Group乙还没处理则可能停在0。甲的consumerOffset=1表示下次从M3开始，并不表示“物理日志只刷到了1字节”。
闭眼自检：M3已进入CommitLog，但CQ分发尚未赶到1280，消费者能否仅靠A/Q0的offset1马上读到它？不能，按队列消费的索引入口尚未建立。业务监听器完成M3后，能否立即物理删除M3？不能，其他Group、回溯与日志文件保留都有自己的生命周期。
0.7 一次send成功经过了什么，哪些动作仍在后面
普通Remoting同步发送主线从DefaultMQProducerImpl.sendDefaultImpl开始：拿路由、选Queue、计算剩余超时、通过Remoting发请求。Broker按请求码进入SendMessageProcessor，校验并构造内部消息，再进入MessageStore和CommitLog。最终响应来自存储结果映射，而消费在另一个推进过程里。
图里从追加处分叉，表示发送响应路径与消费可见性路径分别推进，不表示两者有固定的先后竞赛结果。重负载时Producer先拿到结果、Reput随后赶上；另一些时序下索引已生成，Producer仍在等待刷盘或复制确认。
| 完成点 | 源码抓手 | 此时能说什么 | 下一步还要问什么 |
|---|---|---|---|
| 追加完成 | AppendMessageResult中的位置与长度 |
当前追加路径接受了该记录 | 可读位置、刷盘与副本到了哪里？ |
| 发送结果返回 | 刷盘/复制Future与SendMessageProcessor状态映射 |
满足了本路径实际启用的确认条件，或报告相应错误 | 结果是否到达Producer？CQ是否赶上？ |
| 业务CQ可见 | Reput与CQ分发器 | 按此Queue查询可定位消息 | 消费者有没有拉到、执行是否成功？ |
| 业务事务提交 | 应用自己的事务事实 | 业务副作用按其事务约束生效 | 消费确认是否送达、重投如何判重？ |
| 进度持久化 | OffsetStore及Broker位点管理 | 重启可按已保存进度续读 | 日志还在保留范围内吗？是否允许回溯？ |
关键反例：同步刷盘配置下，刷盘等待超时可以发生在记录已追加之后。把FLUSH_DISK_TIMEOUT或RPC超时一律理解为“Broker没收过”，会在重发时制造重复。代码中的Future超时结束的是一次等待，不会倒着擦除已追加日志。
源码接续：第4章、第6章、第9章、第10章。先追成功判定，再追超时返回；不要只背调用栈。
0.8 三个写入位置：画清wrote、committed、flushed
这三个字段首先是单个MappedFile内部的位置；MappedFileQueue的全局位置还要加上文件起始offset。讨论数值大小前，先确定是否启用暂存缓冲。以下限定经典文件路径，省略替代写入方式。
| 路径 | 追加后数据在哪 | 什么位置可读 | commit做什么 | flush做什么 |
|---|---|---|---|---|
| 没有writeBuffer的映射追加 | 映射区域/PageCache | 以wrotePosition为可读边界 | 不存在把暂存区搬入文件的必要步骤 | 对可读数据执行相应force并推进刷盘位置 |
| 有writeBuffer的暂存追加 | 暂存缓冲 | 以committedPosition为可读边界 | 把wrote与committed之间的字节写入文件通道 | 对已提交的可读数据force并推进刷盘位置 |
不能无条件写成flushed≤committed≤wrote后套给所有配置。无暂存缓冲时，getReadPosition()直接取wrotePosition，committedPosition不是该路径读取的门槛。有暂存缓冲时，三个阶段才构成“已追加→已提交到文件→已刷盘”的推进关系。
假设有暂存缓冲，一个文件起始全局offset为0：初始W=C=F=1000；追加M1后W=1120、C=F=1000；commit后W=C=1120、F=1000；flush后W=C=F=1120。第一步后，暂存区有M1，却还不能据此认为文件读取路径或HA已能读到M1。第二步后，进程可从文件侧读取，断电持久化保证还取决于后续刷盘。
读源码的顺序：第8章中先看appendMessage写哪个Buffer，再看getReadPosition选择哪个位置，最后看commit0、flush和队列全局位置怎样组合。这样才能解释为什么“写入吞吐很高”与“同步刷盘RT很高”可以同时发生。
0.9 消费不是一个ACK：先追本地缓存，再追远端进度
经典PushConsumer用后台Pull主动取消息。Broker长轮询挂起的是请求上下文，不是为每条请求占住一个业务线程不停等待。客户端把拉取结果放进ProcessQueue，再交给消费线程池。于是同一个Queue至少有三种客户端状态：下次Pull位置、本地未完成集合、安全消费位点；三者职责不同。
下面假设正常集群并发消费，消息100、101、102已拉入同一ProcessQueue，没有过期清理或队列撤销。pull的nextOffset可以已经到103，consumerOffset却仍在100。消费者已“取走”不等于已“处理完”。
| 步骤 | ProcessQueue剩余记录 | 本次removeMessage结果 | 内存安全位点 | Broker持久化位点 |
|---|---|---|---|---|
| 都在处理 | 100、101、102 | 尚未移除 | 100 | 例如仍为100 |
| 102先成功 | 100、101 | 100 | 100 | 不应据此跳到103 |
| 100再成功 | 101 | 101 | 101 | 可能仍为100，尚未完成提交 |
| 101最后成功 | 空 | 103 | 103 | 完成持久化后才可到103 |
为什么回调不能“谁先完成就把offset写成谁+1”？102先完成就提交103，重启后会跳过100、101。源码用有序的msgTreeMap保留尚未交接完成的记录，removeMessage取最小残留位置，空时返回queueOffsetMax+1。它保护的是本地待处理集合下的续读边界；普通Pull过滤掉的消息无需假装也都在这个集合里。
101业务失败怎么办？若sendMessageBack成功，失败责任交给Broker重试消息，原Queue可以继续推进；回送失败则把101排除出本次可删除集合，保留本地并延后消费。允许原位点越过失败消息的理由，是重试责任已被安全交接，绝不是“失败也算成功”。
最后还有一条裂缝：数据库已提交，但进度尚未持久化时宕机，重启可能再消费；业务幂等必须依据稳定业务事件标识，而非只看ProcessQueue是否为空。广播、顺序服务与POP各有自己的分支，不能套这张并发集群表。
源码接续：第13章的ProcessQueue→第14章的Pull循环→第15章的结果处理→第17章的持久化。
0.10 失败由谁接手：把重试、事务回查与POP恢复分开
三个机制都会“以后再做”，但补的是三个不同的缺口。
| 机制 | 要弥补的缺口 | 保存什么依据 | 谁推动恢复 | 应用仍需做什么 |
|---|---|---|---|---|
| 经典消费重试 | Listener未完成处理 | 回送后的重试消息及重试次数等信息 | Broker调度，消费者再次处理 | 幂等、可重试分类、死信处置 |
| 事务回查 | Broker不知道生产者本地事务结果 | half及op处理记录、业务事务标识 | Broker发起检查，Producer查本地事务事实 | 可靠查询COMMIT/ROLLBACK/未知，不能凭猜测返回 |
| POP超时恢复 | 投递过但没有完成有效ACK | checkpoint/ACK或KV在途记录 | 到期后的Revive或KV恢复路径 | 处理与ACK分离，续期、超时与重复处理控制 |
事务纸面时序：half先被接受→本地订单事务提交→Producer还没送达COMMIT就宕机。Broker看不到订单数据库，无法自行判断订单成功，只能回查Producer端的持久化事实。回查若只查原进程内的Map，进程重启后事实丢失，便无法可靠收敛。若数据库暂不可读，应区分未知与已回滚；查不到是否代表回滚，取决于事务记录、查询隔离与业务状态设计。
COMMIT处理把最终消息写回原业务Topic，再记录half已处理的op。这里不是把磁盘上的half原地翻一个布尔位，也不是马上删除旧字节。最终业务消息与half处于不同逻辑队列，offset会变化。失败或重复处理窗口仍需结合完整源码追踪；下游数据库也没有自动加入生产者事务。官方事务说明明确区分消息生产与本地事务的一致性，以及下游消费处理责任。
POP纸面时序：消费者甲取到M，获得本次投递的收据和不可见期限；甲执行很慢，期限到期且缺乏有效ACK/续期；恢复路径使M有机会再投递给乙；甲后来也提交了业务。不可见期限限制的是本次投递可见性，不是给甲的数据库加锁，也不是永久所有权。两个业务处理可能重叠，所以需要判重及业务并发约束。
ReceiptHandle带本次投递的上下文，业务事件ID表达“这件业务是什么”。前者不能代替后者。旧checkpoint/ReviveTopic与KV在途记录是替代实现分支，必须看实际开关，不能把两个后端同时画成每条消息必经路径。
源码接续：第18章重试、第20章事务、第25章POP、第26章KV/收据。
0.11 把顺序、吞吐和可靠性放在同一张因果表里
| 设计选择 | 为什么这样做 | 由此产生的代价 | 延伸问题怎么接 |
|---|---|---|---|
| 共享CommitLog顺序追加 | 汇聚写入，减少分散随机写 | 物理追加仍要协调，共享路径可能形成瓶颈 | 先拆队列锁与物理追加锁，再谈锁内工作 |
| CQ异步派生 | 把日志事实与消费索引解耦 | 追加到可见之间有分发滞后 | 发送成功但拉不到，先确认CQ与分发位置 |
| 并发消费+安全位点 | 让后续记录并行做，又不越过未完成集合 | 最早未完成记录拖住原队列进度 | 失败重试成功交接后，原位点为什么可以前进？ |
| 同Queue顺序消费 | 限制同一序列内的并发执行 | 头部失败阻塞后续，吞吐受单条RT牵制 | 增加Queue只提高多个独立序列的并发，不能凭空保全局顺序 |
| 异步刷盘/异步复制 | 减少发送路径等待 | 确认之后仍有数据未持久化或未复制窗口 | 指定进程故障、主机故障或断电，再说可能丢在哪里 |
| POP不可见期限 | 用时限回收失联消费者的投递责任 | 超时后旧、新处理可能重叠 | 续期减少重投窗口，业务幂等仍需单独证明 |
经典顺序消息需要把“同一个业务序列稳定路由到同一Queue”和“该Queue按顺序执行”同时说清。只启用顺序Listener、发送时却随机选Queue，不会得到同一订单跨Queue顺序。不同Producer并发发送时，“数据库中先发生”也不自动等于“Broker先追加”。5.x消息组FIFO要按对应SDK和服务端分支解释，不能靠旧顺序服务的Queue锁代替。
吞吐算例也要明确范围：假设某个必须串行执行的业务序列，平均一条占用10ms，且不计其他开销，则该序列的处理上限约为100条/秒。这是讲解用的服务时间估算，不是RocketMQ基准值。加消费者若没有新的可分配Queue，经典队列分配模式下不会自动拆开这个序列；消息级负载均衡也不能免除FIFO序列约束。官方负载均衡说明按消费者类型区分队列级与消息级分配。
一分钟复述：RocketMQ把物理日志、逻辑索引和消费进度分开。发送先选队列，经Broker追加日志，按配置等待刷盘和副本确认；CQ分发决定按队列消费的可见性。经典Push是后台Pull，消费完成顺序可以不同，所以用本地未完成集合算安全位点，失败则交给重试路径。事务消息解决生产者本地事务与消息可见性的收敛，POP用不可见期限和逐投递ACK管理在途责任。每个阶段都存在确认丢失或宕机窗口，业务幂等、顺序边界和日志保留需要分别设计。
能复述不等于读懂，试着继续答这六个追问：
- 为什么SEND_OK后CQ仍可能没赶上？答出异步分发和不同完成点，见0.7。
- 为什么committedPosition不能总作可读边界？答出writeBuffer分支，见0.8。
- 102已成功而位点仍在100，是重复消费Bug吗？答出安全边界及持久化窗口，见0.9。
- 失败101为什么允许从原Queue移除？答出重试责任交接及回送失败保留，见0.9。
- 事务回查为什么不能只查内存？答出进程重启与持久化业务事实，见0.10。
- 不可见时间延长能否保证只扣款一次？不能，它管理投递期限，数据库重复副作用需要业务约束，见0.10。
第一轮读到这里，应能画出消息和三本账的关系；第二轮回到各章的源码卡，验证每一次状态前进的条件；第三轮读版本对照和特殊路径，检查哪些前提已经变化。详细源码保留在原章，不要求逐行背诵，也不要求搭建集群或运行实验。
本章纸面推演
第一遍沿发送→CommitLog→CQ→拉取→消费位点读完；第二遍补事务、定时、HA、Proxy与POP。遇到复杂分支，先写下自己的客户端协议和Broker配置，再进入对应章节。