RocketMQ消息发送流程源码剖析
正文
就是说，我们打了个比方，把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消息发送流程的资料请关注脚本之家其它相关文章！
相关文章
这篇文章主要介绍了MyBatis-plus数据库字段排序不准确的解决方案，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2022-02-02
本篇文章给大家详细总结了java中jvm逃逸问题的相关内容，有兴趣的朋友可以根据小编一起学习下。2018-02-02
这篇文章主要给大家介绍了关于Spring的常用配置,文中通过示例代码介绍的非常详细,对大家学习或者使用springboot具有一定的参考学习价值,需要的朋友可以参考下2022-07-07
本文主要介绍了SpringBoot 统一公共返回类的实现，配置后台的统一公共返回类，这样做目的是为了统一返回信息，文中示例代码介绍的很详细，感兴趣的可以了解一下2022-01-01
本文主要分析了hadoop客户端read和write block的流程. 以及client和datanode通信的协议, 数据流格式等2012-11-11
这篇文章主要介绍了SpringBoot 使用jwt进行身份验证的方法示例，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧2018-12-12
本节主要介绍了模拟打印机排队打印效果的具体实现，感兴趣的朋友可以参考下2014-07-07
文章总结：本文介绍了如何在IDEA中配置JDK、Git和Maven,包括设置Java编译器路径、配置Git版本控制、修改Maven根目录以加快jar包下载速度,并提供了一个解决方案以确保配置在新项目中生效2025-01-01
我们在入门案例中,其实已经是一个非常简单的认证,但是用户名是写死的,密码也需要从控制台查看,很显然实际中并不能这么做,下面的学习中,我们来实现基于内存模型的认证以及用户的自定义认证,密码加密等内容,需要的朋友可以参考下2024-07-07
这篇文章主要介绍了微服务Redis-Session共享登录状态,本文采取Spring security做登录校验，用redis做session共享，实现单服务登录可靠性，微服务之间调用的可靠性与通用性，需要的朋友可以参考下2023-12-12
最新评论