六、消息的生产与发送
Producer 启动流程与路由获取
我们先从 Producer 的启动开始。一个 Producer 启动的时候，它可不是傻乎乎地直接就开始发消息了——它得先弄清楚“把消息发到哪里去”。

Producer 启动的核心步骤：

创建 Producer 实例

1. 设置 ProducerGroup 名称
   （同一个业务使用同一个 Group）

2. 设置 NameServer 地址列表
   （指定集群注册中心的地址）

3. 配置其他参数
   （发送超时时间、重试次数等）

4. 调用 start 方法启动 Producer

5. 初始化 MQClientInstance
   （客户端核心实例，负责网络通信和路由管理）

6. 启动 Netty 通信客户端
   （建立与 NameServer 的连接）

7. 从 NameServer 拉取所有 Topic 的路由信息

8. 启动定时任务
   （每 30 秒更新一次路由信息）

Producer 就绪，等待发送消息

关键点解析：

ProducerGroup：一个业务标识，同一个业务里的 Producer 实例归属于同一个 Group。在事务消息中，同一个 Group 的 Producer 可以互相回查事务状态。

MQClientInstance：这是客户端最核心的实例，所有 Producer 和 Consumer 共用一个 MQClientInstance（同一个 JVM 里）。它负责：

管理与 NameServer 的连接
管理与 Broker 的连接
维护本地路由缓存
统一的心跳和网络通信
路由获取：Producer 启动时，会从 NameServer 拉取所有 Topic 的路由信息（不仅仅是某个特定 Topic），这样后续发消息时就不需要再等路由查询了。

定时更新：路由信息在本地缓存后，会有一个定时任务每隔 30 秒 从 NameServer 拉取最新路由，保证路由信息的时效性。

消息发送的三种方式：同步、异步、单向
在《入门认知篇》里我们简单提过三种发送方式，现在我们从底层实现的角度再来看一遍。

1. 同步发送（Sync）

这是最常用、最简单的方式。发送消息后，线程阻塞等待 Broker 返回响应，收到响应后才继续执行。

SendResult sendResult = producer.send(msg);
// 阻塞等待，直到收到 Broker 的响应
Broker
Producer
Broker
Producer
线程阻塞等待
线程恢复
⏱️ 整个过程中线程处于阻塞状态

1. 构造消息

2. 发送请求（同步）

3. 返回响应（成功/失败）

4. 根据结果处理业务逻辑
   适用场景：关键业务——比如下单成功后的订单消息，必须确认 Broker 收到了才能继续。

5. 异步发送（Async）

发送消息后，线程不阻塞，立即返回。等 Broker 响应回来后，通过回调函数来处理结果。

producer.sendAsync(msg, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
// 处理成功逻辑
}
@Override
public void onException(Throwable e) {
// 处理失败逻辑
}
});
// 这里立即返回，不阻塞
Broker
Producer 异步线程
Producer 主线程
Broker
Producer 异步线程
Producer 主线程
主线程立即返回
可以继续其他工作
✅ 主线程未被阻塞
⏱️ 响应处理在异步线程中

1. 构造消息

2. 提交异步发送任务

3. 发送请求

4. 返回响应

5. 回调 SendCallback
   适用场景：对延迟敏感但需要知道结果——比如前端请求发消息，不能阻塞用户操作，但失败时需要通知用户。

6. 单向发送（Oneway）

只发不管，连响应都不等。最轻量、最快，但最不可靠。

producer.sendOneway(msg);
// 不管结果，直接继续
Broker
Producer
Broker
Producer
发完就走
不等待任何响应
🚀 最快，但没有任何可靠性保证

1. 构造消息
2. 发送请求（单向）
3. 继续执行后续代码
   适用场景：日志上报、监控数据等——丢了就丢了，业务上能接受。

💡 小贴士：同步发送和异步发送虽然看起来差异很大，但在底层，它们都使用了异步网络通信（Netty）。同步发送只是在异步通信的基础上，用 CountDownLatch 等同步工具做了一个“阻塞等待”的封装。本质上，RocketMQ 的网络通信模型是全异步的。

消息发送的负载均衡策略
当一个 Topic 有多个 MessageQueue（分布在不同的 Broker 上）时，Producer 如何选择将消息发到哪个 Queue？这就涉及负载均衡策略。

RocketMQ 默认提供了两种内置策略，你也可以自定义实现：

策略一：轮询（Round Robin）—— 默认策略

依次轮流选择 Queue，确保消息均匀分布。在顺序消息场景下，会结合消息 Key 做哈希，保证同一个 Key 的消息落到同一个 Queue。

MessageQueue

消息序列

第 1 条

第 2 条

第 3 条

