• RocketMQ源码阅读(九)DefaultMQProducer消息发送


    目录

    消息对象Message

    消息发送

    ​--消息验证

    --消息路由

    tryToFindTopicPublishInfo

    selectOneMessageQueue

    一、默认情况下的选择策略

    二、故障延迟机制

    --消息发送

    Broker入口


    消息发送代码

    消息对象Message

    消息发送

    --消息验证

    验证Topic命名是否合规,有些RocketMQ自用的topic名不允许我们用

    验证消息体是否为空,是否超过最大长度,默认最大4M

    --消息路由

    如上代码,消息路由分为两步,一获取Topic的路由信息,二从多个队列中选择一个队列

    tryToFindTopicPublishInfo

    先查本地缓存,没有再访问NameSrv获取Topic对应的路由信息

    MQClientInstance#updateTopicRouteInfoFromNameServer

    selectOneMessageQueue

    在消息队列的选择时有两种机制,使用perducer的sendLatencyFaultEnable参数控制

    一、默认情况下的选择策略

    sendLatencyFaultEnable=false,如果第一次发送失败,再次尝试发送时会选择其他broker上的queue进行发送

    二、故障延迟机制

    sendLatencyFaultEnable=true,有了上边的默认选择机制,例如有两个Broker(A/B),在一次send()方法中,BrokerA宕机,第一次send失败,重试时自动跳过A选择BrokerB上的queue发送。

    但是在新一轮的send()方法中很可能依旧需要向失败的A中队列发起请求,消息很可能失败,再次引发重试,有不必要的性能损失。所以故障延迟机制提供了一种解决办法,在BrokerA发送失败时,在之后的一段时间内暂时将BrokerA隔离,不提供BrokerA的队列选择。

    首先:在发送异常时记录失败的brokerName和隔离时间

    正常发送成功消息,使用本次发送延迟时间进行隔离

    发送消息失败,使用默认的30000进行隔离

    延时时间和隔离时间对应关系和代码如下:

    例如,发送耗时50和100毫秒,则当前broker正常不隔离;发送耗时3秒,则当前broker延迟3分钟;如果发送失败,直接隔离10分钟。

    其次:选择消息队列时判断隔离时间是否结束

    --消息发送

    设置消息ID,消息压缩,设置各种标记

    钩子函数进行发送前的增强

     消息发送

     组装请求命令RemotingCommand,主要是设置请求code,本次发送是RequestCode.SEND_MESSAGE_V2

    Broker入口

    broker和namesrv的接口逻辑处理都是由BrokerController和NameSrvController的start()方法中注册的Processor(registerProcessor方法)来实现的,如下

    org.apache.rocketmq.broker.BrokerController#registerProcessor

    org.apache.rocketmq.broker.processor.SendMessageProcessor#processRequest

    内部就是解析 RemotingCommand中的消息体,调用DefaultMessageStore写CommitLog刷盘等

  • 相关阅读:
    java基于ssm+vue高校人事管理系统
    Proxy-Reflect
    No module ‘xformers‘. Proceeding without it.
    虚拟主机什么时候适合更换成云主机?
    mysql oracle统计报表每天每月每年SQL
    领域驱动设计——模型
    【iOS】折叠cell
    使用CEF(六)— 解读CEF的cmake工程配置
    Android和JS互相调用
    C#中实现按位域操作
  • 原文地址:https://blog.csdn.net/xyjy11/article/details/126144182