RocketMQ 源码分析——Producer

刺骨的言语ヽ痛彻心扉 2023-10-16 10:34 89阅读 0赞

#### 文章目录 ####

*  消息发送代码实现
 *  消息发送者启动流程
 *   *  检查配置
     *  获得MQ客户端实例
     *  启动实例
     *  定时任务
 *  Producer 消息发送流程
 *   *  选择队列
     *   *  默认选择队列策略
         *  故障延迟机制策略\*
         *  两种策略的选择
 *  技术亮点:ThreadLocal

## 消息发送代码实现 ##

下面是一个生产者发送消息的demo（同步发送）

![image.png][]

**主要做了几件事：**

*  初始化一个生产者（DefaultMQProducer）对象
 *  设置 NameServer 的地址
 *  启动生产者
 *  发送消息

## 消息发送者启动流程 ##

![image.png][image.png 1]

**DefaultMQProducerImpl**类start()

![image.png][image.png 2]

### 检查配置 ###

**DefaultMQProducerImpl**类

![image.png][image.png 3]

### 获得MQ客户端实例 ###

整个JVM中只存在一个MQClientManager实例，维护一个MQClientInstance缓存表。DefaultMQProducerImpl类start()

![image.png][image.png 4]

一个clientId只会创建一个MQClientInstance

![image.png][image.png 5]

clientId生成规则：IP@instanceName@unitName

![image.png][image.png 6]

RocketMQ中消息发送者、消息消费者都属于”客户端“。每一个客户端就是一个MQClientInstance，每一个ClientConfig对应一个实例。

不同的生产者、消费端如果引用同一个客户端配置(ClientConfig)，则它们共享一个MQClientInstance实例。所以我们在定义的的时候要注意这种问题（生产者和消费者如果分组名相同容易导致这个问题）

![image.png][image.png 7]

### 启动实例 ###

MQClientInstance类start()

![image.png][image.png 8]

### 定时任务 ###

MQClientInstance类startScheduledTask()

![image.png][image.png 9]

## Producer 消息发送流程 ##

我们从一个生产者案例的代码进入代码可知：DefaultMQProducerImpl中的sendDefaultImpl()是生产者消息发送的核心方法

![image.png][image.png 10]

![image.png][image.png 11]

从核心方法可知消息发送就是4个步骤：验证消息、查找路由、选择队列、消息发送。

![image.png][image.png 12]

![image.png][image.png 13]

### 选择队列 ###

#### 默认选择队列策略 ####

采用了最简单的轮询算法，这种算法有个很好的特性就是，保证每一个Queue队列的消息投递数量尽可能均匀。这种算法只要消息投递过程中没有发生重试的话，基本上可以保证每一个Queue队列的消息投递数量尽可能均匀。当然如果投递中发生问题，比如第一次投递就失败，那么很大的可能性是集群状态下的一台Broker挂了，所以在重试发送中进行规避。这样设置也是比较合理的。

#### 故障延迟机制策略\* ####

采用此策略后，每次向Broker成功或者异常的发送，RocketMQ都会计算出该Borker的可用时间（发送结束时间-发送开始时间，失败的按照30S计算），并且保存，方便下次发送时做筛选。

![image.png][image.png 14]

除了记录Broker的发送消息时长之外，还要计算一个Broker的不可用时长。这里采用一个经验值：

如果发送时长在550ms之内，不可用时长为0。

达到550ms，不可用时长为30S

达到1000ms，不可用时长为60S

达到2000ms，不可用时长为120S

达到3000ms，不可用时长为180S

达到15000ms，不可用时长为600S

![image.png][image.png 15]

![image.png][image.png 16]

有了以上的Broker规避信息后发送消息就非常简单了。

在开启故障延迟机制策略步骤如下：

1.  根据消息队列表时做轮训
2.  选好一个队列
3.  判断该队列所在Broker是否可用
4.  如果是可用则返回该队列，队列选择逻辑结束
5.  如果不可用，则接着步骤2继续
6.  如果都不可用，则随机选一个

