从零开始读RocketMq源码(一)生产者启动

前言

获取源码

总概论

生产者实例

源码

A-01:设置生产者组名称

A-02:生产者服务启动

B-01：初始化状态

B-02：该方法再次对生产者组名称进行校验

B-03：判断是否为默认生产者组名称

B-04: 该方法是为了实例化MQClientInstance对象，mq客户端对象实例

B-05: 该方法就是将当前生产者对象注册到mqClientInstance中的producerTable集合中，并且生产者组名称作为key

B-06: 启动相关核心服务以及开启一系列定时任务（核心逻辑）

1. 开启请求-响应通道- this.mQClientAPIImpl.start();

2. 开启拉动式服务- this.mQClientAPIImpl.start();

3. 开启负载均衡服务- this.rebalanceService.start();

4. 开启推送服务- this.defaultMQProducer.getDefaultMQProducerImpl().start(false);

5. 启动各种计划任务- this.startScheduledTask();

a. 启动定时任务获取MQ注册中心nameServer的地址- MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();

b. 定时从nameServer拉取topic信息到本地存储 - MQClientInstance.this.updateTopicRouteInfoFromNameServer();

c. 定时清除离线的broker服务并给所有在线的broker发送心跳

d. 定时持久化消费偏移量数据

e. 定时调整消费者消息的线程池数量

B-07:初始化topic路由信息、topic訂閲信息以及topic端点映射信息

B-08:开启定时监测broker故障信息任务

B-09:发送心跳给所有的broker服务

B-10:开启定时扫描异步请求响应任务

A-03:开启监控和处理同步发送和异步发送操作的守护线程

A-04:开启消息轨迹和发送机制

总结

展望

前言

大概一年半前自己写了一篇《云原生》一文搞懂RocketMQ队列概述，这篇对rocketmq的相关概念和使用方法进行了整理概述，就像结尾说的都太局限于表面，简单使用还能将就，但一出现问题自己也很难排查，为了迈向技能的下一个阶段，还得是要读源码，学习大佬们的编码风格和技巧，对于使用mq以及排除问题也会更得心应手。因为最近一年自己的工作充满了波折，让自己没法静下心来学习整理，虽然现在也好不了多少，但可算能回归本心。本章篇幅比较长，将近万字，博主也是自己读源码一步一步跟踪的，所以尽量想描述得通俗易懂一些。

获取源码

首先我们从github上拉取rocketmqd的源码链接到本地，使用idea打开。

源码地址：https://github.com/apache/rocketmq

目前最新版本为：5.2.0

那么我们在idea上切换分支为 release-5.2.0

注：rocketmq5.x与4.x官方改动的东西比较多，尽量使用一直的版本，具体差异可查看官网，这里只对源码逻辑进行分析

总概论

我们知道rocketmq的组成需要四大模块构成，缺一不可

nameserver mq注册中心(状态管理)
broker mq的服务端（核心）
producer 生产者
consumer 消费者

本章我们先从应该大家接触最多的生产者开始学习源码吧。

生产者实例

在idea的rocketmq源码中找到 example 模块，这个模块中都是官方给出的简单案例
然后找到simple 包下面的 Producer类打开
然后在producer类中配置自己的mq的地址，topic以及tag就能成功启动生产者并且发送消息

注意：这里成功启动的前提是必须提前启动了mq的nameserver服务和broker服务才能成功，若没有可不用启动，直接跳过看下面源码

源码

根据上面简单生产者实例可知，生产者端的两大核心就是，启动生产者和发送消息，分别对应下面两行代码。看似简单的两行其实里面的功能逻辑很强大。

```
producer.start();
```
```
producer.send(msg);
```

生产者包含4中状态：

CREATE_JUST 服务刚刚创建，尚未启动

```
RUNNING 服务运行中
```
```
SHUTDOWN_ALREADY 服务已关闭
```
```
START_FAILED 启动出错
```

按照顺序，我们从 生产者的启动开始

