RocketMQ消息发送流程
共 7637字，需浏览 16分钟
· 2022-05-22
记得点击 "欢少的成长之路", 设为星标⭐
后台点击【联系我】，申请加入优质技术学习社群
大家好，我是Leo。
今天聊一下RocketMQ消息发送，重试机制，故障延迟机制，获取路由机制，消息队列的选择
消息发送
关系图
首先放一下Broker Cluster，Broker，Topic，Queue的关系图。因为下文主要会沿着这四块进行梳理
发送的三种方式
消息发送的三种方式
同步：发送者向MQ发送一条消息后，一直等待服务器返回成功才继续下一个。 异步：发送者向MQ发送一条消息后，通过回调函数调用消息发布函数继续发送，主线程立即返回。 单向：发送者向MQ发送一条消息后，直接返回，不等待消息服务器的结果，也不注册函数，简单来说，就是只管发。其他啥也不管。
源码
package org.apache.rocketmq.client.impl;
/**
* 消息发送的三种方式
*/
public enum CommunicationMode {
// 同步发送
SYNC,
// 异步发送
ASYNC,
// 单向发送
ONEWAY,
}
重试机制
RocketMQ的重试机制，主要由下列两个参数决定。默认重试次数为2次，重试机制提高了消息发送成功的几率。
/**
* 同步模式下内部尝试发送消息的最大次数
*/
private int retryTimesWhenSendFailed = 2
/**
* 异步模式下内部尝试发送消息的最大次数
*/
private int retryTimesWhenSendAsyncFailed = 2;
故障延迟机制
RocketMQ的故障延迟机制，主要由下列参数决定，默认是不开启的。故障延迟机制，主要体现在集群的时候，当broker发送错误时，可以有效的规避多次发送消息都发往一个broker（queue）的错误。
/**
* 默认不启用Broker故障延迟机制。
*/
private boolean sendLatencyFaultEnable = false;
获取路由信息机制
消息在发送时，需要知道，要发往哪个broker。首先会去 brokerAddrTable 中查找当前brokerName是否存在在本地的缓存中
如果成功返回brokerName 否则就返回null
/* Broker Name */ /* brokerId */ /* address */
private final ConcurrentMap> brokerAddrTable;
如果成功一帆风顺，如果找不到的话，肯定要做一些安全处理。
如果找不到的话，会通过 tryToFindTopicPublishInfo 函数尝试查找主题发布信息
在 topicPublishInfoTable缓存中根据topic名称查找是否存在如果没有缓存，会创建一个以topic名称为key，空 TopicPublishInfo为value到topicPublishInfoTable，然后更新到NameServer。如果消息的路由信息存在，并且 MessageQueue不为空 直接返回路由信息否则使用默认主题
/* topic */
private final ConcurrentMap topicPublishInfoTable
消息队列的选择
开启故障延迟
遍历主题队列的消息队列，根据访问次数进行随机自增取模。
如果当前消息队列是可用的就直接返回。函数名 isAvailable
如果是不可用的，从失败的brokeName列表中通过 pickOneAtLeast 函数选择一个可用的broker。拿到brokerName之后，再根据brokerName反查这个队列的写队列数
如果小于0说明该broker依据恢复，从失败的条目中移出当前broker 如果大于0通过 selectOneMessageQueue函数选出一个消息队列
/**
* 失败的broker列表
*/
private final LatencyFaultTolerance < String > latencyFaultTolerance;
没有开启故障延迟
如果上一次选择的执行发送消息失败的broker名称为空，它会通过 selectOneMessageQueue() 函数对当前访问的次数取绝对值，然后与消息队列的大小取模得到一个下标，然后从 messageQueueList 中根据下标取出 MessageQueue
如果上一次选择的执行发送消息失败的broker名称不为空，会遍历消息队列，对当前访问的次数取绝对值，然后与消息队列的大小取模得到一个下标后，拿着下标获取对应的 brokerName 并且判断当前的 brokerName 是否与上一次发送消息失败的 brokerName 相等，
如果相等就遍历所有主题内的消息队列。假如还是没有找到一个合适的，就会随机选择一个 如果不相等，就把当前下标的 MessageQueue返出去。
下图的json字符串就是 MessageQueue 信息
/**
* 该主题队列的消息队列
*/
private List < MessageQueue > messageQueueList = new ArrayList < MessageQueue > ();
[
{
"brokerName": "broker-a",
"queueId": 0
},
{
"brokerName": "broker-a",
"queueId": 1
}, {
"brokerName": "broker-b",
"queueId": 0
}, {
"brokerName": "broker-b",
"queueId": 1
}, {
"brokerName": "broker-c",
"queueId": 0
}, {
"brokerName": "broker-c",
"queueId": 1
}
]
开启故障延迟机制中的可用依据是：检查时间是否到达了下次可使用的时间点
如果没有该机制，如果broker宕机，由于路由算法中的消息队列是按broker排序的，顺序选择，如果上一次根据路由算法选择的是宕机的broker的第一个队列，那么随后的下次选择的是宕机broker的第二个队列，消息发送很有可能会失败，再次引发重试，带来不必要的性能损耗。
selectOneMessageQueue()也可以看成是兜底策略-轮询算法
同步发送
由上文得知，消息发送有三种方式。我们先看一下同步发送主要做了哪些事情。
DefaultMQProducerImpl的send函数是发送消息的入口
通过 makeSureStateOK函数检查服务状态是否正常通过 checkMessage函数校验Message与DefaultMQProducer是否符合发送的规则校验消息的主题不能等于消息队列集合的主题信息以及以上操作是否超时 校验brokerName是否存在，如果不存在通过 findBrokerAddressInPublish函数去nameserver拉取通过 brokerVIPChannel函数校验是否使用了vip管道，如果使用了管道在原来的基础上把端口-2通过配置信息获取生成uniqId的算法规则以及封装 Message的实例信息对 Message 的 body信息进行压缩获取当前的配置信息，是否启用事务。 封装发送消息模板权限信息 SendMessageContext，构造请求头发送之前，校验一下 Topid类型是否属于重试类型消息（这里可以看看下列注释）通过 CommunicationMode枚举类型判断当前是什么发送方式判断当前是正常指令发送，还是RPC指令发送，判断是否对字段进行压缩处理（简化压缩有助于提速序列化速度） 根据broker地址获取Netty对应的Channel，并远程调用（这里的发送，用的是Netty框架） 通过 processSendResponse函数处理同步返回的参数，如果参数为0，说明发送成功。最后封装SendResult返回
第四步中，如果在nameserver拉取不到，说明服务宕机了。
第五步中，vip的管道配置从配置文件中的com.rocketmq.sendMessageWithVIPChannel得知
第六步中，批量信息不支持压缩
第十步中，如果是重试消息，通过获取自定义重试次数，在请求头区分特别处理
第十一步中，因为这里介绍的是同步发送，就只写同步发送流程了，异步，单向会在下面段落体现出来
第十二步中，通过配置文件中org.apache.rocketmq.client.sendSmartMsg得知字段是否简化压缩
异步发送
聊完同步发送，我们看一下异步发送
DefaultMQProducerImpl的send函数是发送消息的入口（这里跟同步的区别是多了一个 SendCallback）
在生产者生产消息发送时，通过 ExecutorService新增一个异步任务进行发送（可看下列注释，可看源码区）通过 makeSureStateOK函数检查服务状态是否正常通过 checkMessage函数校验Message与DefaultMQProducer是否符合发送的规则校验消息的主题不能等于消息队列集合的主题信息以及以上操作是否超时 校验brokerName是否存在，如果不存在通过 findBrokerAddressInPublish函数去nameserver拉取通过 brokerVIPChannel函数校验是否使用了vip管道，如果使用了管道在原来的基础上把端口-2通过配置信息获取生成uniqId的算法规则以及封装 Message的实例信息对 Message 的 body信息进行压缩获取当前的配置信息，是否启用事务。 封装发送消息模板权限信息 SendMessageContext，构造请求头发送之前，校验一下 Topid类型是否属于重试类型消息（这里可以看看下列注释）通过 CommunicationMode枚举类型判断当前是什么发送方式判断当前是正常指令发送，还是RPC指令发送，判断是否对字段进行压缩处理（简化压缩有助于提速序列化速度） 根据broker地址获取Netty对应的Channel，并远程调用（这里的发送，用的是Netty框架） 通过 processSendResponse函数处理并且利用委托remotingClient.invokeAsync等待返回的SendResult结构体上一步骤再插一句，异步发送会处理一个 updateFaultItem函数记录当前不可以时间/可用时间时间
第一步中 借助 java.util.concurrent.ExecutorService ，实现一个线程池达到可以让任务在后台执行。
第十五步中 RemotingClient的invokeAsync函数
单向发送
单向发送，与同步发送相似。与同步发送不同的是通过 RemotingClient#invokeOneway 函数委托发送。
从 invokeOnway进入后
如果当前addr为空，获取和创建Nameserver通道 创建成功后，只要通道是活跃的，并且不为空，就利用Netty框架进行 writeAndFlush
创建通道时，通过 ReentrantLock 对nameSeverChannel加锁，超时时长为3秒
源码
DefaultMQProducerImpl#send同步函数
/**
* 内核同步发送
* @param msg
* @param mq
* @return
* @throws MQClientException
* @throws RemotingException
* @throws MQBrokerException
* @throws InterruptedException
*/
public SendResult send(Message msg, MessageQueue mq) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
return send(msg, mq, this.defaultMQProducer.getSendMsgTimeout());
}
/**
* 内核同步发送下的 send子函数
* @param msg
* @param mq
* @param timeout
* @return
* @throws MQClientException
* @throws RemotingException
* @throws MQBrokerException
* @throws InterruptedException
*/
public SendResult send(Message msg, MessageQueue mq, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
long beginStartTime = System.currentTimeMillis();
this.makeSureStateOK();
Validators.checkMessage(msg, this.defaultMQProducer);
if (!msg.getTopic().equals(mq.getTopic())) {
// 消息的主题不等于mq的主题
throw new MQClientException("message's topic not equal mq's topic", null);
}
long costTime = System.currentTimeMillis() - beginStartTime;
if (timeout < costTime) {
throw new RemotingTooMuchRequestException("call timeout");
}
return this.sendKernelImpl(msg, mq, CommunicationMode.SYNC, null, null, timeout);
}
DefaultMQProducerImpl#send异步函数
/**
* 内核异步
*
* @param msg
* @param mq
* @param sendCallback
* @throws MQClientException
* @throws RemotingException
* @throws InterruptedException
*/
public void send(Message msg, MessageQueue mq, SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {
send(msg, mq, sendCallback, this.defaultMQProducer.getSendMsgTimeout());
}
/**
* 内核异步发送下的 send 子函数
* @param msg
* @param mq
* @param sendCallback
* @param timeout the sendCallback will be invoked at most time
* @throws MQClientException
* @throws RemotingException
* @throws InterruptedException
*/
@Deprecated
public void send(final Message msg, final MessageQueue mq, final SendCallback sendCallback, final long timeout) throws MQClientException, RemotingException, InterruptedException {
final long beginStartTime = System.currentTimeMillis();
ExecutorService executor = this.getAsyncSenderExecutor();
try {
executor.submit(new Runnable() {
@Override
public void run() {
try {
makeSureStateOK();
Validators.checkMessage(msg, defaultMQProducer);
if (!msg.getTopic().equals(mq.getTopic())) {
throw new MQClientException("message's topic not equal mq's topic", null);
}
long costTime = System.currentTimeMillis() - beginStartTime;
if (timeout > costTime) {
try {
sendKernelImpl(msg, mq, CommunicationMode.ASYNC, sendCallback, null, timeout - costTime);
} catch (MQBrokerException e) {
throw new MQClientException("unknown exception", e);
}
} else {
sendCallback.onException(new RemotingTooMuchRequestException("call timeout"));
}
} catch (Exception e) {
sendCallback.onException(e);
}
}
});
} catch (RejectedExecutionException e) {
throw new MQClientException("executor rejected ", e);
}
}
DefaultMQProducerImpl#sendOneway函数
/**
* 内核单向发送
*/
public void sendOneway(Message msg, MessageQueue mq) throws MQClientException, RemotingException, InterruptedException {
this.makeSureStateOK();
Validators.checkMessage(msg, this.defaultMQProducer);
try {
this.sendKernelImpl(msg, mq, CommunicationMode.ONEWAY, null, null, this.defaultMQProducer.getSendMsgTimeout());
} catch (MQBrokerException e) {
throw new MQClientException("unknown exception", e);
}
}
往期推荐
结尾
单向发送那里如果有问题，可以私信我。我们一起交流！
关于整篇的思路与总结。主要是从RocketMQ的消息发送入手的，消息发送主要分三种
同步 异步 单向
从三种方式各自深入源码进行分析得知，同步，单向，异步流程大致相同
异步发送与同步发送最大的不同： 异步发送在同步发送的基础上利用ExecutorService 进行初始化异步任务。在执行完成之后，还会有一个 updateFaultItem 时间记录处理。
正常情况下，会传一个false值，false代表没有问题，会采用我们自己计算的时间戳赋值 异常情况下，会传一个true值，true代表有问题，会采用默认时间30s赋值
单向又与同步，异步有些不同，单向因为不需要知道是否成功，所以他把这条发送请求进行委托处理（利用Netty框架Channel的 writeAndFlush）
非常欢迎大家加我个人微信有关后端方面的问题我们在群内一起讨论！ 我们下期再见！
欢迎『点赞』、『在看』、『转发』三连支持一下，下次见~