RocketMQ消息发送流程源码剖析

 更新时间:2022年08月01日 11:31:50   作者:奔跑的毛球  
这篇文章主要为大家介绍了RocketMQ消息发送流程源码剖析,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
(福利推荐:【腾讯云】服务器最新限时优惠活动,云服务器1核2G仅99元/年、2核4G仅768元/3年,立即抢购>>>:9i0i.cn/qcloud

(福利推荐:你还在原价购买阿里云服务器?现在阿里云0.8折限时抢购活动来啦!4核8G企业云服务器仅2998元/3年,立即抢购>>>:9i0i.cn/aliyun

正文

就是说,我们打了个比方,把RocketMQ比作码头上的一个小房子,来送孩子登船的家长比作生产者,拉走孩子们的船夫比作消费者,所以,RocketMQ的故事就这么展开了。

这节我们研究研究,消息的发送流程。也就是说,消息孩子从进门到坐到message queue座位上都经历了啥。

父母把消息孩子送到码头之后,门口的门童defaultMQProducerImpl.send()接过孩子,进入到MQ房子内部,然后引导孩子进入Broker候船大厅内的message queue座位上就坐。这就是消息发送的流程了。

而且孩子在刚被门童接到之后,就被规定了能在候船大厅待多久,默认是3秒。也就是说,要是再小房子内等了三秒没走,就离开吧,你怕是没想明白自己来干啥的。这就是消息的超时时间。

读源码

1 调用defaultMQProducerImpl.send()

public SendResult send(
    Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
    return send(msg, this.defaultMQProducer.getSendMsgTimeout());
}

2 设置过期时间

public SendResult send(Message msg,
    long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
    return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
}

3 执行defaultMQProducerImpl.sendDefaultImpl()方法

private SendResult sendDefaultImpl(
    Message msg,
    final CommunicationMode communicationMode,
    final SendCallback sendCallback,
    final long timeout
) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {}

这里看看这几个参数,

  • communicationMode 是通信模式,同步异步还是单向
  • sendCallback 是针对异步模式的,异步模式需要设置发送完成后的回调。

sendDefaultImpl是发送消息的核心方法。

这里消息孩子进到第一个卡口,先要检查送孩子来的家长是否还能联系上,若是能联系到,就继续。要是联系不到,这孩子岂不是被抛弃了,不敢接不敢接,送到孤儿院吧。

然后需要检查消息孩子了,首先是检查孩子还在不在,别扔个衣服跑了。
然后看看孩子指定的这个topic,不能说我想去内个topic哈,必须是实实在在的名字。而且上头也规定了,这个topic的名字也不能太长,也不能包含特殊字符。已有的一些领导定过的也不能用哈。
接下来就是检查孩子的body了,之前说body就是孩子的技能,首先,技能为空,不行不行,啥都不会是不行的。再者太长也不行,你唱首歌两年,这没法玩。

检查message不为null

检查topic

  • topic不能为空
  • topic不能太长
  • 不能包含特殊字符

检查话题的名字是否被系统已占用

检查body

  • 检查是否为空
  • 检查长度是否过长,最大为4MB 这样

下边我们看看sendDefaultImpl这个方法。给他拆成一段一段的看。

1 两个校验

//校验生产者服务是ok的,可以联系到的
this.makeSureStateOK();
//校验消息的参数
Validators.checkMessage(msg, this.defaultMQProducer);
  • 第一个检查,检查生产者服务是否是正常工作的,若是不正常工作,就抛出异常。
private void makeSureStateOK() throws MQClientException {
    if (this.serviceState != ServiceState.RUNNING) {
        throw new MQClientException("The producer service state not OK, "
            + this.serviceState
            + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
            null);
    }
}
  • 第二个检查,检查消息本身是否为空,检查topic,检查消息的body
public static void checkMessage(Message msg, DefaultMQProducer defaultMQProducer) throws MQClientException {
    if (null == msg) {
        throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message is null");
    }
    // 这里校验Topic的时候,校验了不能为空,长度和特殊字符
    Validators.checkTopic(msg.getTopic());
    //这里则校验了一些不允许使用的topic名字
    Validators.isNotAllowedSendTopic(msg.getTopic());
    // body不为空
    if (null == msg.getBody()) {
        throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body is null");
    }
    // body长度不为0
    if (0 == msg.getBody().length) {
        throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body length is zero");
    }
    // body 长度不能过长
    if (msg.getBody().length > defaultMQProducer.getMaxMessageSize()) {
        throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL,
            "the message body size over max value, MAX: " + defaultMQProducer.getMaxMessageSize());
    }
}

2 获取topic路由信息

嗯,这里孩子终于通过了检查,服务人员开始带着他去找自己指定的topic区域,指定是自己指定,划分还是工作人员划分的。咱总得知道这个topic区域在哪吧。

先去缓存笔记里找,有没有这个区域的信息,若是没有这个topic,就新建一个,然后更新到缓存笔记里边。若有topic但是不知道在哪,就找name server大脑去申请这个topic在哪的信息。

执行tryToFindTopicPublishInfo方法去获取Topic的路由信息,若是不存在就新建,若是有topic但是缓存中没有路由信息,则通过name server获取路由信息。

TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
    //获取topic信息
    TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
    //不存在
    if (null == topicPublishInfo || !topicPublishInfo.ok()) {
        //新建
        this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
        //修改topic的路由信息并更新到本地
        this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
        topicPublishInfo = this.topicPublishInfoTable.get(topic);
    }
    //包含路由信息就直接返回
    if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
        return topicPublishInfo;
    } else {
        //不包含路由信息则向name server申请,修改topic的路由信息并更新到本地
        this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
        topicPublishInfo = this.topicPublishInfoTable.get(topic);
        return topicPublishInfo;
    }
}

