RocketMQ消息发送流程

Posted 2022-05-21 Leo聊技术

大家好，我是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是否存在在本地的缓存中

1. 如果成功返回brokerName
2. 否则就返回null

/* Broker Name */   /* brokerId */    /* address */
private final ConcurrentMap<String, HashMap<Long, String>> brokerAddrTable;

如果成功一帆风顺，如果找不到的话，肯定要做一些安全处理。

如果找不到的话，会通过 ​​tryToFindTopicPublishInfo​​ 函数尝试查找主题发布信息

1. 在 topicPublishInfoTable 缓存中根据topic名称查找是否存在
2. 如果没有缓存，会创建一个以topic名称为key，空 TopicPublishInfo 为value到 topicPublishInfoTable ，然后更新到 NameServer 。
3. 如果消息的路由信息存在，并且 MessageQueue 不为空 直接返回路由信息
4. 否则使用默认主题

/* topic */
private final ConcurrentMap<String, TopicPublishInfo> topicPublishInfoTable

消息队列的选择

开启故障延迟

遍历主题队列的消息队列，根据访问次数进行随机自增取模。

如果当前消息队列是可用的就直接返回。 函数名 ​​isAvailable​​

如果是不可用的，从失败的brokeName列表中通过 ​​pickOneAtLeast​​ 函数选择一个可用的broker。拿到brokerName之后，再根据brokerName反查这个队列的写队列数

1. 如果小于0说明该broker依据恢复，从失败的条目中移出当前broker
2. 如果大于0通过 selectOneMessageQueue 函数选出一个消息队列

/**
 * 失败的broker列表
 */
private final LatencyFaultTolerance < String > latencyFaultTolerance;

没有开启故障延迟

如果上一次选择的执行发送消息失败的broker名称为空，它会通过 ​​selectOneMessageQueue()​​ 函数对当前访问的次数取绝对值，然后与消息队列的大小取模得到一个下标，然后从 ​​messageQueueList​​ 中根据下标取出 ​​MessageQueue​​

如果上一次选择的执行发送消息失败的broker名称不为空，会遍历消息队列，对当前访问的次数取绝对值，然后与消息队列的大小取模得到一个下标后，拿着下标获取对应的 ​​brokerName​​ 并且判断当前的 ​​brokerName​​ 是否与上一次发送消息失败的 ​​brokerName​​ 相等，

1. 如果相等就遍历所有主题内的消息队列。假如还是没有找到一个合适的，就会随机选择一个
2. 如果不相等，就把当前下标的 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函数是发送消息的入口

1. 通过 makeSureStateOK 函数检查服务状态是否正常
2. 通过 checkMessage 函数校验 Message 与 DefaultMQProducer 是否符合发送的规则
3. 校验消息的主题不能等于消息队列集合的主题信息以及以上操作是否超时
4. 校验brokerName是否存在，如果不存在通过 findBrokerAddressInPublish 函数去nameserver拉取
5. 通过 brokerVIPChannel 函数校验是否使用了vip管道，如果使用了管道在原来的基础上把 端口-2
6. 通过配置信息获取生成uniqId的算法规则以及封装 Message 的实例信息
7. 对 Message 的 body 信息进行压缩
8. 获取当前的配置信息，是否启用事务。
9. 封装发送消息模板权限信息 SendMessageContext，构造请求头
10. 发送之前，校验一下 Topid 类型是否属于重试类型消息（这里可以看看下列注释）
11. 通过 CommunicationMode 枚举类型判断当前是什么发送方式
12. 判断当前是正常指令发送，还是RPC指令发送，判断是否对字段进行压缩处理（简化压缩有助于提速序列化速度）
13. 根据broker地址获取Netty对应的Channel，并远程调用（这里的发送，用的是Netty框架）
14. 通过 processSendResponse 函数处理同步返回的参数，如果参数为0，说明发送成功。最后封装 SendResult 返回

第四步中，如果在nameserver拉取不到，说明服务宕机了。

第五步中，vip的管道配置从配置文件中的com.rocketmq.sendMessageWithVIPChannel得知

第六步中，批量信息不支持压缩