第 4 条

第 5 条

第 6 条

消息 1

消息 2

消息 3

消息 4

消息 5

消息 6

Queue 0

Queue 1

Queue 2

按顺序轮流分配：1→Q1, 2→Q2, 3→Q3, 4→Q1…

策略二：一致性哈希（Consistent Hash）

根据消息 Key 的哈希值，通过哈希环来决定消息发到哪个 Queue。保证同一个 Key 的消息始终落在同一个 Queue 上（对于顺序消息特别重要）。

消息 Key

哈希值映射到环上

哈希值映射到环上

哈希值映射到环上

顺时针查找第一个节点

顺时针查找第一个节点

顺时针查找第一个节点

一致性哈希环

节点 1
Queue 0

节点 2
Queue 1

节点 3
Queue 2

节点 4
Queue 3

节点 5
Queue 0

节点 6
Queue 1

order_123

order_456

order_789

位置 A

位置 B

位置 C

同一 Key 始终落到同一 Queue
新增节点时只影响局部 Key 的重新分配

自定义策略：实现 MessageQueueSelector 接口，根据自己的业务逻辑选择 Queue（比如按订单 ID 取模）。

消息发送的重试机制与故障规避
消息发送不是总能一次成功的。网络抖动、Broker 故障、磁盘满了……各种原因都可能导致发送失败。RocketMQ 有一套完善的重试和故障规避机制。

重试机制

是

否

否

是

是

否

是

否

发送消息

第 1 次尝试

成功？

返回成功

是否为
可重试的异常？

直接返回失败
不重试

第 2 次尝试
等待 50ms

成功？

… 最多重试 retryTimesWhenSendFailed 次

最后一次成功？

重试的关键配置：

参数 默认值 说明
retryTimesWhenSendFailed 2 同步发送失败时的重试次数（不含第一次）
retryTimesWhenSendAsyncFailed 2 异步发送失败时的重试次数（不含第一次）
retryAnotherBrokerWhenNotStoreOK false 同步模式下，如果写入失败是否换 Broker 重试
哪些异常会触发重试？

✅ 可重试：网络超时、连接异常、Broker 返回 SYSTEM_BUSY、SERVICE_NOT_AVAILABLE 等
❌ 不重试：消息体超限（4MB）、Topic 不存在、消息格式错误等（重试也没用）
故障规避机制（故障延迟机制）

这是 RocketMQ 一个非常巧妙的设计。它的核心思想是：某个 Broker 如果发送失败了，在接下来的一段时间内，尽量不要再往这个 Broker 上发消息。

Broker B 故障后

不发送

Producer

Broker B ❌

Broker A ✅

Broker C ✅

在故障延迟时间内
Broker B 被加入黑名单
自动规避

正常情况

Producer

Broker A

Broker B

Broker C

具体工作方式：

每次发送失败时，记录当前系统时间和失败的 Broker
在接下来的 sendLatencyFaultEnable 时间内（默认 30 秒），该 Broker 会被加入黑名单
Producer 选择 Queue 时，会过滤掉黑名单中的 Broker
30 秒后自动恢复，重新尝试往该 Broker 发送
这个机制可以避免 Producer 持续向故障 Broker 发送消息，从而减少不必要的超时等待，提升整体的发送成功率。

消息的路由队列选择算法
消息从 Producer 到 Broker 的过程，实际上经历了两层选择：

需要发送消息

1. 根据 Topic 获取本地缓存的路由信息

2. 从路由信息中获取所有可用的
   MessageQueue 列表

3. 过滤掉
   处于故障延迟期的 Broker 上的 Queue

4. 根据负载均衡策略
   选择一个 MessageQueue

5. 根据选中的 Queue 获取
   对应的 Broker 地址

6. 构造消息发送请求
   通过 Netty 发送到目标 Broker

等待响应

完整的选择决策流程包含以下因素：

路由信息：从本地缓存中获取 Topic 对应的所有 Broker 和 Queue 信息
Broker 可用性：过滤掉 NameServer 已剔除的、或故障延迟期内的 Broker
负载均衡策略：轮询或一致性哈希
Queue 状态：如果某个 Queue 满了，可能被临时剔除（虽然 RocketMQ 很少出现这种情况，因为文件是滚动的）
消息钩子（MessageHook）的使用
MessageHook 是什么？

MessageHook 是 RocketMQ 提供的一个扩展点，允许开发者在消息发送的前后插入自定义逻辑。

典型用途

消息发送钩子

发送前
executeBefore

消息发送

发送后
executeAfter

📊 链路追踪
埋点记录耗时

🔐 权限校验
检查是否有发送权限

📈 流量染色
记录调用方信息

📝 审计日志
记录消息发送记录

