RocketMQ源码分析之生产者
RocketMQ源码分析我们主要从NameSrv、路由、生产者、消费者、消息存储等方面一点点分析,本章主要讲的是生产者的源码分析。
生产者方面的源码主要分为三个地方,启动生产者、发送消息、批量发送消息,然后每个地方又会细分为不同的小步骤,我会一点点分析。
一、前提
使用过RocketMQ的都知道生产者类是DefaultMQProducer,该类在源码的org.apache.rocketmq.client.producer下。通过源码可以看到该类继承了ClientConfig客户端参数配置类同时实现了MQProducer接口,ClientConfig看名字大家都知道是干什么的了,这个就不细说了,至于MQProducer接口咱们继续细看,MQProducer接口继承了MQAdmin接口。
所以从上到下的关系是DefaultMQProducer--》MQProducer --》MQAdmin。MQAdmin中的方法不多,就9个,都是最基础的方法,创建topic、最大最小偏移量等,最主要的消息发送、客服端的启动和注销等方法其实在MQProducer接口中,然后DefaultMQProducer对这些方法进行具体实现。
其实等你接下来细看的时候,会发现DefaultMQProducer方法的实现其实都是DefaultMQProducerImpl类做的具体实现,该类在org.apache.rocketmq.client.impl.producer包下。
二、启动生产者
生产者的启动在DefaultMQProducer的start()方法中实现的,具体看其实是调用DefaultMQProducerImpl的start()的方法。
在start()方法中又细分为一下几个步骤,接下来我会一个个详细讲解每一个步骤。
1、参数校验
其实主要是对生产者组进行校验,组名的长度、组名的规则、组是否为空等的校验。
2、 将MQClient的名字改为当前进程ID
3、创建一个MQClientMananger并通过该管理器创建一个MQClient
JVM中只有一个MQclientManager,通过该管理器创建不同的MQClient,同时维护一个MQClientInstance缓存表,该表中记录了已经创建的MQClient。每次创建新的MQClient的时候,都会先创建一个clientId,然后根据clinetId去该表中查询,如果存在则返回存在的,若不存在,则创建一个新的MQClient实例,然后将该实例信息存储到MQClientInstance缓存表中。
4、将当前Producer注册到MQClient中
5、启动produer
启动produer的时候,会启动一堆东西,主要的启动方法还是this.mQClientAPIImpl.start(),细看会发现其实是启动一个remotingClient,然后remotingClient其实就是一个netty客户端。
三、发送消息
消息的发送分为很多种,异步发送、同步发送、oneway的方式等等,咱们就随便分析其中一个同步发送消息,其实消息的发送也是用defaultMQProducerImpl做的的具体实现,发现消息前其实是先对消息做了校验、然后就是查找topic对应路由的信息、选择发送的对列、发送消息。
1、消息校验
校验消息不为空,消息体不为空,消息体的长度不为空,然后消息的大小不能大于默认值4M。其实还有对topic的校验,这里就不细说了,可以自己进去看看。
2、路由的查找,查找topic对应路由的信息
跟创建MQClient一样,路由相关的信息也会存在本地的一张缓存表,若能根据topic在缓存表中找到路由信息,则直接返回,若找不到则向namesrv请求去获取该topic的路由信息
向namesrv获取topic的路由信息时,若传过来的参数isDefault是true则查询默认的topic的路由信息,怎么这肯定不是查找默认topic的信息,所以查找默认的咱们就不看了。其实查找topic的路由信息,还是通过一个nettyclient向namesrv发送查找路由的请求,然后namesrv返回。
若topic的路由信息获取到了,然后跟topic的路由表信息进行对比,如果两者不一致,则更新路由表中该topic对应的路由信息。同时将获取到的路由信息类topicRouteData转换成需要的TopicPublishInfo类。
到此为止,无论是从本地缓存表中获取的还是向namesrv获取的,topic对应路由的信息就拿到了。接下来就是选择对列,每个broker上有不用的queue,需要选择队列,决定将消息发送到哪个或者哪些队列。
3、选择队列
找到topic对应的路由信息以后,开始选择队列,将该topic的路由信息和接收最后一次消息的broker的名字传过去,为什么要传lastBrokerName?是因为要做负载均衡,每次发送消息都发送到不同的broker上,保证消息不会都到一个broker上,达到负载均衡的作用,具体怎么实现,我们继续看。
selectOneMessageQueue()方法中有个判断,判断sendLatencyFaultEnable值为true或者false,然后做不同的操作。sendLatencyFaultEnable的值默认为false,该属性是消息发送失败延迟发送的开关,既然默认是false,咱们就先看为false时候对于队列的选取,然后再看为true的情况。
如果没有传入 lastBrokerName,则进入selectOneMessageQueue()方法。
该方法中也就是,系统自动生成一个索引,然后该索引对消息队列大小进行取余,然后再根据取余结果进行获取消息队列。
再返回,如果lastBrokerName不为null,则还是根据索引对消息队列大小取余,然后根据取余结果获取到消息队列,如果该消息队列所属的broke跟传过来的是同一个broker,则跳过重新获取。若一直只有一个broker,那就没办法了,只能重新调用selectOneMessageQueue()方法,在该broker的队列进行分配。
我们再返回到selectOneMessageQueue()中sendLatencyFaultEnable值为true的情况,即设置了发送失败延迟再发送的功能。
还是跟之前一样,利用索引对消息队列大小进行取余,然后根据结果进行获取消息队列。唯一不同的是比较该消息队列所属的broker是否可用,producer本地会存储消息发送失败的broker的名字,然后跟它们进行对比,如果不存在其中,则代表该broker可用,将MessageQueue返回。
若选取的MessageQueue所属的broker在那些失败的broker之内,则放弃那个MessageQueue,从这些失败的broker中挑出一个不是最好的broker,然后在topic的路由信息中获取该broker的读队列,如果读队列大于0,则代表该broker的队列处于空闲可被写入。然后再在topic的路由信息中获取一个MessageQueue,然后把这个MessageQueue进行填充为那个不是不好的broker的队列。
至此,消息队列的选取已分析结束,接下来就是发送消息。
4、发送消息
现在消息已经校验、路由已经获取、消息队列已经确定,那么就该发送消息了,我们就回到DefaultMQProducerImpl的
sendDefaultImpl消息发送方法了。发送消息其实是调用了DefaultMQProducerImpl的sendKernelImpl()方法。
sendKernelImpl()方法中先是根据MessageQueue获取broker,然后再获取broker对应的brokerAddr,ip地址。
然后给message设置唯一id
接着就是判断消息是否大于4k,如果大于4k,则设置一个压缩标识,消息发送时根据该标识进行压缩。
执行钩子函数,如果之前给注册了一个钩子函数,做一些特殊的处理工作,那么发送消息之前会将钩子函数中的业务给执行了
由于是网络通信,所以封装消息头,然后就是消息的发送了,根据第二步启动创建的MQClient进行发送消息。
四、批量发送消息
批量发送消息可以将同一个topic下的多条消息一起发送,节省资源浪费,提高效率,但不是说该topic下所有的消息都合并成批量消息发送,消息的发送有限制的,之前说过4M,消息不是越多越好,而且大于4M就无法发送了。
其实批量消息的发送跟单条消息的发送就只有一个区别,批量是讲多条消息合并,然后再按照单条消息发送的方式发送,因此,咱们只要看消息合并的那个方法即可,至于消息发送,跟单条的一样。
还是DefaultMQProducer的send方法,不过是参数是个集合,多条消息的集合。
其实很简单,就是将传过来的消息集合转换成MessageBatch,该类其实也是继承Message。然后针对每条消息进行验证,给消息设置唯一id,给消息设置topic,然后对批量消息类进行编码,再给批量消息类设置topic,然后返回批量消息类MessageBatch。