RocketMQ - 生产者消息发送流程

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表示当前请求是发送单个消息还是批量消息。