使用方式：

// 实现 MessageHook 接口
public class CustomSendHook implements MessageHook {
@Override
public String hookName() {
return “CustomSendHook”;
}

@Override
public void executeBefore(SendMessageContext context) {
    // 发送前：记录开始时间、校验权限等
    System.out.println("开始发送，Topic: " + context.getMessage().getTopic());
}

@Override
public void executeAfter(SendMessageContext context) {
    // 发送后：记录耗时、审计日志等
    System.out.println("发送结束，耗时: " + context.getCostTime());
}

}

// 注册钩子
producer.getDefaultMQProducerImpl().registerMessageHook(new CustomSendHook());
典型应用场景：

全链路追踪：在发送前生成 TraceId，发送后记录日志
性能监控：统计消息发送的耗时分布
流量控制：在发送前进行限流或权限校验
消息发送的超时处理与异常场景
消息发送过程中可能遇到各种异常，我们来看 RcoketMQ 的异常处理是如何设计的：

超时处理

同步发送有一个超时时间（默认 3 秒）。如果在这个时间内没收到 Broker 的响应，Producer 会抛出 RemotingTimeoutException。

发送请求

启动超时计时器

未超时

收到响应

3秒到了

Producer

Broker

Timer 3s

等待响应

✅ 返回成功

❌ 抛出超时异常

⚠️ 注意：超时异常发生时，你无法确定 Broker 是否收到了消息。可能消息根本没到 Broker，也可能 Broker 已经写入了只是网络响应慢了。所以，遇到超时异常时，业务方需要根据消息 Key 做幂等处理，避免重复消费。

常见异常场景

异常类型 可能原因 是否重试 处理建议
RemotingTimeoutException 网络慢或 Broker 响应慢 ✅ 是 适当增大超时时间，或考虑异步发送
RemotingConnectException Broker 连接不上 ✅ 是 检查 Broker 是否存活，故障规避机制会处理
MQClientException Topic 不存在或消息体超限 ❌ 否 检查 Topic 配置和消息大小
MQBrokerException Broker 返回业务错误 看情况 根据错误码判断是否可重试
RemotingSendRequestException 网络发送失败 ✅ 是 检查网络连接
消息发送的请求响应流程（Netty 通信）
RocketMQ 的底层网络通信基于 Netty 实现。一条消息从 Producer 到 Broker，经过的请求响应流程是标准化的：

存储层
Broker 业务处理器
Netty Server
(Broker 端口 10911)
Netty Client
Producer
存储层
Broker 业务处理器
Netty Server
(Broker 端口 10911)
Netty Client
Producer
Broker 接收请求

1. 调用发送 API
2. 通过 Netty Channel 发送
   请求码: SEND_MESSAGE
3. 根据请求码路由到
   SendMessageProcessor
4. 解析请求头
   （Topic、Queue、消息体等）
5. 调用存储层写入 CommitLog
6. 返回写入结果和偏移量
7. 构造响应体
8. 通过 Netty 返回响应
9. 响应返回客户端
10. 回调/唤醒等待线程
11. 处理发送结果
    关键设计细节：

请求码路由：Netty 收到请求后，根据请求码（SEND_MESSAGE、PULL_MESSAGE 等）将请求分发到不同的处理器
单线程模型：RocketMQ 的 Broker 使用单线程处理消息写入（通过 putMessageLock），保证 CommitLog 的顺序写入
异步化：Broker 处理完写入后，通过 Netty 的异步回调机制返回响应，不阻塞 Netty 的 I/O 线程
超时控制：客户端的超时是基于 Netty 的 ResponseFuture 机制实现的
批量消息发送的实现与限制
批量消息发送允许将多条消息打包成一条网络请求发送，从而显著提升吞吐量。

批量发送

1 次请求

消息1

批量消息
消息1+消息2+消息3

消息2

消息3

Broker

1 次网络请求

单条发送

请求1

请求2

请求3

消息1

Broker

消息2

消息3

3 次网络请求

批量发送的使用限制：

同一 Topic：批量消息必须属于同一个 Topic
同一 Broker：批量消息的 Queue 列表必须落在同一个 Broker 上（因为一次网络请求只能发给一个 Broker）
大小限制：批量消息的总大小不能超过 maxMessageSize（默认 4MB）
不能有延迟/事务：批量消息不支持延迟消息和事务消息
使用示例：

List messages = new ArrayList<>();
messages.add(new Message(“topic_test”, “消息1”.getBytes()));
messages.add(new Message(“topic_test”, “消息2”.getBytes()));
messages.add(new Message(“topic_test”, “消息3”.getBytes()));

