目录
消息对象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刷盘等
版权声明：本文不是「本站」原创文章，版权归原作者所有 | 原文地址: