RocketMQ Producer发送消息过程

Posted 2022-12-02 乐观男孩

目录

• 说明
• 消息发送入口
• 根据Topic获取路由信息
• 选择消息发送的目标队列
• 发送消息到Broker

说明

在Producer启动完成后，就可以利用Producer向Broker发送消息，我们以最简单的发送同步消息进行流程讲解。发送消息的核心逻辑分三个步骤：
1、根据Topic获取路由信息；
2、选择消息发送的目标队列；
3、发送消息到Broker。

消息发送入口

Producer发送消息的入口在DefaultMQProducer类的send()方法，DefaultMQProducer提供了很多重载的方法，这些重载方法，内部调用了DefaultMQProducerImpl内的send方法，发送的核心逻辑就在该类的sendDefaultImpl()方法内





下面，我们分步骤看一下每个主要步骤的逻辑。

根据Topic获取路由信息

DefaultMQProducerImpl的tryToFindTopicPublishInfo方法，将获取到的Topic路由信息（TopicRouteData）封装成TopicPublishInfo。这里先会根据指定的topic获取路由信息，如果获取不到，则会使用默认的topic（TBW102，DefaultMQProducer类createTopicKey属性定义）再获取一次

接着我们进入MQClientInstance.updateTopicRouteInfoFromNameServer()方法：该方法是通过MQClientAPIImpl.getTopicRouteInfoFromNameServer()方法获取路由信息。

进入MQClientAPIImpl.getTopicRouteInfoFromNameServer()方法：该方法其实很简单，构造请求命令->发送请求指令->解析结果。

获取到的路由信息，包括队列数据、Broker数据和Broker地址信息：

在获取到topic的路由信息后，我们再回到MQClientInstance.updateTopicRouteInfoFromNameServer()方法：判断MQClientInstance.topicRouteTable属性以前缓存的路由信息是否需要更新、DefaultMQProducerImpl.topicPublishInfoTable属性以前缓存的路由信息是否需要更新

如果需要更新：

更新完之后，DefaultMQProducerImpl.tryToFindTopicPublishInfo()方法就可以从缓存（Map）中获取Topic的路由信息TopicRouteInfo了。TopicRouteInfo结构

所以，整个流程如下（实线为请求过程，虚线为响应过程）：

选择消息发送的目标队列

在获取到topic的路由信息后，将选择消息发送的目标队列。DefaultMQProducerImpl.selectOneMessageQueue()方法是通过MQFaultStrategy.selectOneMessageQueue()方法实现的。这里分两种情况讨论：
1、如果开启延迟容错策略（producer.setSendLatencyFaultEnable(true)），则要考虑Broker的响应时间。
这里解释一下延迟容错策略，它指的是要考虑Broker的响应时间。如果Broker响应缓慢，则我们选择队列时，就不应该选择该Broker，而应该选择响应更快的Broker里的队列。Broker的响应时间是在消息发送阶段采集的：

如果Broker响应时间达到一定的标准，则一段时间内认为该Broker尽量不要选用，这个标准定义在MQFaultStrategy（没错，源码是直接写死的）

明白了延迟容错策略后，我们再来看看开启了延迟容错策略后如何选择消息发送的队列

步骤：
1、遍历topic路由信息里的队列，看队列对应的Broker是否健康，如果健康，则直接返回该队列；
2、如果遍历了所有队列后，依然没有找到比较健康的Broker，则择优取一个，择优规则（择优规则通过FaultItem的排序实现）按以下规则进行：
（1）、哪个Broker健康，哪个优先被选择。
（2）、哪个Broker响应较快，哪个优先被选择。
（3）、哪个Broker剩余不可用时间越小，哪个优先被选择。

2、如果没有开启延迟容错策略，则直接从Topic路由信息获取

第一次随机，后续往后选另一个Broker的队列

发送消息到Broker

队列选好后，就是发送消息了。发送消息的核心逻辑在DefaultMQProducerImpl.sendKernelImpl()方法


最终通过NettyRemotingClient的同步调用，发送同步消息