3 计算重试次数

这就是计算消息孩子可以尝试去找地方坐几次,没坐上,欸,我又来了,没坐上,欸,我又来了。

这行代码就是计算重试次数的,根据communicationMode传入的值,同步异步还是单向的来决定重试次数是几次。 很明显,若是同步的,就会尝试三次。若是异步的或者单向的就只发送一次。

int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;

4 执行队列选择方法

我们之前说了,Broker类似于候船大厅,为了均分压力,每次都要进与上次不同的候船大厅。

执行selectOneMessageQueue方法通过Queue将消息发送到与上次不同的一个Broker。也可以通过 sendLatencyFaultEnable判断是否启用延迟容错开关

MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);

5 发送消息

这就是走过巷道坐到属于自己的座位上了

然后就通过sendKernelImpl发送消息了,这是发送消息的核心方法。会准备通信层的入参,并将请求发送给通信层,内部实现是基于Netty的。

sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);

以上就是RocketMQ消息发送流程源码剖析的详细内容,更多关于RocketMQ消息发送流程的资料请关注程序员之家其它相关文章!

相关文章

  • Eclipse/MyEclipse转IntelliJ IDEA完全攻略(图文)

    Eclipse/MyEclipse转IntelliJ IDEA完全攻略(图文)

    这篇文章主要介绍了Eclipse/MyEclipse转IntelliJ IDEA完全攻略(图文),小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2019-01-01
  • 浅谈SpringMVC中的session用法及细节记录

    浅谈SpringMVC中的session用法及细节记录

    下面小编就为大家带来一篇浅谈SpringMVC中的session用法及细节记录。小编觉得挺不错的,现在就分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2017-05-05
  • Java 判断实体对象及所有属性是否为空的操作

    Java 判断实体对象及所有属性是否为空的操作

    这篇文章主要介绍了Java 判断实体对象及所有属性是否为空的操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2020-12-12
  • Java之String.format()方法案例讲解

    Java之String.format()方法案例讲解

    这篇文章主要介绍了Java之String.format()方法案例讲解,本篇文章通过简要的案例,讲解了该项技术的了解与使用,以下就是详细内容,需要的朋友可以参考下
    2021-08-08
  • springcloud feign传输List的坑及解决

    springcloud feign传输List的坑及解决

    这篇文章主要介绍了springcloud feign传输List的坑及解决方案,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2022-03-03
  • Java中Switch的使用方法及新特性

    Java中Switch的使用方法及新特性

    在java中控制流程语句是由选择语句、循环语句、跳转语句构成,选择语句包括if和switch,在过多的使用if语句嵌套会使程序很难阅读,这时就可以用到switch语句,这篇文章主要给大家介绍了关于Java中Switch的使用方法及新特性的相关资料,需要的朋友可以参考下
    2023-11-11
  • JAVA自定义注解详情

    JAVA自定义注解详情

    这篇文章主要介绍了Java自定义注解,结合实例形式总结分析了java常见的自定义注解类型、功能、用法及操作注意事项,需要的朋友可以参考下
    2021-10-10
  • Java线程的调度与优先级详解

    Java线程的调度与优先级详解

    这篇文章主要为大家详细介绍了Java线程的调度与优先级,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下,希望能够给你带来帮助
    2022-03-03
  • 手把手教学Win10同时安装两个版本的JDK并随时切换(JDK8和JDK11)

    手把手教学Win10同时安装两个版本的JDK并随时切换(JDK8和JDK11)

    最近在学习JDK11的一些新特性,但是日常使用基本上都是基于JDK8,因此,需要在win环境下安装多个版本的JDK,下面这篇文章主要给大家介绍了手把手教学Win10同时安装两个版本的JDK(JDK8和JDK11)并随时切换的相关资料,需要的朋友可以参考下
    2023-03-03
  • SpringBoot读取excel表格的示例代码

    SpringBoot读取excel表格的示例代码

    这篇文章主要介绍了SpringBoot读取excel表格的示例代码,代码简单易懂,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2020-10-10

最新评论

?


http://www.vxiaotou.com