RocketMQ Producer发送消息过程

Posted 乐观男孩

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RocketMQ Producer发送消息过程相关的知识,希望对你有一定的参考价值。

目录

说明

在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的同步调用,发送同步消息

以上是关于RocketMQ Producer发送消息过程的主要内容,如果未能解决你的问题,请参考以下文章

RocketMQ 三种消息发送方式

rocketmq源码心得

RocketMQ源码—Producer发送消息源码—发送消息的总体流程一万字

RocketMq producer 发送一条消息所经过的流程

RocketMQ源码—Producer发送消息源码—单向同步异步发送消息一万字

RocketMQ源码—Producer发送消息的总体流程一万字