基于RocketMQ 4.9.3，详细的介绍了Producer发送消息的总体流程的源码，包括生产者重试机制、生产者故障转移机制、VIP通道等知识都会一一介绍。
文章目录
下面是一个最简单的producer的使用案例：
publicclassProducer{publicstaticvoidmain(String[] args)throwsMQClientException,InterruptedException{/*
* Instantiate with a producer group name.
*/DefaultMQProducer producer =newDefaultMQProducer("please_rename_unique_group_name");
producer.setNamesrvAddr("127.0.0.1:9876");/*
* Launch the instance.
*/
producer.start();try{/*
* Create a message instance, specifying topic, tag and message body.
*/Message msg =newMessage("Topic1"/* Topic */,"TagA"/* Tag */,("Hello RocketMQ ").getBytes(RemotingHelper.DEFAULT_CHARSET)/* Message body */);/*
* Call send message to deliver message to one of brokers.
*/SendResult sendResult = producer.send(msg);System.out.printf("%s%n", sendResult);}catch(Exception e){
e.printStackTrace();Thread.sleep(1000);}//producer.shutdown();}}
可以看到producer通过调用send方法发送消息，实际上RocketMQ的producer发送消息的模式可以分为三种：
- 单向发送：把消息发向Broker服务器，而不用管消息是否成功发送到Broker服务器，只管发送，不管结果。
- 同步发送：把消息发送给Broker服务器，如果消息成功发送给Broker服务器，能得到Broker服务器的响应结果。
- 异步发送：把消息发送给Broker服务器，如果消息成功发送给Broker服务器，能得到Broker服务器的响应结果。因为是异步发送，发送完消息以后，不用等待，等到Broker服务器的响应调用回调。
DefaultMQProducer提供了更多的send的重载方法，来实现上面三种发送模式：
模式方法描述同步SendResult send(Collection msgs)同步批量发送消息SendResult send(Collection msgs, long timeout)同步批量发送消息SendResult send(Collection msgs, MessageQueue messageQueue)向指定的消息队列同步批量发送消息SendResult send(Collection msgs, MessageQueue messageQueue, long timeout)向指定的消息队列同步批量发送消息，并指定超时时间SendResult send(Message msg)同步单条发送消息SendResult send(Message msg, long timeout)同步发送单条消息，并指定超时时间SendResult send(Message msg, MessageQueue mq)向指定的消息队列同步发送单条消息SendResult send(Message msg, MessageQueue mq, long timeout)向指定的消息队列同步单条发送消息，并指定超时时间SendResult send(Message msg, MessageQueueSelector selector, Object arg)向消息队列同步单条发送消息，并指定发送队列选择器SendResult send(Message msg, MessageQueueSelector selector, Object arg, long timeout)向消息队列同步单条发送消息，并指定发送队列选择器与超时时间异步void send(Message msg, MessageQueueSelector selector, Object arg, SendCallback sendCallback)向指定的消息队列异步单条发送消息void send(Message msg, MessageQueueSelector selector, Object arg, SendCallback sendCallback, long timeout)向指定的消息队列异步单条发送消息，并指定超时时间void send(Message msg, SendCallback sendCallback)异步发送消息void send(Message msg, SendCallback sendCallback, long timeout)异步发送消息，并指定回调方法和超时时间void send(Message msg, MessageQueue mq, SendCallback sendCallback)向指定的消息队列异步单条发送消息，并指定回调方法void send(Message msg, MessageQueue mq, SendCallback sendCallback, long timeout)向指定的消息队列异步单条发送消息，并指定回调方法和超时时间单向void sendOneway(Message msg)单向发送消息，不等待broker响应void sendOneway(Message msg, MessageQueue mq)单向发送消息到指定队列，不等待broker响应void sendOneway(Message msg, MessageQueueSelector selector, Object arg)单向发送消息到队列选择器的选中的队列，不等待broker响应
上次我们分析了producer的启动流程源码，这次我们分析producer发送消息的源码。
1 send源码入口
DefaultMQProducer#send方法作为源码分析的入口方法，该方法被使用者直接调用。其内部调用defaultMQProducerImpl#send方法发送消息。
1.1 同步消息
publicSendResultsend(Message msg)throwsMQClientException,RemotingException,MQBrokerException,InterruptedException{//根据namespace设置topic
msg.setTopic(withNamespace(msg.getTopic()));//调用defaultMQProducerImpl#send发送消息returnthis.defaultMQProducerImpl.send(msg);}
该方法内部调用defaultMQProducerImpl#send发送消息。
publicSendResultsend(Message msg)throwsMQClientException,RemotingException,MQBrokerException,InterruptedException{//调用另一个send方法，设置超时时间参数，默认3000msreturnsend(msg,this.defaultMQProducer.getSendMsgTimeout());}
该方法内部又调用另一个send方法，设置超时时间参数，默认3000ms。
/**
* DefaultMQProducerImpl的方法
*
* @param msg 消息
* @param timeout 超时时间，毫秒值
*/publicSendResultsend(Message msg,long timeout)throwsMQClientException,RemotingException,MQBrokerException,InterruptedException{//调用另一个sendDefaultImpl方法，设置消息发送模式为SYNC，即同步；设置回调函数为nullreturnthis.sendDefaultImpl(msg,CommunicationMode.SYNC,null, timeout);}
该方法内部又调用另一个sendDefaultImpl方法，设置消息发送模式为SYNC，即同步；设置回调函数为null。
1.2 单向消息
单向消息使用sendOneway发送。
publicvoidsendOneway(Message msg)throwsMQClientException,RemotingException,InterruptedException{//根据namespace设置topic
msg.setTopic(withNamespace(msg.getTopic()));//调用defaultMQProducerImpl#sendOneway发送消息this.defaultMQProducerImpl.sendOneway(msg);}
该方法内部调用defaultMQProducerImpl#sendOneway。
publicvoidsendOneway(Message msg)throwsMQClientException,RemotingException,InterruptedException{try{//调用sendDefaultImpl方法，设置消息发送模式为ONEWAY，即单向；设置回调函数为null；设置超时时间参数，默认3000msthis.sendDefaultImpl(msg,CommunicationMode.ONEWAY,null,this.defaultMQProducer.getSendMsgTimeout());}catch(MQBrokerException e){thrownewMQClientException("unknown exception", e);}}
最终调用sendDefaultImpl方法，设置消息发送模式为ONEWAY，即单向；设置回调函数为null；设置超时时间参数，默认3000ms。
1.3 异步消息
异步消息使用带有callback函数的send方法发送。
publicvoidsend(Message msg,SendCallback sendCallback)throwsMQClientException,RemotingException,InterruptedException{//根据namespace设置topic
msg.setTopic(withNamespace(msg.getTopic()));//调用defaultMQProducerImpl#send发送消息，带有sendCallback参数this.defaultMQProducerImpl.send(msg, sendCallback);}
该方法内部调用defaultMQProducerImpl#send方法发送消息，带有sendCallback参数。
publicvoidsend(Message msg,SendCallback sendCallback)throwsMQClientException,RemotingException,InterruptedException{//该方法内部又调用另一个send方法，设置超时时间参数，默认3000ms。send(msg, sendCallback,this.defaultMQProducer.getSendMsgTimeout());}
该方法内部又调用另一个send方法，设置超时时间参数，默认3000ms。
publicvoidsend(finalMessage msg,finalSendCallback sendCallback,finallong timeout)throwsMQClientException,RemotingException,InterruptedException{//调用起始时间finallong beginStartTime =System.currentTimeMillis();//获取异步发送执行器线程池ExecutorService executor =this.getAsyncSenderExecutor();try{/*
* 使用线程池异步的执行sendDefaultImpl方法，即异步发送消息
*/
executor.submit(newRunnable(){@Overridepublicvoidrun(){/*
* 发送之前计算超时时间，如果超时则不发送，直接执行回调函数onException方法
*/long costTime =System.currentTimeMillis()- beginStartTime;if(timeout > costTime){try{//调用sendDefaultImpl方法执行发送操作sendDefaultImpl(msg,CommunicationMode.ASYNC, sendCallback, timeout - costTime);}catch(Exception e){//抛出异常，执行回调函数onException方法
sendCallback.onException(e);}}else{//超时，执行回调函数onException方法
sendCallback.onException(newRemotingTooMuchRequestException("DEFAULT ASYNC send call timeout"));}}});}catch(RejectedExecutionException e){thrownewMQClientException("executor rejected ", e);}}
该方法内部会获取获取异步发送执行器线程池，使用线程池异步的执行sendDefaultImpl方法，即异步发送消息。
发送之前计算超时时间，如果超时则不发送，直接执行回调函数onException方法。
2 sendDefaultImpl发送消息实现
该方法位于DefaultMQProducerImpl中，无论是同步消息、异步消息还是单向消息，最终都是调用该方法实现发送消息的逻辑的，因此该方法是真正的发送消息的方法入口。
该方法的大概步骤为：
- 调用makeSureStateOK方法，确定此producer的服务状态正常，如果服务状态不是RUNNING，那么抛出异常。
- 调用checkMessage方法，校验消息的合法性。
- 调用tryToFindTopicPublishInfo方法，尝试查找消息的一个topic路由，用以发送消息。
- 计算循环发送消息的总次数timesTotal，默认情况下，同步模式为3，即默认允许重试2次，可更改重试次数；其他模式为1，即不允许重试，不可更改。实际上异步发送消息也会重试，最多两次，只不过不是通过这里的逻辑重试的。
- 调用selectOneMessageQueue方法，选择一个消息队列MessageQueue，该犯法支持失败故障转移。
- 调用sendKernelImpl方法发送消息，异步、同步、单向发送消息的模式都是通过该方法实现的。
- 调用updateFaultItem方法，更新本地错误表缓存数据，用于延迟时间的故障转移的功能。
- 根据发送模式执行不同的处理，如果是异步或者单向模式则直接返回，如果是同步模式，如果开启了retryAnotherBrokerWhenNotStoreOK开关，那么如果返回值不是返回SEND_OK状态，则仍然会执行重试发送。
- 此过程中，如果抛出了RemotingException、MQClientException、以及部分MQBrokerException异常时，那么会进行重试，如果抛出了InterruptedException，或者因为超时则不再重试。
/**
* DefaultMQProducerImpl的方法
*
* @param msg 方法
* @param communicationMode 通信模式
* @param sendCallback 回调方法
* @param timeout 超时时间
*/privateSendResultsendDefaultImpl(Message msg,finalCommunicationMode communicationMode,finalSendCallback sendCallback,finallong timeout
)throwsMQClientException,RemotingException,MQBrokerException,InterruptedException{/*
* 1 确定此producer的服务状态正常，如果服务状态不是RUNNING，那么抛出异常
*/this.makeSureStateOK();/*
* 2 校验消息的合法性
*/Validators.checkMessage(msg,this.defaultMQProducer);//生成本次调用idfinallong invokeID = random.nextLong();//开始时间戳long beginTimestampFirst =System.currentTimeMillis();long beginTimestampPrev = beginTimestampFirst;//结束时间戳long endTimestamp = beginTimestampFirst;/*
* 3 尝试查找消息的一个topic路由，用以发送消息
*/TopicPublishInfo topicPublishInfo =this.tryToFindTopicPublishInfo(msg.getTopic());//找到有效的topic信息if(topicPublishInfo !=null&& topicPublishInfo.ok()){boolean callTimeout =false;MessageQueue mq =null;Exception exception =null;SendResult sendResult =null;/*
* 4 计算发送消息的总次数
* 同步模式为3，即默认允许重试2次，可更改重试次数；其他模式为1，即不允许重试，不可更改
*/int timesTotal = communicationMode ==CommunicationMode.SYNC ?1+this.defaultMQProducer.getRetryTimesWhenSendFailed():1;int times =0;//记录每一次重试时候发送消息目标Broker名字的数组String[] brokersSent =newString[timesTotal];/*
* 在循环中，发送消息，包含消息重试的逻辑，总次数默认不超过3
*/for(; times < timesTotal; times++){//上次使用过的broker，可以为空，表示第一次选择String lastBrokerName =null== mq ?null: mq.getBrokerName();/*
* 5 选择一个消息队列MessageQueue
*/MessageQueue mqSelected =this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);if(mqSelected !=null){
mq = mqSelected;//设置brokerName
brokersSent[times]= mq.getBrokerName();try{//调用的开始时间
beginTimestampPrev =System.currentTimeMillis();//如果还有可调用次数，那么if(times >0){//在重新发送期间用名称空间重置topic
msg.setTopic(this.defaultMQProducer.withNamespace(msg.getTopic()));}//现在调用的开始时间 减去 开始时间，判断时候在调用发起之前就超时了long costTime = beginTimestampPrev - beginTimestampFirst;//如果已经超时了，那么直接结束循环，不再发送//即超时的时候，即使还剩下重试次数，也不会再继续重试if(timeout < costTime){
callTimeout =true;break;}/*
* 6 异步、同步、单向发送消息
*/
sendResult =this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);//方法调用结束时间戳
endTimestamp =System.currentTimeMillis();/*
* 7 更新本地错误表缓存数据，用于延迟时间的故障转移的功能
*/this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev,false);/*
* 8 根据发送模式执行不同的处理
*/switch(communicationMode){//异步和单向模式直接返回nullcase ASYNC:returnnull;case ONEWAY:returnnull;case SYNC://同步模式，如果开启了retryAnotherBrokerWhenNotStoreOK开关，那么如果不是返回SEND_OK状态，则仍然会执行重试发送if(sendResult.getSendStatus()!=SendStatus.SEND_OK){if(this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()){continue;}}//如果发送成功，则返回return sendResult;default:break;}}catch(RemotingException e){//RemotingException异常，会执行重试
endTimestamp =System.currentTimeMillis();this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev,true);
log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
log.warn(msg.toString());
exception = e;continue;}catch(MQClientException e){//MQClientException异常，会执行重试
endTimestamp =System.currentTimeMillis();this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev,true);
log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
log.warn(msg.toString());
exception = e;continue;}catch(MQBrokerException e){//MQBrokerException异常
endTimestamp =System.currentTimeMillis();this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev,true);
log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
log.warn(msg.toString());
exception = e;//如果返回的状态码属于一下几种，则支持重试：//ResponseCode.TOPIC_NOT_EXIST,//ResponseCode.SERVICE_NOT_AVAILABLE,//ResponseCode.SYSTEM_ERROR,//ResponseCode.NO_PERMISSION,//ResponseCode.NO_BUYER_ID,//ResponseCode.NOT_IN_CURRENT_UNITif(this.defaultMQProducer.getRetryResponseCodes().contains(e.getResponseCode())){continue;}else{//其他状态码不支持重试，如果有结果则返回，否则直接抛出异常if(sendResult !=null){return sendResult;}throw e;}}catch(InterruptedException e){//InterruptedException异常，不会执行重试
endTimestamp =System.currentTimeMillis();this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev,false);
log.warn(String.format("sendKernelImpl exception, throw exception, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
log.warn(msg.toString());
log.warn("sendKernelImpl exception", e);
log.warn(msg.toString());throw e;}}else{break;}}/*
* 抛出异常的操作
*/if(sendResult !=null){return sendResult;}String info =String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s",
times,System.currentTimeMillis()- beginTimestampFirst,
msg.getTopic(),Arrays.toString(brokersSent));
info +=FAQUrl.suggestTodo(FAQUrl.SEND_MSG_FAILED);MQClientException mqClientException =newMQClientException(info, exception);if(callTimeout){thrownewRemotingTooMuchRequestException("sendDefaultImpl call timeout");}if(exception instanceofMQBrokerException){
mqClientException.setResponseCode(((MQBrokerException) exception).getResponseCode());}elseif(exception instanceofRemotingConnectException){
mqClientException.setResponseCode(ClientErrorCode.CONNECT_BROKER_EXCEPTION);}elseif(exception instanceofRemotingTimeoutException){
mqClientException.setResponseCode(ClientErrorCode.ACCESS_BROKER_TIMEOUT);}elseif(exception instanceofMQClientException){
mqClientException.setResponseCode(ClientErrorCode.BROKER_NOT_EXIST_EXCEPTION);}throw mqClientException;}validateNameServerSetting();thrownewMQClientException("No route info of this topic: "+ msg.getTopic()+FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO),null).setResponseCode(ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION);}
2.1 makeSureStateOK确定生产者服务状态
首先会确定此producer的服务状态正常，如果服务状态不是RUNNING，那么抛出异常。
/**
* DefaultMQProducerImpl的方法
*/privatevoidmakeSureStateOK()throwsMQClientException{//服务状态不是RUNNING，那么抛出MQClientException异常。if(this.serviceState !=ServiceState.RUNNING){thrownewMQClientException("The producer service state not OK, "+this.serviceState
+FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),null);}}
2.2 checkMessage校验消息的合法性
确定服务状态正常之后，还需要校验消息的合法性。校验规则为：
- 如果msg消息为null，抛出异常。
- 校验topic。如果topic为空，或者长度大于127个字符，或者topic的字符串不符合 "+$"模式，即包含非法字符，那么抛出异常。如果当前topic是不为允许使用的系统topic，那么抛出异常
- 校验消息体。如果消息体为null，或者为空数组，或者消息字节数组长度大于4,194,304，即消息的大小大于4M，那么抛出异常。
/**
* Validators的方法
*/publicstaticvoidcheckMessage(Message msg,DefaultMQProducer defaultMQProducer)throwsMQClientException{//如果消息为null，抛出异常if(null== msg){thrownewMQClientException(ResponseCode.MESSAGE_ILLEGAL,"the message is null");}/*
* 校验topic
*///如果topic为空，或者长度大于127个字符，或者topic的字符串不符合 "^[%|a-zA-Z0-9_-]+$"模式，即包含非法字符，那么抛出异常Validators.checkTopic(msg.getTopic());//如果当前topic是不为允许使用的系统topic SCHEDULE_TOPIC_XXXX，那么抛出异常Validators.isNotAllowedSendTopic(msg.getTopic());// body//如果消息体为null，那么抛出异常if(null== msg.getBody()){thrownewMQClientException(ResponseCode.MESSAGE_ILLEGAL,"the message body is null");}//如果消息体为空数组，那么抛出异常if(0== msg.getBody().length){thrownewMQClientException(ResponseCode.MESSAGE_ILLEGAL,"the message body length is zero");}//如果消息 字节数组长度大于4,194,304，即消息的大小大于4M，那么抛出异常if(msg.getBody().length > defaultMQProducer.getMaxMessageSize()){thrownewMQClientException(ResponseCode.MESSAGE_ILLEGAL,"the message body size over max value, MAX: "+ defaultMQProducer.getMaxMessageSize());}}publicstaticvoidcheckTopic(String topic)throwsMQClientException{//如果topic为空，那么抛出异常if(UtilAll.isBlank(topic)){thrownewMQClientException("The specified topic is blank",null);}//如果topic长度大于127个字符，那么抛出异常if(topic.length()> TOPIC_MAX_LENGTH){thrownewMQClientException(String.format("The specified topic is longer than topic max length %d.", TOPIC_MAX_LENGTH),null);}//如果topic字符串包含非法字符，那么抛出异常if(isTopicOrGroupIllegal(topic)){thrownewMQClientException(String.format("The specified topic[%s] contains illegal characters, allowing only %s", topic,"^[%|a-zA-Z0-9_-]+$"),null);}}
2.3 tryToFindTopicPublishInfo查找topic的发布信息
该方法用于查找指定topic的发布信息TopicPublishInfo。
/**
* DefaultMQProducerImpl的方法
* <p>
* 查找指定topic的推送信息
*/privateTopicPublishInfotryToFindTopicPublishInfo(finalString topic){//尝试直接从producer的topicPublishInfoTable中获取topic信息TopicPublishInfo topicPublishInfo =this.topicPublishInfoTable.get(topic);//如果没有获取到有效信息，if(null== topicPublishInfo ||!topicPublishInfo.ok()){//那么立即创建一个TopicPublishInfothis.topicPublishInfoTable.putIfAbsent(topic,newTopicPublishInfo());//立即从nameServer同步此topic的路由配置信息，并且更新本地缓存this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);//再次获取topicPublishInfo
topicPublishInfo =this.topicPublishInfoTable.get(topic);}//如果找到的路由信息是可用的，直接返回if(topicPublishInfo.isHaveTopicRouterInfo()|| topicPublishInfo.ok()){return topicPublishInfo;}else{//再次从nameServer同步topic的数据，不过这次使用默认的topic “TBW102”去找路由配置信息作为本topic参数信息this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic,true,this.defaultMQProducer);
topicPublishInfo =this.topicPublishInfoTable.get(topic);return topicPublishInfo;}}
首先在本地缓存topicPublishInfoTable获取，如果没有获取到有效数据，那么立即调用updateTopicRouteInfoFromNameServer方法从nameServer同步此topic的路由配置信息，并且更新本地缓存，如果还是没有获取到有效数据，那么再次从nameServer同步topic的数据，不过这次使用默认的topic “TBW102”去找路由配置信息作为本topic参数信息。
updateTopicRouteInfoFromNameServer 方法我们在此前的producer启动流程中已经介绍了。
TopicPublishInfo包含topic的各种属性：
/**
* 是否是顺序消息
*/privateboolean orderTopic =false;/**
* 是否包含路由信息
*/privateboolean haveTopicRouterInfo =false;/**
* topic的消息队列集合
*/privateList<MessageQueue> messageQueueList =newArrayList<MessageQueue>();/**
* 当前线程线程的消息队列的下标，循环选择消息队列使用+1
*/privatevolatileThreadLocalIndex sendWhichQueue =newThreadLocalIndex();/**
* topic路由信息，包括topic的队列信息queueDatas，topic的broker信息brokerDatas，顺序topic配置orderTopicConf，消费过滤信息filterServerTable等属性
*/privateTopicRouteData topicRouteData;
2.4 计算发送次数timesTotal
在发送消息之前，会先计算最大发送次数，同步模式为3，即默认允许重试2次，可更改重试次数；其他模式为1，即不允许重试，不可更改。
int timesTotal = communicationMode ==CommunicationMode.SYNC ?1+this.defaultMQProducer.getRetryTimesWhenSendFailed():1;
注意，异步发送同样有重试，并且也是两次，只不过它的重试不在这个循环里面，而是是在MQClientAPIImpl#sendMessage方法中，后面会讲到。
2.5 selectOneMessageQueue选择消息队列
selectOneMessageQueue方法用于查找一个可用的消息队列，该方法内部调用mqFaultStrategy#selectOneMessageQueue方法：
/**
* DefaultMQProducerImpl的方法
*
* 选择一个消息队列
* @param tpInfo topic信息
* @param lastBrokerName 上次使用过的broker
*/publicMessageQueueselectOneMessageQueue(finalTopicPublishInfo tpInfo,finalString lastBrokerName){//调用mqFaultStrategy#selectOneMessageQueue方法returnthis.mqFaultStrategy.selectOneMessageQueue(tpInfo, lastBrokerName);}
mqFaultStrategy#selectOneMessageQueue方法支持故障转移机制，其选择步骤为：
- 首先判断是否开启了发送延迟故障转移机制，即sendLatencyFaultEnable属性是否为true，默认false不打开。如果开启了该机制： 1. 首先仍然是遍历消息队列，按照轮询的方式选取一个消息队列，当消息队列可用（无故障）时，选择消息队列的工作就结束，否则循环选择其他队列。如果该mq的broker不存在LatencyFaultTolerance维护的faultItemTable集合属性中，或者当前时间戳已经大于该broker下一次开始可用的时间戳，表示无故障。2. 没有选出无故障的mq，那么从LatencyFaultTolerance维护的不是最好的broker集合faultItemTable中随机选择一个broker，随后判断如果写队列数大于0，那么选择该broker。然后遍历消息队列，采用取模的方式获取一个队列，即轮询的方式，重置其brokerName，queueId，进行消息发送。3. 如果上面的步骤抛出了异常，那么遍历消息队列，采用取模的方式获取一个队列，即轮询的方式。
- 如果没有发送延迟故障转移机制，那么那么遍历消息队列，即采用取模轮询的方式获取一个brokerName与lastBrokerName不相等的队列，即不会再次选择上次发送失败的broker。如果没有找到一个不同broker的mq，那么退回到轮询的方式。
selectOneMessageQueue方法选择mq的时候的故障转移机制，其目的就是为了保证每次发送消息尽量更快的成功，是一种保证高可用的手段。总的来说，包括两种故障转移：
- 一种是延迟时间的故障转移，这需要将sendLatencyFaultEnable属性中设置为true，默认false。对于请求响应较慢的broker，可以在一段时间内将其状态置为不可用，消息队列选择时，会过滤掉mq认为不可用的broker，以此来避免不断向宕机的broker发送消息，选取一个延迟较短的broker，实现消息发送高可用。
- 另一种是没有开启延迟时间的故障转移的时候，在轮询选择mq的时候，不会选择上次发送失败的broker，实现消息发送高可用。
/**
* MQFaultStrategy的方法
* <p>
* 选择一个消息队列，支持故障延迟转移
*
* @param tpInfo topic信息
* @param lastBrokerName 上次使用过的broker
*/publicMessageQueueselectOneMessageQueue(finalTopicPublishInfo tpInfo,finalString lastBrokerName){/*
* 判断是否开启了发送延迟故障转移机制，默认false不打开
* 如果开启了该机制，那么每次选取topic下对应的queue时，会基于之前执行的耗时，在有存在符合条件的broker的前提下，优选选取一个延迟较短的broker，否则再考虑随机选取。
*/if(this.sendLatencyFaultEnable){try{//当前线程线程的消息队列的下标，循环选择消息队列使用+1int index = tpInfo.getSendWhichQueue().incrementAndGet();//遍历消息队列，采用取模的方式获取一个队列，即轮询的方式for(int i =0; i < tpInfo.getMessageQueueList().size(); i++){//取模int pos =Math.abs(index++)% tpInfo.getMessageQueueList().size();if(pos <0)
pos =0;//获取该消息队列MessageQueue mq = tpInfo.getMessageQueueList().get(pos);//如果当前消息队列是可用的，即无故障，那么直接返回该mq//如果该broker不存在LatencyFaultTolerance维护的faultItemTable集合属性中，或者当前时间已经大于该broker下一次开始可用的时间点，表示无故障if(latencyFaultTolerance.isAvailable(mq.getBrokerName()))return mq;}//没有选出无故障的mq，那么一个不是最好的broker集合中随机选择一个finalString notBestBroker = latencyFaultTolerance.pickOneAtLeast();//如果写队列数大于0，那么选择该brokerint writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);if(writeQueueNums >0){//遍历消息队列，采用取模的方式获取一个队列，即轮询的方式finalMessageQueue mq = tpInfo.selectOneMessageQueue();if(notBestBroker !=null){//重置其brokerName，queueId，进行消息发送
mq.setBrokerName(notBestBroker);
mq.setQueueId(tpInfo.getSendWhichQueue().incrementAndGet()% writeQueueNums);}return mq;}else{//如果写队列数小于0，那么移除该broker
latencyFaultTolerance.remove(notBestBroker);}}catch(Exception e){
log.error("Error occurred when selecting message queue", e);}//如果上面的步骤抛出了异常，那么遍历消息队列，采用取模的方式获取一个队列，即轮询的方式return tpInfo.selectOneMessageQueue();}//如果没有发送延迟故障转移机制，那么那么遍历消息队列，即采用取模轮询的方式//获取一个brokerName与lastBrokerName不相等的队列，即不会再次选择上次发送失败的brokerreturn tpInfo.selectOneMessageQueue(lastBrokerName);}
2.5.1 selectOneMessageQueue选择一个mq
selectOneMessageQueue方法有两个重载方法，一个是有参数的，另一个是无参数的。
无参数的方法，即轮询选择一个mq，没有任何限制：
/**
* TopicPublishInfo的方法
* <p>
* 轮询的选择一个mq
*/publicMessageQueueselectOneMessageQueue(){//获取下一个indexint index =this.sendWhichQueue.incrementAndGet();//取模计算索引int pos =Math.abs(index)%this.messageQueueList.size();if(pos <0)
pos =0;//获取该索引的mqreturnthis.messageQueueList.get(pos);}
有参数的方法，其参数是上一次发送失败的brokerName，并且在选择的时候，不会选择上一次发送失败的brokerName的mq，即避免选择发送失败的broker继续发送。当然如果最后没有选出来，那么还是走轮询获取的逻辑。
/**
* TopicPublishInfo的方法
*
* @param lastBrokerName 上一次发送失败的brokerName
*/publicMessageQueueselectOneMessageQueue(finalString lastBrokerName){//如果lastBrokerName为null，即第一次发送，那么轮询选择一个if(lastBrokerName ==null){returnselectOneMessageQueue();}else{for(int i =0; i <this.messageQueueList.size(); i++){//轮询选择一个mqint index =this.sendWhichQueue.incrementAndGet();int pos =Math.abs(index)%this.messageQueueList.size();if(pos <0)
pos =0;MessageQueue mq =this.messageQueueList.get(pos);//如果mq的brokerName不等于lastBrokerName，就返回，否则选择下一个if(!mq.getBrokerName().equals(lastBrokerName)){return mq;}}//没有选出来，那么轮询选择一个returnselectOneMessageQueue();}}
2.6 sendKernelImpl发送消息
选择了消息队列之后，会调用sendKernelImpl方法进行消息的发送。该方法的大概步骤为：
- 首先调用findBrokerAddressInPublish方法从brokerAddrTable中查找Master broker地址。如果找不到，那么再次调用tryToFindTopicPublishInfo方法从nameServer远程拉取配置，并更新本地缓存，随后再次尝试获取Master broker地址。
- 调用brokerVIPChannel判断是否开启vip通道，如果开启了，那么将brokerAddr的port – 2，因为vip通道的端口为普通端口 – 2。
- 如果不是批量消息，那么设置唯一的uniqId。
- 如果不是批量消息，并且消息体大于4K，那么进行消息压缩。
- 如果存在CheckForbiddenHook，则执行checkForbidden钩子方法。如果存在SendMessageHook，则执行sendMessageBefore钩子方法。
- 设置请求头信息SendMessageRequestHeader，请求头包含各种基本属性，例如producerGroup、topic、queueId等，并且针对重试消息的处理，将消息重试次数和最大重试次数存入请求头中。
- 根据不同的发送模式发送消息。如果是异步发送模式，则需要先克隆并还原消息。最终异步、单向、同步模式都是调用MQClientAPIImpl#sendMessage方法发送消息的。
- 如果MQClientAPIImpl#sendMessage方法正常发送或者抛出RemotingException、MQBrokerException、InterruptedException异常，那么会判断如果存在SendMessageHook，则执行sendMessageAfter钩子方法。
- 在finally块中，对原始消息进行恢复。
/**
* DefaultMQProducerImpl的方法
* 发送消息
*
* @param msg 消息
* @param mq mq
* @param communicationMode 发送模式
* @param sendCallback 发送回调
* @param topicPublishInfo topic信息
* @param timeout 超时时间
* @return 发送结果
*/privateSendResultsendKernelImpl(finalMessage msg,finalMessageQueue mq,finalCommunicationMode communicationMode,finalSendCallback sendCallback,finalTopicPublishInfo topicPublishInfo,finallong timeout)throwsMQClientException,RemotingException,MQBrokerException,InterruptedException{//开始时间long beginStartTime =System.currentTimeMillis();/*
* 1 根据brokerName从brokerAddrTable中查找broker地址
*/String brokerAddr =this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());//如果本地找不到 broker 的地址if(null== brokerAddr){/*
* 2 从nameServer远程拉取配置，并更新本地缓存
* 该方法此前就学习过了
*/tryToFindTopicPublishInfo(mq.getTopic());//再次获取地址
brokerAddr =this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());}SendMessageContext context =null;if(brokerAddr !=null){/*
* 3 vip通道判断
*/
brokerAddr =MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);byte[] prevBody = msg.getBody();try{//for MessageBatch,ID has been set in the generating process/*
* 4 如果不是批量消息，那么尝试生成唯一uniqId，即UNIQ_KEY属性。MessageBatch批量消息在生成时就已经设置uniqId
* uniqId也被称为客户端生成的msgId，从逻辑上代表唯一一条消息
*/if(!(msg instanceofMessageBatch)){MessageClientIDSetter.setUniqID(msg);}/*
* 设置nameSpace为实例Id
*/boolean topicWithNamespace =false;if(null!=this.mQClientFactory.getClientConfig().getNamespace()){
msg.setInstanceId(this.mQClientFactory.getClientConfig().getNamespace());
topicWithNamespace =true;}//消息标识符int sysFlag =0;//消息压缩标识boolean msgBodyCompressed =false;/*
* 5 尝试压缩消息
*/if(this.tryToCompressMessage(msg)){
sysFlag |=MessageSysFlag.COMPRESSED_FLAG;
msgBodyCompressed =true;}//事务消息标志，prepare消息finalString tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);if(Boolean.parseBoolean(tranMsg)){
sysFlag |=MessageSysFlag.TRANSACTION_PREPARED_TYPE;}/*
* 6 如果存在CheckForbiddenHook，则执行checkForbidden方法
* 为什么叫禁止钩子呢，可能是想要使用者将不可发送消息的检查放在这个钩子函数里面吧（猜测）
*/if(hasCheckForbiddenHook()){CheckForbiddenContext checkForbiddenContext =newCheckForbiddenContext();
checkForbiddenContext.setNameSrvAddr(this.defaultMQProducer.getNamesrvAddr());
checkForbiddenContext.setGroup(this.defaultMQProducer.getProducerGroup());
checkForbiddenContext.setCommunicationMode(communicationMode);
checkForbiddenContext.setBrokerAddr(brokerAddr);
checkForbiddenContext.setMessage(msg);
checkForbiddenContext.setMq(mq);
checkForbiddenContext.setUnitMode(this.isUnitMode());this.executeCheckForbiddenHook(checkForbiddenContext);}/*
* 7 如果存在SendMessageHook，则执行sendMessageBefore方法
*/if(this.hasSendMessageHook()){
context =newSendMessageContext();
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());String isTrans = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);if(isTrans !=null&& isTrans.equals("true")){
context.setMsgType(MessageType.Trans_Msg_Half);}if(msg.getProperty("__STARTDELIVERTIME")!=null|| msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL)!=null){
context.setMsgType(MessageType.Delay_Msg);}this.executeSendMessageHookBefore(context);}/*
* 8 设置请求头信息
*/SendMessageRequestHeader requestHeader =newSendMessageRequestHeader();
requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
requestHeader.setTopic(msg.getTopic());
requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
requestHeader.setQueueId(mq.getQueueId());
requestHeader.setSysFlag(sysFlag);
requestHeader.setBornTimestamp(System.currentTimeMillis());
requestHeader.setFlag(msg.getFlag());
requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
requestHeader.setReconsumeTimes(0);
requestHeader.setUnitMode(this.isUnitMode());
requestHeader.setBatch(msg instanceofMessageBatch);//针对重试消息的处理if(requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)){//获取消息重新消费次数属性值String reconsumeTimes =MessageAccessor.getReconsumeTime(msg);if(reconsumeTimes !=null){//将重新消费次数设置到请求头中，并且清除该属性
requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));MessageAccessor.clearProperty(msg,MessageConst.PROPERTY_RECONSUME_TIME);}//获取消息的最大重试次数属性值String maxReconsumeTimes =MessageAccessor.getMaxReconsumeTimes(msg);if(maxReconsumeTimes !=null){//将最大重新消费次数设置到请求头中，并且清除该属性
requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));MessageAccessor.clearProperty(msg,MessageConst.PROPERTY_MAX_RECONSUME_TIMES);}}/*
* 9 根据不同的发送模式，发送消息
*/SendResult sendResult =null;switch(communicationMode){/*
* 异步发送模式
*/case ASYNC:/*
* 首先克隆并还原消息
*
* 该方法的finally中已经有还原消息的代码了，为什么在异步发送消息之前，还要先还原消息呢？
*
* 因为异步发送时 finally 重新赋值的时机并不确定，有很大概率是在第一次发送结束前就完成了 finally 中的赋值，
* 因此在内部重试前 msg.body 大概率已经被重新赋值过，而 onExceptionImpl 中的重试逻辑 MQClientAPIImpl.sendMessageAsync 不会再对数据进行压缩，
* 简言之，在异步发送的情况下，如果调用 onExceptionImpl 内部的重试，有很大概率发送的是无压缩的数据
*/Message tmpMessage = msg;boolean messageCloned =false;//如果开启了消息压缩if(msgBodyCompressed){//If msg body was compressed, msgbody should be reset using prevBody.//Clone new message using commpressed message body and recover origin massage.//Fix bug:https://github.com/apache/rocketmq-externals/issues/66//克隆一个message
tmpMessage =MessageAccessor.cloneMessage(msg);
messageCloned =true;//恢复原来的消息体
msg.setBody(prevBody);}//如果topic整合了namespaceif(topicWithNamespace){if(!messageCloned){
tmpMessage =MessageAccessor.cloneMessage(msg);
messageCloned =true;}//还原topic
msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(),this.defaultMQProducer.getNamespace()));}/*
* 发送消息之前，进行超时检查，如果已经超时了那么取消本次发送操作，抛出异常
*/long costTimeAsync =System.currentTimeMillis()- beginStartTime;if(timeout < costTimeAsync){thrownewRemotingTooMuchRequestException("sendKernelImpl call timeout");}/*
* 10 发送异步消息
*/
sendResult =this.mQClientFactory.getMQClientAPIImpl().sendMessage(
brokerAddr,
mq.getBrokerName(),
tmpMessage,
requestHeader,
timeout - costTimeAsync,
communicationMode,
sendCallback,
topicPublishInfo,this.mQClientFactory,this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(),
context,this);break;/*
* 单向、同步发送模式
*/case ONEWAY:case SYNC:/*
* 发送消息之前，进行超时检查，如果已经超时了那么取消本次发送操作，抛出异常
*/long costTimeSync =System.currentTimeMillis()- beginStartTime;if(timeout < costTimeSync){thrownewRemotingTooMuchRequestException("sendKernelImpl call timeout");}/*
* 10 发送单向、同步消息
*/
sendResult =this.mQClientFactory.getMQClientAPIImpl().sendMessage(
brokerAddr,
mq.getBrokerName(),
msg,
requestHeader,
timeout - costTimeSync,
communicationMode,
context,this);break;default:assertfalse;break;}/*
* 9 如果存在SendMessageHook，则执行sendMessageAfter方法
*/if(this.hasSendMessageHook()){
context.setSendResult(sendResult);this.executeSendMessageHookAfter(context);}//返回执行结果return sendResult;//如果抛出了异常，如果存在SendMessageHook，则执行sendMessageAfter方法}catch(RemotingException e){if(this.hasSendMessageHook()){
context.setException(e);this.executeSendMessageHookAfter(context);}throw e;}catch(MQBrokerException e){if(this.hasSendMessageHook()){
context.setException(e);this.executeSendMessageHookAfter(context);}throw e;}catch(InterruptedException e){if(this.hasSendMessageHook()){
context.setException(e);this.executeSendMessageHookAfter(context);}throw e;}finally{/*
* 对消息进行恢复
* 1、因为客户端可能还需要查看原始的消息内容，如果是压缩消息，则无法查看
* 2、另外如果第一次压缩后消息还是大于4k，如果不恢复消息，那么客户端使用该message重新发送的时候，还会进行一次消息压缩
*/
msg.setBody(prevBody);
msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(),this.defaultMQProducer.getNamespace()));}}thrownewMQClientException("The broker["+ mq.getBrokerName()+"] not exist",null);}
2.6.1 findBrokerAddressInPublish查找broker地址
首先会根据brokerName从brokerAddrTable中查找broker地址。生产者只会向Master节点发送消息，因此只会返回Master节点的地址。
/**
* MQClientInstance的方法
*/publicStringfindBrokerAddressInPublish(finalString brokerName){//查询brokerAddrTable缓存的数据HashMap<Long/* brokerId */,String/* address */> map =this.brokerAddrTable.get(brokerName);//返回Mater节点的地址if(map !=null&&!map.isEmpty()){return map.get(MixAll.MASTER_ID);}returnnull;}
2.6.2 brokerVIPChannel判断vip通道
获取到brokerAddr之后，需要判断是否开启vip通道，如果开启了，那么将brokerAddr的port – 2，因为vip通道的端口为普通通道端口– 2。
/**
* MixAll的方法
*/publicstaticStringbrokerVIPChannel(finalboolean isChange,finalString brokerAddr){//如果开启了vip通道if(isChange){int split = brokerAddr.lastIndexOf(":");String ip = brokerAddr.substring(0, split);String port = brokerAddr.substring(split +1);//重新拼接brokerAddr，其中port - 2String brokerAddrNew = ip +":"+(Integer.parseInt(port)-2);return brokerAddrNew;}else{//如果没有开启vip通道，那么返回原地址return brokerAddr;}}
消费者拉取消息只能请求普通通道，但是生产者发送消息可以选择vip通道或者普通通道。
为什么要开启两个端口监听客户端请求呢？答案是隔离读写操作。在消息的API中，最重要的是发送消息，需要高RTT。如果普通端口的请求繁忙，会使得netty的IO线程阻塞，例如消息堆积的时候，消费消息的请求会填满IO线程池，导致写操作被阻塞。在这种情况下，我们可以向VIP频道发送消息，以保证发送消息的RTT。
但是，请注意，在rocketmq 4.5.1版本之后，客户端发送消息的请求选择VIP通道的配置被改为false，想要手动默认开启需要配置com.rocketmq.sendMessageWithVIPChannel属性。或者在创建producer的时候调用producer.setVipChannelEnabled()方法更改当前producer的配置。
因此，现在发送消息和消费消息实际上默认都走10911端口了，无需再关心10909端口的问题了。
2.6.3 setUniqID生成uniqId
该方法用于设置单条消息在客户端的uniqId，即设置到UNIQ_KEY属性中，批量消息在生成时就已经设置uniqId。
uniqId也被称为msgId，从逻辑上代表客户端生成的唯一一条消息，更多见此文章，uniqId生成规则。
/**
* MessageClientIDSetter的方法
*/publicstaticvoidsetUniqID(finalMessage msg){//如果这条消息不存在"UNIQ_KEY"属性，那么创建uniqId并且存入"UNIQ_KEY"属性中if(msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX)==null){
msg.putProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX,createUniqID());}}
2.6.4 tryToCompressMessage压缩消息
在发送单条消息的时候，会判断如果消息体超过4K，那么会进行消息压缩，压缩比默认为5，压缩完毕之后设置压缩标志，批量消息不支持压缩。消息压缩有利于更快的进行网络数据传输。
/**
* DefaultMQProducerImpl的方法
*/privatebooleantryToCompressMessage(finalMessage msg){//如果是批量消息，那么不进行压缩if(msg instanceofMessageBatch){//batch dose not support compressing right nowreturnfalse;}byte[] body = msg.getBody();if(body !=null){//如果消息长度大于4Kif(body.length >=this.defaultMQProducer.getCompressMsgBodyOverHowmuch()){try{//进行压缩，使用的JDK自带的压缩类byte[] data =UtilAll.compress(body, zipCompressLevel);if(data !=null){//重新设置到body中
msg.setBody(data);returntrue;}}catch(IOException e){
log.error("tryToCompressMessage exception", e);
log.warn(msg.toString());}}}returnfalse;}
2.7 updateFaultItem更新故障表
再发送消息完毕之后，无论是正常还是异常状态，都需要调用updateFaultItem方法，更新本地错误表缓存数据，用于延迟时间的故障转移的功能。
故障转移功能在此前的selectOneMessageQueue方法中被使用到，用于查找一个可用的消息队列。updateFaultItem方法在判断开启了故障转移之后，会更新LatencyFaultTolerance维护的faultItemTable集合属性中的异常broker数据。
/**
* DefaultMQProducerImpl的方法
* @param brokerName brokerName
* @param currentLatency 当前延迟
* @param isolation 是否使用默认隔离
*/publicvoidupdateFaultItem(finalString brokerName,finallong currentLatency,boolean isolation){//调用MQFaultStrategy#updateFaultItem方法this.mqFaultStrategy.updateFaultItem(brokerName, currentLatency, isolation);}
看看MQFaultStrategy#updateFaultItem方法。其根据本次发送消息的延迟时间currentLatency，会去计算出该broker的隔离时间duration，即可以计算出该broker的下一个可用时间点。然后更新故障记录表。
/**
* MQFaultStrategy的方法
*
* @param brokerName brokerName
* @param currentLatency 当前延迟
* @param isolation 是否使用默认隔离时间
*/publicvoidupdateFaultItem(finalString brokerName,finallong currentLatency,boolean isolation){//如果开启了故障转移，即sendLatencyFaultEnable为true，默认falseif(this.sendLatencyFaultEnable){//根据消息当前延迟currentLatency计算当前broker的故障延迟的时间duration//如果isolation为true，则使用默认隔离时间30000，即30slong duration =computeNotAvailableDuration(isolation ?30000: currentLatency);//更新故障记录表this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);}}
2.7.1 computeNotAvailableDuration计算隔离时间
computeNotAvailableDuration方法根据本次发送消息的延迟时间currentLatency，会去计算出该broker的隔离时间duration，或者说不可以用时间段，据此即可以计算出该broker的下一个可用时间点。
latencyMax延迟等级和notAvailableDuration隔离时间的对应关系如下：
| latencyMax，
Producer发送消息消耗时长notAvailableDuration，Broker不可用时长50L0L100L0L550L30000L1000L60000L2000L120000L3000L180000L15000L600000L
如果使用默认隔离时间30000，那个实际将会被隔离600000L，即10分钟。当抛出异常的时候，通常会设置isolation，即使用默认隔离时间。并且从这个表可以看出来，发送消息延迟越大，那么被设置的隔离时间也就越大。
//延迟等级privatelong[] latencyMax ={50L,100L,550L,1000L,2000L,3000L,15000L};//不可用时间等级privatelong[] notAvailableDuration ={0L,0L,30000L,60000L,120000L,180000L,600000L};/**
* MQFaultStrategy的方法
*
* @param currentLatency 当前延迟
* @return 故障延迟的时间
*/privatelongcomputeNotAvailableDuration(finallong currentLatency){//倒叙遍历latencyMaxfor(int i = latencyMax.length -1; i >=0; i--){//选择broker延迟时间对应的broker不可用时间，默认30000对应的故障延迟的时间为600000，即10分钟if(currentLatency >= latencyMax[i])returnthis.notAvailableDuration[i];}return0;}
2.7.2 updateFaultItem更新故障表
该方法更新LatencyFaultToleranceImpl维护的faultItemTable集合属性中的异常broker的故障信息，将会设置发送消息的延迟时间currentLatency属性，以及下一个可用时间点LatencyFaultToleranceImpl属性。
下次可用时间LatencyFaultToleranceImpl属性= 现在的时间 + 隔离的时间，在selectOneMessageQueue方法选取消息队列的时候，如果开启了集群故障转移，那么会查找下一个可用时间点小于当前时间点的broker的队列来发送消息。
/**
* LatencyFaultToleranceImpl的方法
*
* @param name brokerName
* @param currentLatency 当前延迟
* @param notAvailableDuration 隔离时间（不可用时间）
*/@OverridepublicvoidupdateFaultItem(finalString name,finallong currentLatency,finallong notAvailableDuration){//获取该broker此前的故障记录数据FaultItem old =this.faultItemTable.get(name);//如果此前没有数据，那么设置一个新对象肌凝乳if(null== old){finalFaultItem faultItem =newFaultItem(name);//设置当前延迟
faultItem.setCurrentLatency(currentLatency);//设置下一次可用时间点
faultItem.setStartTimestamp(System.currentTimeMillis()+ notAvailableDuration);//已有故障记录，更新
old =this.faultItemTable.putIfAbsent(name, faultItem);if(old !=null){
old.setCurrentLatency(currentLatency);
old.setStartTimestamp(System.currentTimeMillis()+ notAvailableDuration);}}else{//已有故障记录，更新
old.setCurrentLatency(currentLatency);
old.setStartTimestamp(System.currentTimeMillis()+ notAvailableDuration);}}
3 总结
本次我们学习了Producer的发送消息的源码总体流程，对于具体的发送消息的sendMessage方法源码将在下文讲解。
从这些源码中，我们得知了一些常见的却容易混淆的概念和知识，例如：
- 生产者消息重试：RocketMQ的消费者消息重试和生产者消息重投。
- 生产者故障转移。通过sendLatencyFaultEnable属性配置是否开启，默认未开启。故障转移机制，其目的就是为了保证每次发送消息尽量更快的成功，是一种保证高可用的手段。总的来说，包括两种故障转移： 1. 一种是延迟时间的故障转移，这需要将sendLatencyFaultEnable属性中设置为true，默认false。对于请求响应较慢的broker，可以在一段时间内将其状态置为不可用，消息队列选择时，会过滤掉mq认为不可用的broker，以此来避免不断向宕机的broker发送消息，选取一个延迟较短的broker，实现消息发送高可用。2. 另一种是没有开启延迟时间的故障转移的时候，在轮询选择mq的时候，不会选择上次发送失败的broker，实现消息发送高可用。
- Vip通道。VIP通道用于隔离读写操作。消费者拉取消息只能请求普通通道，但是生产者发送消息可以选择vip通道或者普通通道。 1. 在消息的API中，最重要的是发送消息，需要高RTT。如果普通端口的请求繁忙，会使得netty的IO线程阻塞，例如消息堆积的时候，消费消息的请求会填满IO线程池，导致写操作被阻塞。在这种情况下，我们可以向VIP频道发送消息，以保证发送消息的RTT。2. 但是，请注意，在rocketmq 4.5.1版本之后，客户端发送消息的请求选择VIP通道的配置被改为false，想要手动默认开启需要配置com.rocketmq.sendMessageWithVIPChannel属性。或者在创建producer的时候调用producer.setVipChannelEnabled()方法更改当前producer的配置。
- 故障转移表，RocketMQ的Producer生产者故障转移依赖于故障转移表实现，他是一个HasmMap。消息发送结束之后，会根据本次发送消息的延迟时间currentLatency，会去计算出该broker对应的的隔离时间duration，即可以计算出该broker的下一个可用时间点，然后更新故障记录表。故障转移表的key为brokerName，value为未来该broker可用时间。
本文转载自: https://blog.csdn.net/weixin_43767015/article/details/126729959
版权归原作者 刘Java 所有， 如有侵权，请联系我们删除。
版权归原作者 刘Java 所有， 如有侵权，请联系我们删除。