目录
前言
获取源码
总概论
生产者实例
源码
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内部则做了许多复杂的处理来保证消息的可靠性和高可用性
A-03:开启监控和处理同步发送和异步发送操作的守护线程
- guardThreadForSyncSend.start();
- guardThreadForAsyncSend.start();
这些线程中,可以实现具体的监控和处理逻辑,例如检测发送超时、重试失败的发送操作等。 并且这些线程在 JVM 退出时会自动终止
A-04:开启消息轨迹和发送机制
通过收集消息轨迹信息,可以了解消息在 RocketMQ 中的流转路径,帮助系统监控和故障排查。
总结
对于RocketMQ我们都知道生产者会从nameserver中拉取数据,并且会在本地存储,就算nameserver服务意外离线了,也能通过本地保存的数据进行消息通信。那么如何远程拉取数据以及心跳监测,如何在本地存储,我想大家通过对上面start启动源码的学习,疑惑都解开了吧。
- 数据更新以及心跳无非就是通过一系列的定时器在不断请远程请求
- 本地存储则是使用已 table为后缀命名的Map集合来存储的
对本章源码中遇到的定时器和table进行了整理,方便大家快速记忆
展望
本章内容比较多,博主也是肝了几天才完成,希望对大家都有所收获,下一章我们继续对生产者send消息源码进行学习!