**代码如下：**

![image.png][image.png 17]

#### 两种策略的选择 ####

从这种策略上可以很明显看到，默认队列选择是轮训策略，而故障延迟选择队列则是优先考虑消息的发送时长短的队列。那么如何选择呢？

首先RocketMQ默认的发送失败有重试策略，默认是2，也就是如果向不同的Broker发送三次都失败了那么这条消息的发送就失败了，作为RocketMQ肯定是尽力要确保消息发送成功。所以给出以下建议。

如果是网络比较好的环境，推荐默认策略，毕竟网络问题导致的发送失败几率比较小。

如果是网络不太好的环境，推荐故障延迟机制，消息队列选择时，会在一段时间内过滤掉RocketMQ认为不可用的broker，以此来避免不断向宕机的broker发送消息，从而实现消息发送高可用。

当然以上成立的条件是一个Topic创建在2个Broker以上的的基础上。

## 技术亮点:ThreadLocal ##

![image.png][image.png 18]

![image.png][image.png 19]

![image.png][image.png 20]

[image.png]: https://img-blog.csdnimg.cn/img_convert/f7664fd08e106ec37f4bf892e89c7282.png
[image.png 1]: https://img-blog.csdnimg.cn/img_convert/90454d8a6b3772e74026ea6c274f08f1.png
[image.png 2]: https://img-blog.csdnimg.cn/img_convert/90216b594d35807b1ecece58db9f1932.png
[image.png 3]: https://img-blog.csdnimg.cn/img_convert/bdcbc5e36b38d643af8065f1ab338ee8.png
[image.png 4]: https://img-blog.csdnimg.cn/img_convert/1d6202ac0d9dbe27b07d9df025905b57.png
[image.png 5]: https://img-blog.csdnimg.cn/img_convert/610b88875e15ca79c485114651c9b3a6.png
[image.png 6]: https://img-blog.csdnimg.cn/img_convert/e270e275563ef1605f657822168feec9.png
[image.png 7]: https://img-blog.csdnimg.cn/img_convert/a0e3d1261665f39ded772955c361d647.png
[image.png 8]: https://img-blog.csdnimg.cn/img_convert/8d3b34e7e8c9764d790206d34cf2c32a.png
[image.png 9]: https://img-blog.csdnimg.cn/img_convert/34c09a76e0188eb24c9dc1ea5c7947ed.png
[image.png 10]: https://img-blog.csdnimg.cn/img_convert/a7f82c7ab21810550f74421725e222f5.png
[image.png 11]: https://img-blog.csdnimg.cn/img_convert/af0dc7ac79a4a76433a0422eca5ca77e.png
[image.png 12]: https://img-blog.csdnimg.cn/img_convert/4aecb2a166ffea8bf75c988d2ca0c7dc.png
[image.png 13]: https://img-blog.csdnimg.cn/img_convert/b923ea12b6d591fa0050356c6b268c1e.png
[image.png 14]: https://img-blog.csdnimg.cn/img_convert/d19889b6a5dda81d71239c65fcaa495b.png
[image.png 15]: https://img-blog.csdnimg.cn/img_convert/9fa4b864629058da27ad271286c00569.png
[image.png 16]: https://img-blog.csdnimg.cn/img_convert/6da10687236d51cff2fd25c4b6762332.png
[image.png 17]: https://img-blog.csdnimg.cn/img_convert/659357212cc69963e96e353be4bba0f2.png
[image.png 18]: https://img-blog.csdnimg.cn/img_convert/cc9b5cfea1db2a6e0aa5389451b0f98a.png
[image.png 19]: https://img-blog.csdnimg.cn/img_convert/061052581cb5559d40c1bebbfcc69d92.png
[image.png 20]: https://img-blog.csdnimg.cn/img_convert/556315d85f2cf04ac72375459fd6c5b5.png