RocketMQ源码学习笔记：Producer发送消息流程

这是本人学习的总结，主要学习资料如下

• 马士兵教育
• rocketMq官方文档

---

1、Overview

消息发送主要可以分成下面四个步骤。

1. 验证消息
2. 查找路由
3. 选择队列
4. 消息发送

之后从源码查看四个步骤的具体内容。

我们建立一个DefaultMQProducer之后，调用DefaultMQProducer#send()方法就可发送信息。

查看send()的代码最终会来到DefaultMQProducerImpl#sendDefaultImpl()，我们从这里开始看源码。

2、验证消息

发送前必然验证一下消息。

主要是检验消息的状态，一些必要的值不能为空等。

this.makeSureStateOK();
// 1、检查消息
Validators.checkMessage(msg, this.defaultMQProducer);

公司内部想设置一些新的规则用来发送前拦截信息就适合放在checkMessage()里。

这部分没有太多内容。

3、查找路由

所谓的路由是指可用的Broker的信息，包括地址，具体的消息队列等。

下面这一句获取到路由。

TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());

在DefaultMQProducer内部缓存这路由信息，维护在ConcurrentHashMap<String, TopicPublishInfo>中

private final ConcurrentMap<String/* topic */, TopicPublishInfo> topicPublishInfoTable = new ConcurrentHashMap<String, TopicPublishInfo>();

在tryToFindTopicPublishInfo()中会先检查路由信息是否存在，不存在还需要从NameServer中获取路由列表。

this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);

更新的时候会加上一个ReentrantLock，更新结束后释放。

获取到路由信息后开始选择队列发送消息。

4、选择消息发送队列

4.1、选择队列的策略

得到路由信息后就开始选择消息队列发送信息。

选择队列有两种策略

• 轮询规避：轮询选择队列。如果上次发送消息失败，那就消息需要重新发送，这时就需要规避掉上次发送失败的队列，寻找下一个队列发送。
• 故障延迟策略：在选择队列发送时根据以往发送时长判断该队列的Broker是否可用。对于发送失败的Broker，Producer会规避该Broker一段时间。

这是发送消息的流程图。

假设我们的Broker是集群，有两个Broker。消息会选择其中一个Broker发送消息，如果失败就重试，直到发送成功或者超过重试次数。

4.2、源码阅读

这里会探索源码如何实现这两种队列选择策略。

选择队列的入口在DefaultMQProducerImpl#sendDefaultImpl -> this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);

从入口进入查看代码，最终在MQFaultStrategy#selectOneMessageQueue，代码通过sendLatencyFaultEnable这个字段来选择不同的选择策略。

4.2.1、轮询规避

这是轮询规避的源码。

其中lastBrokerName是上一次消息发送时选择的broker。这代表该消息上一次发送失败了，所以记录着上一次失败的broker以在这次选择Broker时规避他。

所以lastBrokerName==null时该消息是第一次发送，不需要规避，直接随机选择一个队列发送。

如果上一次发送失败，则开始轮询选择一个队列，保证这个新选出的队列和上一个不同后就可以返回。

4.2.2、故障延迟规避

4.2.2.1、计算规避时间

故障延迟规避策略需要记录发送时间并计算。在看选择Broker的代码时需要看看源码如何记录发送时间并计算出规避时间的。

计算规避时间的代码在this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);。这时消息正常发送会调用一次这个方法；如果出现异常在catch快也会调用这个方法计算规避时间。

进入这个方法，代码如下。

需要注意isolation=true时表示消息发送出现异常，这时便认为延迟时长是30000ms。

同时也可以看到sendLatencyFaultEnable==true表示开启故障规避策略，这种情况才需要计算规避时间。选择Broker时也是通过这个属性判断使用过故障规避还是轮询规避。

规避时间的计算比较简单，阿里根据自己的经验设置了一个对照表来计算时间，如下图所示。比如延迟是550ms以内的Broker不用规避；延迟在550~1000ms的需要规避30s。

这里跳过计算规避时间的代码细节，进入下一行代码this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);。

代码如下，它其实就是将不可用的Broker维护在faultItemTable中，并且记录着解禁时间。以后选择Broker会通过这个集合查看Broker是否可用。

4.2.2.2、选择队列

我们回到选择Broker的代码MQFaultStrategy#selectOneMessageQueue，下图是相关代码。

它的大概流程是，轮询队列，如果可用就返回。实在找不到可用的就随机选择一个Broker发送。

它通过latencyFaultTolerance.isAvailable(mq.getBrokerName())判断队列是否可用，里面实际就是通过前面讲到的faultItemTable来查看队列是否可用。

4.2.3、ThreadLocal的使用

在选择队列时，无论是轮询规避还是故障延迟规避都需要循环遍历messageQueue找到适合的queue发送信息。

获取下标的方式用到了ThreadLocal。如下图所示，sendWhichQueue本质上就是一个ThreadLocal<Integer>对象。

生产者发送信息时可能会有多个线程同时发信息。

这些线程发送信息时应该各自维护一个消息队列的下标，这样每个线程发送信息时才会比较均匀地向每个队列都发送信息。

另外这些线程发送信息时可能会指定消息队列的id，所以线程各自维护一个消息队列的下标是很有必要的。

这个场景就很适合ThreadLocal，选择消息队列时用ThreadLocal来维护下标。

5、发送消息

5.1、客户端建立的时间

客户端发送消息时，建立HTTP连接是在send()方法中而不是在start()方法中。

站在设计者的角度需要考虑到，开发者在start()方法后可能还需要过一段时间才会真正发送信息，甚至不发信息。

那么建立HTTP连接放在start()就比较浪费资源，所以建立HTTP连接放在了send()方法中。