第十步中，如果是重试消息，通过获取自定义重试次数，在请求头区分特别处理

第十一步中，因为这里介绍的是同步发送，就只写同步发送流程了，异步，单向会在下面段落体现出来

第十二步中，通过配置文件中org.apache.rocketmq.client.sendSmartMsg得知字段是否简化压缩

异步发送

聊完同步发送，我们看一下异步发送

DefaultMQProducerImpl的send函数是发送消息的入口（这里跟同步的区别是多了一个 ​​SendCallback​​）

1. 在生产者生产消息发送时，通过 ExecutorService 新增一个异步任务进行发送（可看下列注释，可看源码区）
2. 通过 makeSureStateOK 函数检查服务状态是否正常
3. 通过 checkMessage 函数校验 Message 与 DefaultMQProducer 是否符合发送的规则
4. 校验消息的主题不能等于消息队列集合的主题信息以及以上操作是否超时
5. 校验brokerName是否存在，如果不存在通过 findBrokerAddressInPublish 函数去nameserver拉取
6. 通过 brokerVIPChannel 函数校验是否使用了vip管道，如果使用了管道在原来的基础上把 端口-2
7. 通过配置信息获取生成uniqId的算法规则以及封装 Message 的实例信息
8. 对 Message 的 body 信息进行压缩
9. 获取当前的配置信息，是否启用事务。
10. 封装发送消息模板权限信息 SendMessageContext，构造请求头
11. 发送之前，校验一下 Topid 类型是否属于重试类型消息（这里可以看看下列注释）
12. 通过 CommunicationMode 枚举类型判断当前是什么发送方式
13. 判断当前是正常指令发送，还是RPC指令发送，判断是否对字段进行压缩处理（简化压缩有助于提速序列化速度）
14. 根据broker地址获取Netty对应的Channel，并远程调用（这里的发送，用的是Netty框架）
15. 通过 processSendResponse 函数处理并且利用委托 remotingClient.invokeAsync 等待返回的SendResult 结构体
16. 上一步骤再插一句，异步发送会处理一个 updateFaultItem 函数记录当前 不可以时间/可用时间 时间

第一步中 借助 java.util.concurrent.ExecutorService ，实现一个线程池达到可以让任务在后台执行。

第十五步中 RemotingClient的invokeAsync函数

单向发送

单向发送，与同步发送相似。与同步发送不同的是通过 RemotingClient#invokeOneway 函数委托发送。

从 invokeOnway进入后

1. 如果当前addr为空，获取和创建Nameserver通道
2. 创建成功后，只要通道是活跃的，并且不为空，就利用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("messages topic not equal mqs 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 <code>sendCallback</code> 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("messages topic not equal mqs 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);
    

往期推荐

​​2022年4月文章目录整理​​

​​RocketMQ数据压缩的那套把戏​​

​​图文并茂！深入理解RocketMQ的刷盘机制​​

​​图文并茂！深入了解RocketMQ的过期删除机制​​

​​图文并茂！深入了解RocketMQ的内存映射机制​​

结尾

单向发送那里如果有问题，可以私信我。我们一起交流！

关于整篇的思路与总结。主要是从RocketMQ的消息发送入手的，消息发送主要分三种

1. 同步
2. 异步
3. 单向

从三种方式各自深入源码进行分析得知，同步，单向，异步流程大致相同

异步发送与同步发送最大的不同： 异步发送在同步发送的基础上利用ExecutorService 进行初始化异步任务。在执行完成之后，还会有一个 ​​updateFaultItem​​ 时间记录处理。

1. 正常情况下，会传一个false值，false代表没有问题，会采用我们自己计算的时间戳赋值
2. 异常情况下，会传一个true值，true代表有问题，会采用默认时间30s赋值

单向又与同步，异步有些不同，单向因为不需要知道是否成功，所以他把这条发送请求进行委托处理（利用Netty框架Channel的 ​​writeAndFlush​​）

有关后端方面的问题，非常欢迎大家咨询我，我们在群内一起讨论！ 我们下期再见！

欢迎『点赞』、『在看』、『转发』三连支持一下，下次见~