小编点评

**RocketMQ 消息发送流程** **业务层** - 接收业务层消息。 - 通过 `DefaultMQProducerImpl.send()` 发送 RPC 请求。 **通信层** - 监听 Broker 的连接变化。 - 当生产者状态为正常时，调用 `defaultMQProducerImpl.sendDefaultImpl()` 方法。 **消息处理层** - 检查消息是否有效。 - 如果有效，执行消息发送。 - 设置发送超时时间。 - 调用 `defaultMQProducerImpl.sendDefaultImpl()` 方法。 **发送消息的核心步骤** 1. **检查生产者状态和消息内容** - 确保生产者状态正常。 - 检查消息主题名称是否合法。 - 检查消息体大小是否合规（最大 4MB）。 2. **获取 Topic 路由信息** - 从本地缓存中获取 Topic 路由信息。 - 如果缓存中没有信息，则从 NameServer 获取路由信息并更新本地缓存。 3. **设置重试次数** - 计算消息发送的重试次数。 - 设置同步或异步重试的执行方式。 4. **选择发送队列** - 通过 `selectOneMessageQueue()` 方法选择一个队列，轮询发送消息到 Broker。 5. **执行发送请求** - 调用 `sendKernelImpl()` 方法，准备通信层请求。 - 设置 RequestCode 表示发送单个消息。

正文

RocketMQ客户端的消息发送通常分为以下3层

• 业务层：通常指直接调用RocketMQ Client发送API的业务代码。
• 消息处理层：指RocketMQ Client获取业务发送的消息对象后，一系列的参数检查、消息发送准备、参数包装等操作。
• 通信层：指RocketMQ基于Netty封装的一个RPC通信服务，RocketMQ的各个组件之间的通信全部使用该通信层。

消息发送流程首先是RocketMQ客户端接收业务层消息，然后通过DefaultMQProducerImpl发送一个RPC请求给Broker，再由Broker处理请求并保存消息。

下面以DefaultMQProducer.send（Messagemsg）接口为例讲解发送流程，

• 第一步：调用defaultMQProducerImpl.send（）方法发送消息。
• 第二步：通过设置的发送超时时间，调用defaultMQProducerImpl.send（）方法发送消息。设置的超时时间可以通过sendMsgTimeout进行变更，其默认值为3s。
• 第三步：执行 defaultMQProducerImpl.sendDefaultImpl()方法。

private SendResult sendDefaultImpl(
    Message msg,
    //通信模式，同步、异步还是单向
    final CommunicationMode communicationMode,
    //对于异步模式，需要设置发送完成后的回调
    final SendCallback sendCallback,
    final long timeout
){}

该方法是发送消息的核心方法，执行过程分为5步：
第一步： 两个检查：生产者状态、消息及消息内容。没有运行的生产者不能发送消息。消息检查主要检查消息是否为空，消息的Topic的名字是否为空或者是否符合规范；消息体大小是否符合要求，最大值为4MB，可以通过maxMessageSize进行设置。
第二步： 执行tryToFindTopicPublishInfo（）方法：获取Topic路由信息，如果不存在则发出异常提醒用户。如果本地缓存没有路由信息，就通过Namesrv获取路由信息，更新到本地，再返回。具体实现代码如下：

private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
    TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
    if (null == topicPublishInfo || !topicPublishInfo.ok()) {
        this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
        this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
        topicPublishInfo = this.topicPublishInfoTable.get(topic);
    }

    if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
        return topicPublishInfo;
    } else {
        this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
        topicPublishInfo = this.topicPublishInfoTable.get(topic);
        return topicPublishInfo;
    }
}

第三步： 计算消息发送的重试次数，同步重试和异步重试的执行方式是不同的。
第四步： 执行队列选择方法selectOneMessageQueue（）。根据队列对象中保存的上次发送消息的Broker的名字和Topic路由，选择（轮询）一个Queue将消息发送到 Broker 。 我们可以通过sendLatencyFaultEnable 来设置是否总是发送到延迟级别较低的 Broker，默认值为False。
第五步： 执行sendKernelImpl（）方法。该方法是发送消息的核心方法，主要用于准备通信层的入参（比如Broker地址、请求体等），将请求传递给通信层，内部实现是基于Netty的，在封装为通信层request对象RemotingCommand前，会设置RequestCode表示当前请求是发送单个消息还是批量消息。