面试官你来简单回答一下RocketMQ的默认发送流程

　　今天我们就开始学习下默认消息发送流程，学习他的实现思路，也帮助我们工作中，遇到了问题不会手足无措。
　　思考问题消息发送者是如何做负载均衡的？消息发送者是如何保证高可用的？消息发送批量消息如何保证一致性的？默认发送流程工作原理
　　源码入口：org。apache。rocketmq。client。producer。DefaultMQProducersend（org。apache。rocketmq。common。message。Message）
　　启动Demo：DefaultMQProducerproducernewDefaultMQProducer（group1）；producer。setNamesrvAddr（xxx：9876）；producer。start（）；MessagemsgnewMessage（TopicTestTopic，TagATag，（HelloRocketMQi）。getBytes（RemotingHelper。DEFAULTCHARSET）Messagebody）；SendResultsendResultproducer。send（msg）；
　　流程：
　　1。校验主题，设置主题msg。setTopic（withNamespace（msg。getTopic（）））；publicStringwithNamespace（Stringresource）｛returnNamespaceUtil。wrapNamespace（this。getNamespace（），resource）；｝
　　2。默认发送方式为同步发送，默认超时时间为3sprivateintsendMsgTimeout3000；publicSendResultsend（Messagemsg，longtimeout）throwsMQClientException，RemotingException，MQBrokerException，InterruptedException｛returnthis。sendDefaultImpl（msg，CommunicationMode。SYNC，null，timeout）；｝
　　3。确认producerservice运行状态是否为运行中
　　入口：org。apache。rocketmq。client。impl。producer。DefaultMQProducerImplmakeSureStateOK检查状态，如果不是RUNNING状态则抛出异常privatevoidmakeSureStateOK（）throwsMQClientException｛if（this。serviceState！ServiceState。RUNNING）｛thrownewMQClientException（TheproducerservicestatenotOK，this。serviceStateFAQUrl。suggestTodo（FAQUrl。CLIENTSERVICENOTOK），null）；｝｝
　　4。校验信息topic长达是否大于TOPICMAXLENGTH，topic是否为空是否通过正则校验，body是否为空，body大小是否超过4MpublicstaticvoidcheckTopic（Stringtopic）throwsMQClientException｛if（UtilAll。isBlank（topic））｛thrownewMQClientException（Thespecifiedtopicisblank，null）；｝if（topic。length（）TOPICMAXLENGTH）｛thrownewMQClientException（String。format（Thespecifiedtopicislongerthantopicmaxlengthd。，TOPICMAXLENGTH），null）；｝if（isTopicOrGroupIllegal（topic））｛thrownewMQClientException（String。format（Thespecifiedtopic〔s〕containsillegalcharacters，allowingonlys，topic，〔azAZ09〕34；），null）；｝｝bodyif（nullmsg。getBody（））｛thrownewMQClientException（ResponseCode。MESSAGEILLEGAL，themessagebodyisnull）；｝if（0msg。getBody（）。length）｛thrownewMQClientException（ResponseCode。MESSAGEILLEGAL，themessagebodylengthiszero）；｝if（msg。getBody（）。lengthdefaultMQProducer。getMaxMessageSize（））｛thrownewMQClientException（ResponseCode。MESSAGEILLEGAL，themessagebodysizeovermaxvalue，MAX：defaultMQProducer。getMaxMessageSize（））；
　　5。找到主题发布的信息，未找到则抛出异常
　　入口：org。apache。rocketmq。client。impl。producer。DefaultMQProducerImpltryToFindTopicPublishInfo
　　消息生产者更新和维护路由信息缓存privateTopicPublishInfotryToFindTopicPublishInfo（finalStringtopic）｛TopicPublishInfotopicPublishInfothis。topicPublishInfoTable。get（topic）；if（nulltopicPublishInfo！topicPublishInfo。ok（））｛this。topicPublishInfoTable。putIfAbsent（topic，newTopicPublishInfo（））；消息生产者更新和维护路由信息缓存this。mQClientFactory。updateTopicRouteInfoFromNameServer（topic）；topicPublishInfothis。topicPublishInfoTable。get（topic）；｝if（topicPublishInfo。isHaveTopicRouterInfo（）topicPublishInfo。ok（））｛returntopicPublishInfo；｝else｛this。mQClientFactory。updateTopicRouteInfoFromNameServer（topic，true，this。defaultMQProducer）；topicPublishInfothis。topicPublishInfoTable。get（topic）；returntopicPublishInfo；｝｝
　　6。通过TopicPublishInfo找到对应的MessageQueue下的，BrokerName信息
　　入口：org。apache。rocketmq。client。impl。producer。DefaultMQProducerImplselectOneMessageQueue
　　获取到BrokerName对应的MessageQueue信息publicMessageQueueselectOneMessageQueue（finalStringlastBrokerName）｛if（lastBrokerNamenull）｛returnselectOneMessageQueue（）；｝else｛for（inti0；ithis。messageQueueList。size（）；i）｛intindexthis。sendWhichQueue。incrementAndGet（）；intposMath。abs（index）this。messageQueueList。size（）；if（pos0）pos0；MessageQueuemqthis。messageQueueList。get（pos）；if（！mq。getBrokerName（）。equals（lastBrokerName））｛returnmq；｝｝returnselectOneMessageQueue（）；｝｝
　　如果lastBrokerName为null，通过对sendWhichQueue方法获取一个队列
　　取余，然后从messageQueueList中获取一个MessageQueuepublicMessageQueueselectOneMessageQueue（）｛intindexthis。sendWhichQueue。incrementAndGet（）；intposMath。abs（index）this。messageQueueList。size（）；if（pos0）pos0；returnthis。messageQueueList。get（pos）；｝
　　7。最后消息发送
　　入口：org。apache。rocketmq。client。impl。producer。DefaultMQProducerImplsendKernelImpl
　　1。根绝BrokerName获取到broker地址
　　在启动阶段，对BrokerAddrTable信息进行了维护publicStringfindBrokerAddressInPublish（finalStringbrokerName）｛HashMapLongbrokerId，Stringaddressmapthis。brokerAddrTable。get（brokerName）；if（map！null！map。isEmpty（））｛returnmap。get（MixAll。MASTERID）；｝returnnull；｝
　　如果未找到，则通过主题查找主题信息，通过更新路由信息后，在尝试获取，如果还未找到则抛出异常if（nullbrokerAddr）｛1。1如果未找到，则通过主题查找主题信息，通过更新路由信息后，在尝试获取，如果还未找到则抛出异常tryToFindTopicPublishInfo（mq。getTopic（））；brokerAddrthis。mQClientFactory。findBrokerAddressInPublish（mq。getBrokerName（））；｝
　　2。为消息分配全局唯一ID为消息分配全局唯一IDif（！（msginstanceofMessageBatch））｛MessageClientIDSetter。setUniqID（msg）；｝
　　在RocketMQ消息发送请求与响应文章中，我们已经学习了请求参数中，创建了全局唯一的MsgId，可以回头看一看
　　3。注册钩子消息发送钩子函数
　　这里主要做了三件事情，确认MsgType类型、是否为延迟消息、调用钩子函数内的方法if（this。hasSendMessageHook（））｛contextnewSendMessageContext（）；context。setProducer（this）；context。setProducerGroup（this。defaultMQProducer。getProducerGroup（））；context。setCommunicationMode（communicationMode）；context。setBornHost（this。defaultMQProducer。getClientIP（））；context。setBrokerAddr（brokerAddr）；context。setMessage（msg）；context。setMq（mq）；context。setNamespace（this。defaultMQProducer。getNamespace（））；StringisTransmsg。getProperty（MessageConst。PROPERTYTRANSACTIONPREPARED）；3。1通过isTrans来确定MsgType类型if（true。equals（isTrans））｛context。setMsgType（MessageType。TransMsgHalf）；｝3。2如果msg里面STARTDELIVERTIME或者DELAY不为空，则设置为延迟消息if（msg。getProperty（STARTDELIVERTIME）！nullmsg。getProperty（MessageConst。PROPERTYDELAYTIMELEVEL）！null）｛context。setMsgType（MessageType。DelayMsg）；｝3。3调用钩子函数里的方法this。executeSendMessageHookBefore（context）；｝
　　4。设置发送信息请求头SendMessageRequestHeader
　　最后根据默认发送方式，进行消息的发送
　　主要利用NettyRemotingClient进行发送，这里就先不展开来说了入口：MQClientAPIImpl。sendMessage（）问题答复消息发送者是如何做负载均衡的？默认采用轮询，每一个消息发送者全局会维护一个Topic上一次选择的队列，然后基于这个序号进行递增轮询AllocateMessageQueueAveragely平均分配，按照总数除以消费者个数进行，对每个消费者进行分配AllocateMessageQueueAveragelyByCircle轮流平均分配，按照消费者个数，进行轮询分配消息发送者是如何保证高可用的？在上面的步骤中通过TopicPublishInfo找到对应的MessageQueue下的，BrokerName信息，利用参数sendLatencyFaultEnable来开启关闭故障规避机制sendLatencyFaultEnable设置为true：开启延迟规避机制，一旦消息发送失败会将brokera悲观地认为在接下来的一段时间内该Broker不可用，在为未来某一段时间内所有的客户端不会向该Broker发送消息。使用本次消息发送延迟时间来计算Broker故障规避时长，不参与消息发送队列负载finalStringnotBestBrokerlatencyFaultTolerance。pickOneAtLeast（）；intwriteQueueNumstpInfo。getQueueIdByBroker（notBestBroker）；if（writeQueueNums0）｛finalMessageQueuemqtpInfo。selectOneMessageQueue（）；if（notBestBroker！null）｛mq。setBrokerName（notBestBroker）；mq。setQueueId（tpInfo。getSendWhichQueue（）。incrementAndGet（）writeQueueNums）；｝returnmq；｝
　　但是这样子做可能带来的后果是Broker没有可用的情况，或者是某个Broker数据激增，增加消费者的压力，所以默认不开启规避机制，遇到消息发送失败，规避brokera，但是在下一次消息发送时，即再次调用brokera。消息发送批量消息如何保证一致性的？将一个Topic下的消息，通过batch方法包一起发送客户端ID与使用陷阱
　　摘自丁威老师的文章
　　总结
　　这段时间主要学习了RocketMQ的消息发送，主要是以源码为主，深入了解了消息发送的启动和消息发送的流程，以及认识到客户端ID与使用陷阱一图总结
　　作者：叫我小郭
　　链接：https：juejin。cnpost7105315713157431332