public void start() throws MQClientException {//A-01:设置生产者组名称this.setProducerGroup(withNamespace(this.producerGroup));//A-02:生产者服务启动this.defaultMQProducerImpl.start();//A-03:开启监控和处理同步发送和异步发送操作的守护线程if (this.produceAccumulator != null) {this.produceAccumulator.start();}//A-04:开启消息轨迹和发送机制if (null != traceDispatcher) {try {traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());} catch (MQClientException e) {logger.warn("trace dispatcher start failed ", e);}}
}

A-01:设置生产者组名称

该方法中顾名思义主要用于设置生产者组的名称
withNamespace（）进入该方法发现，其实对生产者组的名称就行各种非空校验和长度校验，最后根据固定格式拼接名称后返回。（对于开源组件大佬，校验方式也是和我们无异的）

A-02:生产者服务启动

该方法为本次的启动核心方法，我们直接深入了解下其内部实现。

方法逻辑太长我们进行分段拆分来解析

public void start(final boolean startFactory) throws MQClientException {switch (this.serviceState) {//B-01：初始化状态case CREATE_JUST:this.serviceState = ServiceState.START_FAILED;//B-02：校验this.checkConfig();//B-03：生产者组名设置if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {this.defaultMQProducer.changeInstanceNameToPID();}//...

B-01：初始化状态

因为现在还是正在启动中，所以状态还是默认未启动状态，那么直接进入第一个case逻辑中，进入后里面把状态至为启动失败，我认为这是一种防御性编码，并且防止未成功启动的生产者被重复启动

B-02：该方法再次对生产者组名称进行校验

B-03：判断是否为默认生产者组名称

前面可知我们已经成功设置自定义名称，所以直接进入if中

changeInstanceNameToPID（），该方法就设置实例名称，进入方法可以看到名称的生成规则，this.instanceName = UtilAll.getPid() + "#" + System.nanoTime();
当前运行的虚拟机的名称截取拼接上当前纳米时间戳，保证唯一性

public void start(final boolean startFactory) throws MQClientException {/......///B-04：该方法是为了实例化MQClientInstance对象，mq客户端对象实例this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);//B-05：注册生产者boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);/....../

B-04: 该方法是为了实例化MQClientInstance对象，mq客户端对象实例

内部首先生成一个唯一的clientId,其组成包含ip地址与之前生成的实例名称instanceName组成，然后new 了一个MQClientInstance对象并设置对应属性。
将clientId作为key维护到一个Map对象中，private final ConcurrentMap<String/* clientId */, MQClientInstance> factoryTable；

注：MQClientInstance对象，该对象非常重要，因为生产者和消费者都在使用

进入该对象我们可以发现，里面维护了两个Map集合,就是分别存储当前客户端的生产者和消费者的对象数据

private final ConcurrentMap<String, MQProducerInner> producerTable ；
private final ConcurrentMap<String, MQConsumerInner> consumerTable ；

B-05: 该方法就是将当前生产者对象注册到mqClientInstance中的producerTable集合中，并且生产者组名称作为key

public void start(final boolean startFactory) throws MQClientException {switch (this.serviceState) {/....///B-06： 启动相关核心服务以及开启一系列定时任务if (startFactory) {mQClientFactory.start();}/.../

B-06: 启动相关核心服务以及开启一系列定时任务（核心逻辑）

1. 开启请求-响应通道- this.mQClientAPIImpl.start();

2. 开启拉动式服务- this.mQClientAPIImpl.start();

3. 开启负载均衡服务- this.rebalanceService.start();

4. 开启推送服务- this.defaultMQProducer.getDefaultMQProducerImpl().start(false);

这个方法是否有点眼熟,没错这就是我们最开始调用的启动方法A-2,参数传的false,说明上面if代码块中startFactory=false，则不进入B-06的代码块中
并且A-2代码块方法中，因为第一次进入时状态已经从CREATE_JUST变更为START_FAILED，所以也不会再次进入第一个case中
阅读后续代码可知，核心就是调用了 this.mQClientFactory.sendHeartbeatToAllBrokerWithLock(); 向所有Broker服务发送一次心跳（具体后面会详解）

5. 启动各种计划任务- this.startScheduledTask();

所有任务都是使用Executors线程池创建一个单独的的单线程定时任务实现,如下格式

private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(r -> new Thread(r, "MQClientFactoryScheduledThread"));
//....
this.scheduledExecutorService.scheduleAtFixedRate(() -> {try {//业务逻辑} catch (Exception e) {log.error("ScheduledTask fetchNameServerAddr exception", e);}
}, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);

a. 启动定时任务获取MQ注册中心nameServer的地址- MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();

首次启动延迟时间：2s

定时间隔时间：2m

mQClientAPIImpl对象是否眼熟，没错就是上面B-06-1启动的服务，所以该服务必须在任务执行之前启动，查看源码如此

深入方法中会发现其实就是获取地址处理后存储在一个List集合中，为什么使用集合，我认为如果是集群那就就会有多条地址存在。 private final AtomicReference<List<String>> namesrvAddrList = new AtomicReference<>();
继续深入会发现有Netty的身影，用于服务间远程通信，这里不再研究。
private final ConcurrentMap<String /* addr */, ChannelWrapper> channelTables;
该Map就是用nameserver地址作为key,而value为ChannelWrapper对象，该对象内部就使用了netty框架包中的对象，一个地址对应一个通道封装器。但是该逻辑中并没有使用put操作，只是get获取。

b. 定时从nameServer拉取topic信息到本地存储 - MQClientInstance.this.updateTopicRouteInfoFromNameServer();

首次启动延迟时间：10ms

定时间隔时间：30s

深入方法内部可知，其实就是分别对producerTable与consumerTable的map进行操作遍历，取出对象里面的topic名称，由前面B-04中可知，分别用于存储生产者对象和消费者对象信息
再将topic名称的set集合进行遍历去远程获取nameserver中的topic的路由详细信息，并将信息存储在另一个map对象中。作用： 用于管理和查询主题的路由信息，帮助生产者和消费者确定消息的发送和接收路径。
private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<>();

c. 定时清除离线的broker服务并给所有在线的broker发送心跳

MQClientInstance.this.cleanOfflineBroker(); 清除离线的broker

MQClientInstance.this.sendHeartbeatToAllBrokerWithLock(); 给所有的broker发送心跳

首次启动延迟时间：1s

定时间隔时间：30s

清除离线的broker，查看源码可知道，大概意思为首先从private final ConcurrentMap<String, HashMap<Long, String>> brokerAddrTable = new ConcurrentHashMap<>(); map中
- 获取所有broker的地址数据，然后进行遍历，
- 在遍历中取出 topicRouteTable,该map存放的是topic的对象信息
- 再对topic map的values进行遍历，取出topic信息对象中存储的对应broker集合，
- 判断上面的brokerAddrTable中的broker是否在topic维护的broker集合中，没有则清除

d. 定时持久化消费偏移量数据

MQClientInstance.this.persistAllConsumerOffset();

首次启动延迟时间：10s

定时间隔时间：5s

同样的维护了一个Map对象：

private ConcurrentMap<MessageQueue, ControllableOffset> offsetTable;

key则为消息队列对象

深入源码可知，它的消费者持久化实现方式有三种
- lite pull
- mp pull
- mp push

e. 定时调整消费者消息的线程池数量

MQClientInstance.this.adjustThreadPool();

首次启动延迟时间：1m

定时间隔时间：1m

public void start(final boolean startFactory) throws MQClientException {switch (this.serviceState) {//...//B-07：初始化topic路由信息、topic訂閲信息以及topic端点映射信息this.initTopicRoute();//B-08：开启定时监测broker故障信息任务this.mqFaultStrategy.startDetector();//...

B-07:初始化topic路由信息、topic訂閲信息以及topic端点映射信息

深入源码可知，首先获取开发者自定义的topic集合，然后分别处理成MQ要求的格式newTopic，然后创建TopicPublishInfo对象，用于存储topic订阅信息newTopic作为key，同样最后放入map中

private final ConcurrentMap<String/* topic */, TopicPublishInfo> topicPublishInfoTable;

查看TopicPublishInfo对象可知，对象里面包含了TopicRouteData对象，我们知道这个对象在上面定时器B-06-5-b中出现过用于存储topic路由信息，并且存储在topicRouteTable map中
所以在本方法中也会通过newTopic去远程从nameserver中拉去TopicRouteData信息，设置到TopicPublishInfo对象中，同样也会对比topic新获取的TopicRouteData与原来定时器存储的topicRouteTable中的是否有变化，有则更新
有变化同时还会更新，上面定时器B-06-5-c中出现的brokerAddrTable map,更新broker的地址信息
同时更新topic端点映射信息-记录每个主题的消息队列与 Broker 之间的映射
private final ConcurrentMap<String/* Topic */, ConcurrentMap<MessageQueue, String/*brokerName*/>> topicEndPointsTable；这是一个嵌套map,因为一个topic可能对应多个broker，那么消息队列也会是对应多个broker, 可以帮助管理和均衡负载，确保消息被分布到不同的 Broker 上。

B-08:开启定时监测broker故障信息任务

深入源码可知，里面维护了一个定时任务，定时监测 Broker 的故障详细信息

首次启动延迟时间：3s

定时间隔时间：3s

同时也维护了一个map,用于存储每一个broker的 故障详细信息，包括故障时间、故障持续时间和可用状态

private final ConcurrentHashMap<String, FaultItem> faultItemTable;

逻辑处理中还会去查询brokerAddrTable中是否还存在当前broker地址信息，不存在则从faultItemTable中移除，然后再去监测broker服务是否可用，若可用则将可用状态 设置为true

public void start(final boolean startFactory) throws MQClientException {//...//B-09：发送心跳给所有的broker服务this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();//B-10：开启定时扫描异步请求响应任务RequestFutureHolder.getInstance().startScheduledTask(this);}

B-09:发送心跳给所有的broker服务

发送心跳其实在上面定时器B-06-5-c中已经出现过了，但是没有深入了解，那么定时器中既然已经在发送心跳了，为什么生产者启动最后还要发送呢？

定时任务的作用：定时任务确保客户端在运行过程中定期发送心跳，保持与 Broker 的连接。
启动时的心跳：生产者在启动完成时立即发送心跳，以确保初始化成功、快速检测连接状态并更新路由信息。

同样的心跳机制中也维护了一个map, 用于记录和管理每个 Broker 的心跳信息，private final ConcurrentMap<String, Integer> brokerAddrHeartbeatFingerprintTable;
其中value值称为心跳指纹， MQ通过比较当前心跳指纹和上次记录的指纹，可以判断 Broker 是否正常工作

B-10:开启定时扫描异步请求响应任务

深入源码可知，里面维护了一个定时任务，定时扫描MQ存储的生产者发布的异步请求以及响应的信息，帮助MQ实现异步请求的超时、回调和状态管理，增强系统的异步处理能力。

次启动延迟时间：3s

定时间隔时间：1s

同样的是维护了一个map数据用于存储异步请求以及响应的信息：

private ConcurrentHashMap<String, RequestResponseFuture> requestFutureTable；

那么key为请求时生成的唯一标识，value为RequestResponseFuture对象则记录了请求信息、超时时间、响应信息、回调信息等，mq根据记录的信息做出响应的处理。

源码内部逻辑有一个地方就判断了isTimeout是否请求超时，为true则抛出异常

该map requestFutureTable在本次启动中只是使用，具体在什么地方存储的，应该会在后续的生产者发送消息源码中再次出现，本次启动使用到的requestFutureTable应该都是没数据的。日常开发看似只是简单的调用了发送消息的api方法，而mq内部则做了许多复杂的处理来保证消息的可靠性和高可用性