SendResult sendResult = producer.send(messages); // 批量发送
延迟消息的实现原理（延迟等级）
RocketMQ 的延迟消息非常有意思——它不是精确到秒的任意延迟，而是基于预定义的延迟等级来实现的。

延迟等级

RocketMQ 内置了 18 个延迟等级（messageDelayLevel），每个等级对应一个固定的延迟时间：

等级 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18
延迟 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
底层实现原理：

延迟消息处理

存储层

发送延迟消息

设置 delayTimeLevel=3

1. 计算投递时间

2. 消息写入 CommitLog

3. 异步构建索引

4. 写入 SCHEDULE_TOPIC_XXXX

5. 定时任务扫描

6. 到达投递时间

7. 消息被 Consumer 拉取

Producer

Broker

投递时间 = 当前时间 + 10s

CommitLog

ConsumeQueue

延迟 Topic 的 Queue
（按延迟等级分队列）

后台定时任务
每秒扫描一次

重新投递到目标 Topic

Consumer 消费

实现逻辑（原理层面）：

Producer 发送消息时，设置 delayTimeLevel 参数（比如 3 表示延迟 10 秒）
Broker 收到消息后，不直接把消息写入目标 Topic 的 ConsumeQueue，而是先写入一个内部的 SCHEDULE_TOPIC_XXXX（这个 Topic 有 18 个 Queue，对应 18 个延迟等级）
Broker 有一个后台定时任务（ScheduleMessageService），每秒扫描一次，检查哪些延迟消息到了投递时间
到时间的消息被重新写入目标 Topic 的 CommitLog 和 ConsumeQueue
Consumer 正常消费——对消费者来说，这个消息就像是刚刚发送的一样
💡 小贴士：这种设计的核心是——延迟消息没有特殊的存储格式，就是普通消息加上一个“定时投递”的机制。所有的延迟逻辑都在 Broker 内部完成，Consumer 完全无感知。所以如果你需要更灵活的定时精度，也可以基于这个机制自行扩展。

5.x 版本的变化：

RocketMQ 5.x 对延迟消息做了增强，支持 任意时间点的定时消息（不再是只有 18 个等级），通过 Timing Wheel 算法实现了更灵活的定时投递。但核心原理和上面的流程是一样的，只是内部的定时机制从“固定等级扫描”升级到了“任意时间点调度”。

消息轨迹（Message Trace）的开启与使用
消息轨迹是什么？

消息轨迹记录了一条消息从生产到消费的完整链路信息。你可以通过消息轨迹，清楚地知道：

这条消息是谁发的、什么时候发的、发到了哪个 Broker
这条消息被谁消费了、什么时候消费的、消费结果如何
轨迹数据

消息轨迹

Producer 发送

Broker 存储

Consumer 消费

发送时间

发送方 IP

消息 Key

存储位置

消费时间

消费方 IP

消费状态

开启方式：

方式一：通过配置文件开启（推荐）

在 broker.conf 中配置：

traceTopicEnable=true
msgTraceTopicName=RMQ_SYS_TRACE_TOPIC
方式二：通过客户端显式开启

// Producer 端开启消息轨迹
DefaultMQProducer producer = new DefaultMQProducer(“producer_group”);
producer.setEnableMsgTrace(true);
producer.setMsgTraceTopic(“RMQ_SYS_TRACE_TOPIC”);
消息轨迹的实现原理

消息发送/消费

通过 MessageHook 和 ConsumeHook
拦截关键事件

构建轨迹数据
（发送时间、IP、状态等）

将轨迹数据作为
普通消息发送到 Trace Topic

Broker 存储轨迹消息

用户通过控制台或 API
查询轨迹信息

核心设计：

无侵入：通过 Hook 机制实现，业务代码不需要改造
异步上报：轨迹数据的发送是异步的，不影响主消息发送的性能
独立存储：轨迹数据存储在独立的 RMQ_SYS_TRACE_TOPIC 中，与业务数据隔离
可查询：通过 RocketMQ 的 Dashboard 或 API 可以按消息 ID/Key 查询轨迹
实际使用中的注意事项：

开启消息轨迹会增加少量性能开销（需要额外发送轨迹消息），生产环境根据实际需求开启
轨迹消息本身也占用存储空间，需要适当设置 fileReservedTime
对于超高吞吐量的场景，可以采样记录轨迹，而不是每条都记录
小结
这篇文章我们完整走通了消息从生产到发送的全链路，通过 8 张流程图，搞清楚了：

Producer 启动时如何初始化并获取路由信息
三种发送方式（同步、异步、单向）的底层实现和适用场景
负载均衡策略（轮询、一致性哈希）如何选择目标 Queue
重试机制和故障规避机制如何在发送失败时保证高可用