rocket mq 源码分析-Message 发送与接收

迷南。 2022-06-08 06:47 167阅读 0赞

*  [1、概述][1]
 *  [2、Producer 发送消息][1]
    
     *  [DefaultMQProducer\#send(Message)][1]
     *  [DefaultMQProducerImpl\#sendDefaultImpl()][1]
        
         *  [DefaultMQProducerImpl\#tryToFindTopicPublishInfo()][1]
         *  [MQFaultStrategy][1]
            
             *  [MQFaultStrategy][1]
             *  [LatencyFaultTolerance][1]
             *  [LatencyFaultToleranceImpl][1]
             *  [FaultItem][1]
         *  [DefaultMQProducerImpl\#sendKernelImpl()][1]
 *  [3、Broker 接收消息][1]
    
     *  [SendMessageProcessor\#sendMessage][1]
        
         *  [AbstractSendMessageProcessor\#msgCheck][1]
     *  [DefaultMessageStore\#putMessage][1]
 *  [4、某种结尾][1]

# 1、概述 #

1.  `Producer` 发送消息。主要是同步发送消息源码，涉及到 异步/Oneway发送消息，事务消息会跳过。
2.  `Broker` 接收消息。(存储消息在[《RocketMQ 源码分析 —— Message 存储》][RocketMQ _ _ Message]解析)

> ![Producer发送消息全局顺序图][Producer]

# 2、Producer 发送消息 #

> ![Producer发送消息顺序图][Producer 1]

## DefaultMQProducer\#send(Message) ##

*  说明：发送同步消息，`DefaultMQProducer#send(Message)` 对 `DefaultMQProducerImpl#send(Message)` 进行封装。

## DefaultMQProducerImpl\#sendDefaultImpl() ##

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style=""> <pre style="">
     
      
     <div style="">  
      
       
      <span style="">1</span>: 
      
       
      <span style=""><span style="">public</span> SendResult <span style="">send</span><span style="">(Message msg)</span> <span style="">throws</span> MQClientException, RemotingException, MQBrokerException, InterruptedException </span>{
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">2</span>:     
      
       
      <span style="">return</span> send(msg, 
      
       
      <span style="">this</span>.defaultMQProducer.getSendMsgTimeout());
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">3</span>: }
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">4</span>: 
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">5</span>: 
      
       
      <span style=""><span style="">public</span> SendResult <span style="">send</span><span style="">(Message msg, <span style="">long</span> timeout)</span> <span style="">throws</span> MQClientException, RemotingException, MQBrokerException, InterruptedException </span>{
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">6</span>:     
      
       
      <span style="">return</span> 
      
       
      <span style="">this</span>.sendDefaultImpl(msg, CommunicationMode.SYNC, 
      
       
      <span style="">null</span>, timeout);
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">7</span>: }
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">8</span>: 
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">9</span>: 
      
       
      <span style=""><span style="">private</span> SendResult <span style="">sendDefaultImpl</span><span style="">(//</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">10</span>: Message msg, //</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">11</span>: <span style="">final</span> CommunicationMode communicationMode, //</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">12</span>: <span style="">final</span> SendCallback sendCallback, //</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">13</span>: <span style="">final</span> <span style="">long</span> timeout//</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">14</span>: )</span> <span style="">throws</span> MQClientException, RemotingException, MQBrokerException, InterruptedException </span>{
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">15</span>:     
      
       
      <span style="">// 校验 Producer 处于运行状态</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">16</span>:     
      
       
      <span style="">this</span>.makeSureStateOK();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">17</span>:     
      
       
      <span style="">// 校验消息格式</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">18</span>:     Validators.checkMessage(msg, 
      
       
      <span style="">this</span>.defaultMQProducer);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">19</span>:     
      
       
      <span style="">//</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">20</span>:     
      
       
      <span style="">final</span> 
      
       
      <span style="">long</span> invokeID = random.nextLong(); 
      
       
      <span style="">// 调用编号；用于下面打印日志，标记为同一次发送消息</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">21</span>:     
      
       
      <span style="">long</span> beginTimestampFirst = System.currentTimeMillis();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">22</span>:     
      
       
      <span style="">long</span> beginTimestampPrev = beginTimestampFirst;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">23</span>:     
      
       
      <span style="">long</span> endTimestamp = beginTimestampFirst;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">24</span>:     
      
       
      <span style="">// 获取 Topic路由信息</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">25</span>:     TopicPublishInfo topicPublishInfo = 
      
       
      <span style="">this</span>.tryToFindTopicPublishInfo(msg.getTopic());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">26</span>:     
      
       
      <span style="">if</span> (topicPublishInfo != 
      
       
      <span style="">null</span> &amp;&amp; topicPublishInfo.ok()) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">27</span>:         MessageQueue mq = 
      
       
      <span style="">null</span>; 
      
       
      <span style="">// 最后选择消息要发送到的队列</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">28</span>:         Exception exception = 
      
       
      <span style="">null</span>;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">29</span>:         SendResult sendResult = 
      
       
      <span style="">null</span>; 
      
       
      <span style="">// 最后一次发送结果</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">30</span>:         
      
       
      <span style="">int</span> timesTotal = communicationMode == CommunicationMode.SYNC ? 
      
       
      <span style="">1</span> + 
      
       
      <span style="">this</span>.defaultMQProducer.getRetryTimesWhenSendFailed() : 
      
       
      <span style="">1</span>; 
      
       
      <span style="">// 同步多次调用</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">31</span>:         
      
       
      <span style="">int</span> times = 
      
       
      <span style="">0</span>; 
      
       
      <span style="">// 第几次发送</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">32</span>:         String[] brokersSent = 
      
       
      <span style="">new</span> String[timesTotal]; 
      
       
      <span style="">// 存储每次发送消息选择的broker名</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">33</span>:         
      
       
      <span style="">// 循环调用发送消息，直到成功</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">34</span>:         
      
       
      <span style="">for</span> (; times &lt; timesTotal; times++) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">35</span>:             String lastBrokerName = 
      
       
      <span style="">null</span> == mq ? 
      
       
      <span style="">null</span> : mq.getBrokerName();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">36</span>:             MessageQueue tmpmq = 
      
       
      <span style="">this</span>.selectOneMessageQueue(topicPublishInfo, lastBrokerName); 
      
       
      <span style="">// 选择消息要发送到的队列</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">37</span>:             
      
       
      <span style="">if</span> (tmpmq != 
      
       
      <span style="">null</span>) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">38</span>:                 mq = tmpmq;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">39</span>:                 brokersSent[times] = mq.getBrokerName();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">40</span>:                 
      
       
      <span style="">try</span> {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">41</span>:                     beginTimestampPrev = System.currentTimeMillis();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">42</span>:                     
      
       
      <span style="">// 调用发送消息核心方法</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">43</span>:                     sendResult = 
      
       
      <span style="">this</span>.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">44</span>:                     endTimestamp = System.currentTimeMillis();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">45</span>:                     
      
       
      <span style="">// 更新Broker可用性信息</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">46</span>:                     
      
       
      <span style="">this</span>.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, 
      
       
      <span style="">false</span>);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">47</span>:                     
      
       
      <span style="">switch</span> (communicationMode) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">48</span>:                         
      
       
      <span style="">case</span> ASYNC:
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">49</span>:                             
      
       
      <span style="">return</span> 
      
       
      <span style="">null</span>;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">50</span>:                         
      
       
      <span style="">case</span> ONEWAY:
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">51</span>:                             
      
       
      <span style="">return</span> 
      
       
      <span style="">null</span>;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">52</span>:                         
      
       
      <span style="">case</span> SYNC:
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">53</span>:                             
      
       
      <span style="">if</span> (sendResult.getSendStatus() != SendStatus.SEND_OK) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">54</span>:                                 
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) { 
      
       
      <span style="">// 同步发送成功但存储有问题时 &amp;&amp; 配置存储异常时重新发送开关 时，进行重试</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">55</span>:                                     
      
       
      <span style="">continue</span>;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">56</span>:                                 }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">57</span>:                             }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">58</span>:                             
      
       
      <span style="">return</span> sendResult;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">59</span>:                         
      
       
      <span style="">default</span>:
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">60</span>:                             
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">61</span>:                     }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">62</span>:                 } 
      
       
      <span style="">catch</span> (RemotingException e) { 
      
       
      <span style="">// 打印异常，更新Broker可用性信息，更新继续循环</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">63</span>:                     endTimestamp = System.currentTimeMillis();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">64</span>:                     
      
       
      <span style="">this</span>.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, 
      
       
      <span style="">true</span>);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">65</span>:                     log.warn(String.format(
      
       
      <span style="">"sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s"</span>, invokeID, endTimestamp - beginTimestampPrev, mq), e);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">66</span>:                     log.warn(msg.toString());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">67</span>:                     exception = e;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">68</span>:                     
      
       
      <span style="">continue</span>;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">69</span>:                 } 
      
       
      <span style="">catch</span> (MQClientException e) { 
      
       
      <span style="">// 打印异常，更新Broker可用性信息，继续循环</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">70</span>:                     endTimestamp = System.currentTimeMillis();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">71</span>:                     
      
       
      <span style="">this</span>.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, 
      
       
      <span style="">true</span>);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">72</span>:                     log.warn(String.format(
      
       
      <span style="">"sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s"</span>, invokeID, endTimestamp - beginTimestampPrev, mq), e);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">73</span>:                     log.warn(msg.toString());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">74</span>:                     exception = e;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">75</span>:                     
      
       
      <span style="">continue</span>;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">76</span>:                 } 
      
       
      <span style="">catch</span> (MQBrokerException e) { 
      
       
      <span style="">// 打印异常，更新Broker可用性信息，部分情况下的异常，直接返回，结束循环</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">77</span>:                     endTimestamp = System.currentTimeMillis();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">78</span>:                     
      
       
      <span style="">this</span>.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, 
      
       
      <span style="">true</span>);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">79</span>:                     log.warn(String.format(
      
       
      <span style="">"sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s"</span>, invokeID, endTimestamp - beginTimestampPrev, mq), e);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">80</span>:                     log.warn(msg.toString());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">81</span>:                     exception = e;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">82</span>:                     
      
       
      <span style="">switch</span> (e.getResponseCode()) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">83</span>:                         
      
       
      <span style="">// 如下异常continue，进行发送消息重试</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">84</span>:                         
      
       
      <span style="">case</span> ResponseCode.TOPIC_NOT_EXIST:
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">85</span>:                         
      
       
      <span style="">case</span> ResponseCode.SERVICE_NOT_AVAILABLE:
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">86</span>:                         
      
       
      <span style="">case</span> ResponseCode.SYSTEM_ERROR:
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">87</span>:                         
      
       
      <span style="">case</span> ResponseCode.NO_PERMISSION:
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">88</span>:                         
      
       
      <span style="">case</span> ResponseCode.NO_BUYER_ID:
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">89</span>:                         
      
       
      <span style="">case</span> ResponseCode.NOT_IN_CURRENT_UNIT:
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">90</span>:                             
      
       
      <span style="">continue</span>;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">91</span>:                         
      
       
      <span style="">// 如果有发送结果，进行返回，否则，抛出异常；</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">92</span>:                         
      
       
      <span style="">default</span>:
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">93</span>:                             
      
       
      <span style="">if</span> (sendResult != 
      
       
      <span style="">null</span>) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">94</span>:                                 
      
       
      <span style="">return</span> sendResult;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">95</span>:                             }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">96</span>:                             
      
       
      <span style="">throw</span> e;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">97</span>:                     }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">98</span>:                 } 
      
       
      <span style="">catch</span> (InterruptedException e) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">99</span>:                     endTimestamp = System.currentTimeMillis();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">100</span>:                     
      
       
      <span style="">this</span>.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, 
      
       
      <span style="">false</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">101</span>:                     log.warn(String.format(
      
       
      <span style="">"sendKernelImpl exception, throw exception, InvokeID: %s, RT: %sms, Broker: %s"</span>, invokeID, endTimestamp - beginTimestampPrev, mq), e);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">102</span>:                     log.warn(msg.toString());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">103</span>:                     
      
       
      <span style="">throw</span> e;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">104</span>:                 }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">105</span>:             } 
      
       
      <span style="">else</span> {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">106</span>:                 
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">107</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">108</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">109</span>:         
      
       
      <span style="">// 返回发送结果</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">110</span>:         
      
       
      <span style="">if</span> (sendResult != 
      
       
      <span style="">null</span>) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">111</span>:             
      
       
      <span style="">return</span> sendResult;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">112</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">113</span>:         
      
       
      <span style="">// 根据不同情况，抛出不同的异常</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">114</span>:         String info = String.format(
      
       
      <span style="">"Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s"</span>, times, System.currentTimeMillis() - beginTimestampFirst,
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">115</span>:                 msg.getTopic(), Arrays.toString(brokersSent)) + FAQUrl.suggestTodo(FAQUrl.SEND_MSG_FAILED);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">116</span>:         MQClientException mqClientException = 
      
       
      <span style="">new</span> MQClientException(info, exception);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">117</span>:         
      
       
      <span style="">if</span> (exception 
      
       
      <span style="">instanceof</span> MQBrokerException) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">118</span>:             mqClientException.setResponseCode(((MQBrokerException) exception).getResponseCode());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">119</span>:         } 
      
       
      <span style="">else</span> 
      
       
      <span style="">if</span> (exception 
      
       
      <span style="">instanceof</span> RemotingConnectException) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">120</span>:             mqClientException.setResponseCode(ClientErrorCode.CONNECT_BROKER_EXCEPTION);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">121</span>:         } 
      
       
      <span style="">else</span> 
      
       
      <span style="">if</span> (exception 
      
       
      <span style="">instanceof</span> RemotingTimeoutException) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">122</span>:             mqClientException.setResponseCode(ClientErrorCode.ACCESS_BROKER_TIMEOUT);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">123</span>:         } 
      
       
      <span style="">else</span> 
      
       
      <span style="">if</span> (exception 
      
       
      <span style="">instanceof</span> MQClientException) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">124</span>:             mqClientException.setResponseCode(ClientErrorCode.BROKER_NOT_EXIST_EXCEPTION);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">125</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">126</span>:         
      
       
      <span style="">throw</span> mqClientException;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">127</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">128</span>:     
      
       
      <span style="">// Namesrv找不到异常</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">129</span>:     List&lt;String&gt; nsList = 
      
       
      <span style="">this</span>.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">130</span>:     
      
       
      <span style="">if</span> (
      
       
      <span style="">null</span> == nsList || nsList.isEmpty()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">131</span>:         
      
       
      <span style="">throw</span> 
      
       
      <span style="">new</span> MQClientException(
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">132</span>:             
      
       
      <span style="">"No name server address, please set it."</span> + FAQUrl.suggestTodo(FAQUrl.NAME_SERVER_ADDR_NOT_EXIST_URL), 
      
       
      <span style="">null</span>).setResponseCode(ClientErrorCode.NO_NAME_SERVER_EXCEPTION);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">133</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">134</span>:     
      
       
      <span style="">// 消息路由找不到异常</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">135</span>:     
      
       
      <span style="">throw</span> 
      
       
      <span style="">new</span> MQClientException(
      
       
      <span style="">"No route info of this topic, "</span> + msg.getTopic() + FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO),
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">136</span>:         
      
       
      <span style="">null</span>).setResponseCode(ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">137</span>: }
     
      
     </div></pre> </td> 
  </tr> 
 </tbody> 
</table>

*  说明 ：发送消息。步骤：获取消息路由信息，选择要发送到的消息队列，执行消息发送核心方法，并对发送结果进行封装返回。
 *  第 1 至 7 行：对`sendsendDefaultImpl(...)`进行封装。
 *  第 20 行 ：`invokeID`仅仅用于打印日志，无实际的业务用途。
 *  第 25 行 ：获取 Topic路由信息， 详细解析见：[DefaultMQProducerImpl\#tryToFindTopicPublishInfo()][DefaultMQProducerImpl_tryToFindTopicPublishInfo]
 *  第 30 & 34 行 ：计算调用发送消息到成功为止的最大次数，并进行循环。同步或异步发送消息会调用多次，默认配置为3次。
 *  第 36 行 ：选择消息要发送到的队列，详细解析见：[MQFaultStrategy][]
 *  第 43 行 ：调用发送消息核心方法，详细解析见：[DefaultMQProducerImpl\#sendKernelImpl()][DefaultMQProducerImpl_sendKernelImpl]
 *  第 46 行 ：更新`Broker`可用性信息。在选择发送到的消息队列时，会参考`Broker`发送消息的延迟，详细解析见：[MQFaultStrategy][]
 *  第 62 至 68 行：当抛出`RemotingException`时，如果进行消息发送失败重试，则可能导致消息发送重复。例如，发送消息超时(`RemotingTimeoutException`)，实际`Broker`接收到该消息并处理成功。因此，`Consumer`在消费时，需要保证幂等性。

### DefaultMQProducerImpl\#tryToFindTopicPublishInfo() ###

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style=""> <pre style="">
     
      
     <div style=""> 
      
       
      <span style="">1</span>: 
      
       
      <span style=""><span style="">private</span> TopicPublishInfo <span style="">tryToFindTopicPublishInfo</span><span style="">(<span style="">final</span> String topic)</span> </span>{
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">2</span>:     
      
       
      <span style="">// 缓存中获取 Topic发布信息</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">3</span>:     TopicPublishInfo topicPublishInfo = 
      
       
      <span style="">this</span>.topicPublishInfoTable.get(topic);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">4</span>:     
      
       
      <span style="">// 当无可用的 Topic发布信息时，从Namesrv获取一次</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">5</span>:     
      
       
      <span style="">if</span> (
      
       
      <span style="">null</span> == topicPublishInfo || !topicPublishInfo.ok()) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">6</span>:         
      
       
      <span style="">this</span>.topicPublishInfoTable.putIfAbsent(topic, 
      
       
      <span style="">new</span> TopicPublishInfo());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">7</span>:         
      
       
      <span style="">this</span>.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">8</span>:         topicPublishInfo = 
      
       
      <span style="">this</span>.topicPublishInfoTable.get(topic);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">9</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">10</span>:     
      
       
      <span style="">// 若获取的 Topic发布信息时候可用，则返回</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">11</span>:     
      
       
      <span style="">if</span> (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">12</span>:         
      
       
      <span style="">return</span> topicPublishInfo;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">13</span>:     } 
      
       
      <span style="">else</span> { 
      
       
      <span style="">// 使用 {@link DefaultMQProducer#createTopicKey} 对应的 Topic发布信息。用于 Topic发布信息不存在 &amp;&amp; Broker支持自动创建Topic</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">14</span>:         
      
       
      <span style="">this</span>.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, 
      
       
      <span style="">true</span>, 
      
       
      <span style="">this</span>.defaultMQProducer);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">15</span>:         topicPublishInfo = 
      
       
      <span style="">this</span>.topicPublishInfoTable.get(topic);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">16</span>:         
      
       
      <span style="">return</span> topicPublishInfo;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">17</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">18</span>: }
     
      
     </div></pre> </td> 
  </tr> 
 </tbody> 
</table>

*  说明 ：获得 Topic发布信息。优先从缓存`topicPublishInfoTable`，其次从`Namesrv`中获得。
 *  第 3 行 ：从缓存`topicPublishInfoTable`中获得 Topic发布信息。
 *  第 5 至 9 行 ：从 `Namesrv` 中获得 Topic发布信息。
 *  第 13 至 17 行 ：当从 `Namesrv` 无法获取时，使用 `{@link DefaultMQProducer#createTopicKey}` 对应的 Topic发布信息。目的是当 `Broker` 开启自动创建 Topic开关时，`Broker` 接收到消息后自动创建Topic，详细解析见[《RocketMQ 源码分析 —— Topic》][RocketMQ _ _ Topic]。

### MQFaultStrategy ###

> ![Latency类图][Latency]

#### MQFaultStrategy ####

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style=""> <pre style="">
     
      
     <div style=""> 
      
       
      <span style="">1</span>: 
      
       
      <span style="">public</span> 
      
       
      <span style=""><span style="">class</span> <span style="">MQFaultStrategy</span> </span>{
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">2</span>:     
      
       
      <span style="">private</span> 
      
       
      <span style="">final</span> 
      
       
      <span style="">static</span> Logger log = ClientLogger.getLog();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">3</span>: 
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">4</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 5: * 延迟故障容错，维护每个Broker的发送消息的延迟</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 6: * key：brokerName</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 7: */</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">8</span>:     
      
       
      <span style="">private</span> 
      
       
      <span style="">final</span> LatencyFaultTolerance&lt;String&gt; latencyFaultTolerance = 
      
       
      <span style="">new</span> LatencyFaultToleranceImpl();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">9</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">10: * 发送消息延迟容错开关</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">11: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">12</span>:     
      
       
      <span style="">private</span> 
      
       
      <span style="">boolean</span> sendLatencyFaultEnable = 
      
       
      <span style="">false</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">13</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">14: * 延迟级别数组</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">15: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">16</span>:     
      
       
      <span style="">private</span> 
      
       
      <span style="">long</span>[] latencyMax = {
      
       
      <span style="">50L</span>, 
      
       
      <span style="">100L</span>, 
      
       
      <span style="">550L</span>, 
      
       
      <span style="">1000L</span>, 
      
       
      <span style="">2000L</span>, 
      
       
      <span style="">3000L</span>, 
      
       
      <span style="">15000L</span>};
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">17</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">18: * 不可用时长数组</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">19: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">20</span>:     
      
       
      <span style="">private</span> 
      
       
      <span style="">long</span>[] notAvailableDuration = {
      
       
      <span style="">0L</span>, 
      
       
      <span style="">0L</span>, 
      
       
      <span style="">30000L</span>, 
      
       
      <span style="">60000L</span>, 
      
       
      <span style="">120000L</span>, 
      
       
      <span style="">180000L</span>, 
      
       
      <span style="">600000L</span>};
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">21</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">22</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">23: * 根据 Topic发布信息 选择一个消息队列</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">24: *</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">25: * <span style="">@param</span> tpInfo Topic发布信息</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">26: * <span style="">@param</span> lastBrokerName brokerName</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">27: * <span style="">@return</span> 消息队列</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">28: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">29</span>:     
      
       
      <span style=""><span style="">public</span> MessageQueue <span style="">selectOneMessageQueue</span><span style="">(<span style="">final</span> TopicPublishInfo tpInfo, <span style="">final</span> String lastBrokerName)</span> </span>{
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">30</span>:         
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.sendLatencyFaultEnable) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">31</span>:             
      
       
      <span style="">try</span> {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">32</span>:                 
      
       
      <span style="">// 获取 brokerName=lastBrokerName &amp;&amp; 可用的一个消息队列</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">33</span>:                 
      
       
      <span style="">int</span> index = tpInfo.getSendWhichQueue().getAndIncrement();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">34</span>:                 
      
       
      <span style="">for</span> (
      
       
      <span style="">int</span> i = 
      
       
      <span style="">0</span>; i &lt; tpInfo.getMessageQueueList().size(); i++) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">35</span>:                     
      
       
      <span style="">int</span> pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">36</span>:                     
      
       
      <span style="">if</span> (pos &lt; 
      
       
      <span style="">0</span>)
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">37</span>:                         pos = 
      
       
      <span style="">0</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">38</span>:                     MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">39</span>:                     
      
       
      <span style="">if</span> (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">40</span>:                         
      
       
      <span style="">if</span> (
      
       
      <span style="">null</span> == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">41</span>:                             
      
       
      <span style="">return</span> mq;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">42</span>:                     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">43</span>:                 }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">44</span>:                 
      
       
      <span style="">// 选择一个相对好的broker，并获得其对应的一个消息队列，不考虑该队列的可用性</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">45</span>:                 
      
       
      <span style="">final</span> String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">46</span>:                 
      
       
      <span style="">int</span> writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">47</span>:                 
      
       
      <span style="">if</span> (writeQueueNums &gt; 
      
       
      <span style="">0</span>) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">48</span>:                     
      
       
      <span style="">final</span> MessageQueue mq = tpInfo.selectOneMessageQueue();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">49</span>:                     
      
       
      <span style="">if</span> (notBestBroker != 
      
       
      <span style="">null</span>) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">50</span>:                         mq.setBrokerName(notBestBroker);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">51</span>:                         mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">52</span>:                     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">53</span>:                     
      
       
      <span style="">return</span> mq;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">54</span>:                 } 
      
       
      <span style="">else</span> {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">55</span>:                     latencyFaultTolerance.remove(notBestBroker);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">56</span>:                 }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">57</span>:             } 
      
       
      <span style="">catch</span> (Exception e) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">58</span>:                 log.error(
      
       
      <span style="">"Error occurred when selecting message queue"</span>, e);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">59</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">60</span>:             
      
       
      <span style="">// 选择一个消息队列，不考虑队列的可用性</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">61</span>:             
      
       
      <span style="">return</span> tpInfo.selectOneMessageQueue();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">62</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">63</span>:         
      
       
      <span style="">// 获得 lastBrokerName 对应的一个消息队列，不考虑该队列的可用性</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">64</span>:         
      
       
      <span style="">return</span> tpInfo.selectOneMessageQueue(lastBrokerName);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">65</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">66</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">67</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">68: * 更新延迟容错信息</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">69: *</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">70: * <span style="">@param</span> brokerName brokerName</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">71: * <span style="">@param</span> currentLatency 延迟</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">72: * <span style="">@param</span> isolation 是否隔离。当开启隔离时，默认延迟为30000。目前主要用于发送消息异常时</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">73: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">74</span>:     
      
       
      <span style=""><span style="">public</span> <span style="">void</span> <span style="">updateFaultItem</span><span style="">(<span style="">final</span> String brokerName, <span style="">final</span> <span style="">long</span> currentLatency, <span style="">boolean</span> isolation)</span> </span>{
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">75</span>:         
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.sendLatencyFaultEnable) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">76</span>:             
      
       
      <span style="">long</span> duration = computeNotAvailableDuration(isolation ? 
      
       
      <span style="">30000</span> : currentLatency);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">77</span>:             
      
       
      <span style="">this</span>.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">78</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">79</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">80</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">81</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">82: * 计算延迟对应的不可用时间</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">83: *</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">84: * <span style="">@param</span> currentLatency 延迟</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">85: * <span style="">@return</span> 不可用时间</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">86: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">87</span>:     
      
       
      <span style=""><span style="">private</span> <span style="">long</span> <span style="">computeNotAvailableDuration</span><span style="">(<span style="">final</span> <span style="">long</span> currentLatency)</span> </span>{
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">88</span>:         
      
       
      <span style="">for</span> (
      
       
      <span style="">int</span> i = latencyMax.length - 
      
       
      <span style="">1</span>; i &gt;= 
      
       
      <span style="">0</span>; i--) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">89</span>:             
      
       
      <span style="">if</span> (currentLatency &gt;= latencyMax[i])
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">90</span>:                 
      
       
      <span style="">return</span> 
      
       
      <span style="">this</span>.notAvailableDuration[i];
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">91</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">92</span>:         
      
       
      <span style="">return</span> 
      
       
      <span style="">0</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">93</span>:     }
     
      
     </div></pre> </td> 
  </tr> 
 </tbody> 
</table>

*  说明 ：`Producer`消息发送容错策略。默认情况下容错策略关闭，即`sendLatencyFaultEnable=false`。
 *  第 30 至 62 行 ：容错策略选择消息队列逻辑。优先获取可用队列，其次选择一个broker获取队列，最差返回任意broker的一个队列。
 *  第 64 行 ：未开启容错策略选择消息队列逻辑。
 *  第 74 至 79 行 ：更新延迟容错信息。当 `Producer` 发送消息时间过长，则逻辑认为N秒内不可用。按照`latencyMax`，`notAvailableDuration`的配置，对应如下：
    
    | Producer发送消息消耗时长 | Broker不可用时长 |  
    | — | — |  
    | >= 15000 ms | 600 1000 ms |  
    | >= 3000 ms | 180 1000 ms |  
    | >= 2000 ms | 120 1000 ms |  
    | >= 1000 ms | 60 1000 ms |  
    | >= 550 ms | 30 \* 1000 ms |  
    | >= 100 ms | 0 ms |  
    | >= 50 ms | 0 ms |

#### LatencyFaultTolerance ####

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style=""> <pre style="">
     
      
     <div style=""> 
      
       
      <span style="">1</span>: 
      
       
      <span style="">public</span> 
      
       
      <span style=""><span style="">interface</span> <span style="">LatencyFaultTolerance</span>&lt;<span style="">T</span>&gt; </span>{
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">2</span>: 
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">3</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 4: * 更新对应的延迟和不可用时长</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 5: *</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 6: * <span style="">@param</span> name 对象</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 7: * <span style="">@param</span> currentLatency 延迟</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 8: * <span style="">@param</span> notAvailableDuration 不可用时长</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 9: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">10</span>:     
      
       
      <span style=""><span style="">void</span> <span style="">updateFaultItem</span><span style="">(<span style="">final</span> T name, <span style="">final</span> <span style="">long</span> currentLatency, <span style="">final</span> <span style="">long</span> notAvailableDuration)</span></span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">11</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">12</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">13: * 对象是否可用</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">14: *</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">15: * <span style="">@param</span> name 对象</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">16: * <span style="">@return</span> 是否可用</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">17: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">18</span>:     
      
       
      <span style=""><span style="">boolean</span> <span style="">isAvailable</span><span style="">(<span style="">final</span> T name)</span></span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">19</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">20</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">21: * 移除对象</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">22: *</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">23: * <span style="">@param</span> name 对象</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">24: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">25</span>:     
      
       
      <span style=""><span style="">void</span> <span style="">remove</span><span style="">(<span style="">final</span> T name)</span></span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">26</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">27</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">28: * 获取一个对象</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">29: *</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">30: * <span style="">@return</span> 对象</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">31: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">32</span>:     
      
       
      <span style="">T <span style="">pickOneAtLeast</span><span style="">()</span></span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">33</span>: }
     
      
     </div></pre> </td> 
  </tr> 
 </tbody> 
</table>

*  说明 ：延迟故障容错接口

#### LatencyFaultToleranceImpl ####

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style=""> <pre style="">
     
      
     <div style=""> 
      
       
      <span style="">1</span>: 
      
       
      <span style="">public</span> 
      
       
      <span style=""><span style="">class</span> <span style="">LatencyFaultToleranceImpl</span> <span style="">implements</span> <span style="">LatencyFaultTolerance</span>&lt;<span style="">String</span>&gt; </span>{
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">2</span>: 
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">3</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 4: * 对象故障信息Table</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 5: */</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">6</span>:     
      
       
      <span style="">private</span> 
      
       
      <span style="">final</span> ConcurrentHashMap&lt;String, FaultItem&gt; faultItemTable = 
      
       
      <span style="">new</span> ConcurrentHashMap&lt;&gt;(
      
       
      <span style="">16</span>);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">7</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 8: * 对象选择Index</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 9: * <span style="">@see</span> #pickOneAtLeast()</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">10: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">11</span>:     
      
       
      <span style="">private</span> 
      
       
      <span style="">final</span> ThreadLocalIndex whichItemWorst = 
      
       
      <span style="">new</span> ThreadLocalIndex();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">12</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">13</span>:     
      
       
      <span style="">@Override</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">14</span>:     
      
       
      <span style=""><span style="">public</span> <span style="">void</span> <span style="">updateFaultItem</span><span style="">(<span style="">final</span> String name, <span style="">final</span> <span style="">long</span> currentLatency, <span style="">final</span> <span style="">long</span> notAvailableDuration)</span> </span>{
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">15</span>:         FaultItem old = 
      
       
      <span style="">this</span>.faultItemTable.get(name);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">16</span>:         
      
       
      <span style="">if</span> (
      
       
      <span style="">null</span> == old) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">17</span>:             
      
       
      <span style="">// 创建对象</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">18</span>:             
      
       
      <span style="">final</span> FaultItem faultItem = 
      
       
      <span style="">new</span> FaultItem(name);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">19</span>:             faultItem.setCurrentLatency(currentLatency);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">20</span>:             faultItem.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">21</span>:             
      
       
      <span style="">// 更新对象</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">22</span>:             old = 
      
       
      <span style="">this</span>.faultItemTable.putIfAbsent(name, faultItem);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">23</span>:             
      
       
      <span style="">if</span> (old != 
      
       
      <span style="">null</span>) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">24</span>:                 old.setCurrentLatency(currentLatency);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">25</span>:                 old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">26</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">27</span>:         } 
      
       
      <span style="">else</span> { 
      
       
      <span style="">// 更新对象</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">28</span>:             old.setCurrentLatency(currentLatency);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">29</span>:             old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">30</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">31</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">32</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">33</span>:     
      
       
      <span style="">@Override</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">34</span>:     
      
       
      <span style=""><span style="">public</span> <span style="">boolean</span> <span style="">isAvailable</span><span style="">(<span style="">final</span> String name)</span> </span>{
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">35</span>:         
      
       
      <span style="">final</span> FaultItem faultItem = 
      
       
      <span style="">this</span>.faultItemTable.get(name);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">36</span>:         
      
       
      <span style="">if</span> (faultItem != 
      
       
      <span style="">null</span>) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">37</span>:             
      
       
      <span style="">return</span> faultItem.isAvailable();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">38</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">39</span>:         
      
       
      <span style="">return</span> 
      
       
      <span style="">true</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">40</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">41</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">42</span>:     
      
       
      <span style="">@Override</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">43</span>:     
      
       
      <span style=""><span style="">public</span> <span style="">void</span> <span style="">remove</span><span style="">(<span style="">final</span> String name)</span> </span>{
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">44</span>:         
      
       
      <span style="">this</span>.faultItemTable.remove(name);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">45</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">46</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">47</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">48: * 选择一个相对优秀的对象</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">49: *</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">50: * <span style="">@return</span> 对象</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">51: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">52</span>:     
      
       
      <span style="">@Override</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">53</span>:     
      
       
      <span style=""><span style="">public</span> String <span style="">pickOneAtLeast</span><span style="">()</span> </span>{
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">54</span>:         
      
       
      <span style="">// 创建数组</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">55</span>:         
      
       
      <span style="">final</span> Enumeration&lt;FaultItem&gt; elements = 
      
       
      <span style="">this</span>.faultItemTable.elements();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">56</span>:         List&lt;FaultItem&gt; tmpList = 
      
       
      <span style="">new</span> LinkedList&lt;&gt;();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">57</span>:         
      
       
      <span style="">while</span> (elements.hasMoreElements()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">58</span>:             
      
       
      <span style="">final</span> FaultItem faultItem = elements.nextElement();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">59</span>:             tmpList.add(faultItem);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">60</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">61</span>:         
      
       
      <span style="">//</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">62</span>:         
      
       
      <span style="">if</span> (!tmpList.isEmpty()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">63</span>:             
      
       
      <span style="">// 打乱 + 排序。TODO 疑问：应该只能二选一。猜测Collections.shuffle(tmpList)去掉。</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">64</span>:             Collections.shuffle(tmpList);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">65</span>:             Collections.sort(tmpList);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">66</span>:             
      
       
      <span style="">// 选择顺序在前一半的对象</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">67</span>:             
      
       
      <span style="">final</span> 
      
       
      <span style="">int</span> half = tmpList.size() / 
      
       
      <span style="">2</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">68</span>:             
      
       
      <span style="">if</span> (half &lt;= 
      
       
      <span style="">0</span>) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">69</span>:                 
      
       
      <span style="">return</span> tmpList.get(
      
       
      <span style="">0</span>).getName();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">70</span>:             } 
      
       
      <span style="">else</span> {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">71</span>:                 
      
       
      <span style="">final</span> 
      
       
      <span style="">int</span> i = 
      
       
      <span style="">this</span>.whichItemWorst.getAndIncrement() % half;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">72</span>:                 
      
       
      <span style="">return</span> tmpList.get(i).getName();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">73</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">74</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">75</span>:         
      
       
      <span style="">return</span> 
      
       
      <span style="">null</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">76</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">77</span>: }
     
      
     </div></pre> </td> 
  </tr> 
 </tbody> 
</table>

*  说明 ：延迟故障容错实现。维护每个对象的信息。

#### FaultItem ####

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style=""> <pre style="">
     
      
     <div style=""> 
      
       
      <span style="">1</span>: 
      
       
      <span style=""><span style="">class</span> <span style="">FaultItem</span> <span style="">implements</span> <span style="">Comparable</span>&lt;<span style="">FaultItem</span>&gt; </span>{
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">2</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 3: * 对象名</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 4: */</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">5</span>:     
      
       
      <span style="">private</span> 
      
       
      <span style="">final</span> String name;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">6</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 7: * 延迟</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""> 8: */</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">9</span>:     
      
       
      <span style="">private</span> 
      
       
      <span style="">volatile</span> 
      
       
      <span style="">long</span> currentLatency;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">10</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">11: * 开始可用时间</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">12: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">13</span>:     
      
       
      <span style="">private</span> 
      
       
      <span style="">volatile</span> 
      
       
      <span style="">long</span> startTimestamp;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">14</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">15</span>:     
      
       
      <span style=""><span style="">public</span> <span style="">FaultItem</span><span style="">(<span style="">final</span> String name)</span> </span>{
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">16</span>:         
      
       
      <span style="">this</span>.name = name;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">17</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">18</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">19</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">20: * 比较对象</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">21: * 可用性 &gt; 延迟 &gt; 开始可用时间</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">22: *</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">23: * <span style="">@param</span> other other</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">24: * <span style="">@return</span> 升序</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">25: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">26</span>:     
      
       
      <span style="">@Override</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">27</span>:     
      
       
      <span style=""><span style="">public</span> <span style="">int</span> <span style="">compareTo</span><span style="">(<span style="">final</span> FaultItem other)</span> </span>{
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">28</span>:         
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.isAvailable() != other.isAvailable()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">29</span>:             
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.isAvailable())
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">30</span>:                 
      
       
      <span style="">return</span> -
      
       
      <span style="">1</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">31</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">32</span>:             
      
       
      <span style="">if</span> (other.isAvailable())
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">33</span>:                 
      
       
      <span style="">return</span> 
      
       
      <span style="">1</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">34</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">35</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">36</span>:         
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.currentLatency &lt; other.currentLatency)
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">37</span>:             
      
       
      <span style="">return</span> -
      
       
      <span style="">1</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">38</span>:         
      
       
      <span style="">else</span> 
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.currentLatency &gt; other.currentLatency) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">39</span>:             
      
       
      <span style="">return</span> 
      
       
      <span style="">1</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">40</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">41</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">42</span>:         
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.startTimestamp &lt; other.startTimestamp)
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">43</span>:             
      
       
      <span style="">return</span> -
      
       
      <span style="">1</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">44</span>:         
      
       
      <span style="">else</span> 
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.startTimestamp &gt; other.startTimestamp) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">45</span>:             
      
       
      <span style="">return</span> 
      
       
      <span style="">1</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">46</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">47</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">48</span>:         
      
       
      <span style="">return</span> 
      
       
      <span style="">0</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">49</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">50</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">51</span>:     
      
       
      <span style="">/**</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">52: * 是否可用：当开始可用时间大于当前时间</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">53: *</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">54: * <span style="">@return</span> 是否可用</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">55: */</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">56</span>:     
      
       
      <span style=""><span style="">public</span> <span style="">boolean</span> <span style="">isAvailable</span><span style="">()</span> </span>{
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">57</span>:         
      
       
      <span style="">return</span> (System.currentTimeMillis() - startTimestamp) &gt;= 
      
       
      <span style="">0</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">58</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">59</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">60</span>:     
      
       
      <span style="">@Override</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">61</span>:     
      
       
      <span style=""><span style="">public</span> <span style="">int</span> <span style="">hashCode</span><span style="">()</span> </span>{
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">62</span>:         
      
       
      <span style="">int</span> result = getName() != 
      
       
      <span style="">null</span> ? getName().hashCode() : 
      
       
      <span style="">0</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">63</span>:         result = 
      
       
      <span style="">31</span> * result + (
      
       
      <span style="">int</span>) (getCurrentLatency() ^ (getCurrentLatency() &gt;&gt;&gt; 
      
       
      <span style="">32</span>));
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">64</span>:         result = 
      
       
      <span style="">31</span> * result + (
      
       
      <span style="">int</span>) (getStartTimestamp() ^ (getStartTimestamp() &gt;&gt;&gt; 
      
       
      <span style="">32</span>));
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">65</span>:         
      
       
      <span style="">return</span> result;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">66</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">67</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">68</span>:     
      
       
      <span style="">@Override</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">69</span>:     
      
       
      <span style=""><span style="">public</span> <span style="">boolean</span> <span style="">equals</span><span style="">(<span style="">final</span> Object o)</span> </span>{
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">70</span>:         
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span> == o)
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">71</span>:             
      
       
      <span style="">return</span> 
      
       
      <span style="">true</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">72</span>:         
      
       
      <span style="">if</span> (!(o 
      
       
      <span style="">instanceof</span> FaultItem))
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">73</span>:             
      
       
      <span style="">return</span> 
      
       
      <span style="">false</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">74</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">75</span>:         
      
       
      <span style="">final</span> FaultItem faultItem = (FaultItem) o;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">76</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">77</span>:         
      
       
      <span style="">if</span> (getCurrentLatency() != faultItem.getCurrentLatency())
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">78</span>:             
      
       
      <span style="">return</span> 
      
       
      <span style="">false</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">79</span>:         
      
       
      <span style="">if</span> (getStartTimestamp() != faultItem.getStartTimestamp())
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">80</span>:             
      
       
      <span style="">return</span> 
      
       
      <span style="">false</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">81</span>:         
      
       
      <span style="">return</span> getName() != 
      
       
      <span style="">null</span> ? getName().equals(faultItem.getName()) : faultItem.getName() == 
      
       
      <span style="">null</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">82</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">83</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">84</span>: }
     
      
     </div></pre> </td> 
  </tr> 
 </tbody> 
</table>

*  说明 ：对象故障信息。维护对象的名字、延迟、开始可用的时间。

### DefaultMQProducerImpl\#sendKernelImpl() ###

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style=""> <pre style="">
     
      
     <div style="">  
      
       
      <span style="">1</span>: 
      
       
      <span style=""><span style="">private</span> SendResult <span style="">sendKernelImpl</span><span style="">(<span style="">final</span> Message msg, //</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">2</span>: <span style="">final</span> MessageQueue mq, //</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">3</span>: <span style="">final</span> CommunicationMode communicationMode, //</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">4</span>: <span style="">final</span> SendCallback sendCallback, //</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">5</span>: <span style="">final</span> TopicPublishInfo topicPublishInfo, //</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">6</span>: <span style="">final</span> <span style="">long</span> timeout)</span> <span style="">throws</span> MQClientException, RemotingException, MQBrokerException, InterruptedException </span>{
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">7</span>:     
      
       
      <span style="">// 获取 broker地址</span>
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">8</span>:     String brokerAddr = 
      
       
      <span style="">this</span>.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">9</span>:     
      
       
      <span style="">if</span> (
      
       
      <span style="">null</span> == brokerAddr) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">10</span>:         tryToFindTopicPublishInfo(mq.getTopic());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">11</span>:         brokerAddr = 
      
       
      <span style="">this</span>.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">12</span>:     }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">13</span>:     
      
       
      <span style="">//</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">14</span>:     SendMessageContext context = 
      
       
      <span style="">null</span>;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">15</span>:     
      
       
      <span style="">if</span> (brokerAddr != 
      
       
      <span style="">null</span>) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">16</span>:         
      
       
      <span style="">// 是否使用broker vip通道。broker会开启两个端口对外服务。</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">17</span>:         brokerAddr = MixAll.brokerVIPChannel(
      
       
      <span style="">this</span>.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">18</span>:         
      
       
      <span style="">byte</span>[] prevBody = msg.getBody(); 
      
       
      <span style="">// 记录消息内容。下面逻辑可能改变消息内容，例如消息压缩。</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">19</span>:         
      
       
      <span style="">try</span> {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">20</span>:             
      
       
      <span style="">// 设置唯一编号</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">21</span>:             MessageClientIDSetter.setUniqID(msg);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">22</span>:             
      
       
      <span style="">// 消息压缩</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">23</span>:             
      
       
      <span style="">int</span> sysFlag = 
      
       
      <span style="">0</span>;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">24</span>:             
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.tryToCompressMessage(msg)) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">25</span>:                 sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">26</span>:             }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">27</span>:             
      
       
      <span style="">// 事务</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">28</span>:             
      
       
      <span style="">final</span> String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">29</span>:             
      
       
      <span style="">if</span> (tranMsg != 
      
       
      <span style="">null</span> &amp;&amp; Boolean.parseBoolean(tranMsg)) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">30</span>:                 sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">31</span>:             }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">32</span>:             
      
       
      <span style="">// hook：发送消息校验</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">33</span>:             
      
       
      <span style="">if</span> (hasCheckForbiddenHook()) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">34</span>:                 CheckForbiddenContext checkForbiddenContext = 
      
       
      <span style="">new</span> CheckForbiddenContext();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">35</span>:                 checkForbiddenContext.setNameSrvAddr(
      
       
      <span style="">this</span>.defaultMQProducer.getNamesrvAddr());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">36</span>:                 checkForbiddenContext.setGroup(
      
       
      <span style="">this</span>.defaultMQProducer.getProducerGroup());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">37</span>:                 checkForbiddenContext.setCommunicationMode(communicationMode);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">38</span>:                 checkForbiddenContext.setBrokerAddr(brokerAddr);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">39</span>:                 checkForbiddenContext.setMessage(msg);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">40</span>:                 checkForbiddenContext.setMq(mq);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">41</span>:                 checkForbiddenContext.setUnitMode(
      
       
      <span style="">this</span>.isUnitMode());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">42</span>:                 
      
       
      <span style="">this</span>.executeCheckForbiddenHook(checkForbiddenContext);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">43</span>:             }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">44</span>:             
      
       
      <span style="">// hook：发送消息前逻辑</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">45</span>:             
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.hasSendMessageHook()) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">46</span>:                 context = 
      
       
      <span style="">new</span> SendMessageContext();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">47</span>:                 context.setProducer(
      
       
      <span style="">this</span>);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">48</span>:                 context.setProducerGroup(
      
       
      <span style="">this</span>.defaultMQProducer.getProducerGroup());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">49</span>:                 context.setCommunicationMode(communicationMode);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">50</span>:                 context.setBornHost(
      
       
      <span style="">this</span>.defaultMQProducer.getClientIP());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">51</span>:                 context.setBrokerAddr(brokerAddr);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">52</span>:                 context.setMessage(msg);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">53</span>:                 context.setMq(mq);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">54</span>:                 String isTrans = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">55</span>:                 
      
       
      <span style="">if</span> (isTrans != 
      
       
      <span style="">null</span> &amp;&amp; isTrans.equals(
      
       
      <span style="">"true"</span>)) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">56</span>:                     context.setMsgType(MessageType.Trans_Msg_Half);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">57</span>:                 }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">58</span>:                 
      
       
      <span style="">if</span> (msg.getProperty(
      
       
      <span style="">"__STARTDELIVERTIME"</span>) != 
      
       
      <span style="">null</span> || msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != 
      
       
      <span style="">null</span>) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">59</span>:                     context.setMsgType(MessageType.Delay_Msg);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">60</span>:                 }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">61</span>:                 
      
       
      <span style="">this</span>.executeSendMessageHookBefore(context);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">62</span>:             }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">63</span>:             
      
       
      <span style="">// 构建发送消息请求</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">64</span>:             SendMessageRequestHeader requestHeader = 
      
       
      <span style="">new</span> SendMessageRequestHeader();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">65</span>:             requestHeader.setProducerGroup(
      
       
      <span style="">this</span>.defaultMQProducer.getProducerGroup());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">66</span>:             requestHeader.setTopic(msg.getTopic());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">67</span>:             requestHeader.setDefaultTopic(
      
       
      <span style="">this</span>.defaultMQProducer.getCreateTopicKey());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">68</span>:             requestHeader.setDefaultTopicQueueNums(
      
       
      <span style="">this</span>.defaultMQProducer.getDefaultTopicQueueNums());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">69</span>:             requestHeader.setQueueId(mq.getQueueId());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">70</span>:             requestHeader.setSysFlag(sysFlag);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">71</span>:             requestHeader.setBornTimestamp(System.currentTimeMillis());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">72</span>:             requestHeader.setFlag(msg.getFlag());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">73</span>:             requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">74</span>:             requestHeader.setReconsumeTimes(
      
       
      <span style="">0</span>);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">75</span>:             requestHeader.setUnitMode(
      
       
      <span style="">this</span>.isUnitMode());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">76</span>:             
      
       
      <span style="">if</span> (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { 
      
       
      <span style="">// 消息重发Topic</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">77</span>:                 String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">78</span>:                 
      
       
      <span style="">if</span> (reconsumeTimes != 
      
       
      <span style="">null</span>) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">79</span>:                     requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">80</span>:                     MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">81</span>:                 }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">82</span>:                 String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(msg);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">83</span>:                 
      
       
      <span style="">if</span> (maxReconsumeTimes != 
      
       
      <span style="">null</span>) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">84</span>:                     requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">85</span>:                     MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_MAX_RECONSUME_TIMES);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">86</span>:                 }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">87</span>:             }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">88</span>:             
      
       
      <span style="">// 发送消息</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">89</span>:             SendResult sendResult = 
      
       
      <span style="">null</span>;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">90</span>:             
      
       
      <span style="">switch</span> (communicationMode) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">91</span>:                 
      
       
      <span style="">case</span> ASYNC:
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">92</span>:                     sendResult = 
      
       
      <span style="">this</span>.mQClientFactory.getMQClientAPIImpl().sendMessage(
      
       
      <span style="">//</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">93</span>:                         brokerAddr, 
      
       
      <span style="">// 1</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">94</span>:                         mq.getBrokerName(), 
      
       
      <span style="">// 2</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">95</span>:                         msg, 
      
       
      <span style="">// 3</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">96</span>:                         requestHeader, 
      
       
      <span style="">// 4</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">97</span>:                         timeout, 
      
       
      <span style="">// 5</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">98</span>:                         communicationMode, 
      
       
      <span style="">// 6</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">99</span>:                         sendCallback, 
      
       
      <span style="">// 7</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">100</span>:                         topicPublishInfo, 
      
       
      <span style="">// 8</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">101</span>:                         
      
       
      <span style="">this</span>.mQClientFactory, 
      
       
      <span style="">// 9</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">102</span>:                         
      
       
      <span style="">this</span>.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(), 
      
       
      <span style="">// 10</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">103</span>:                         context, 
      
       
      <span style="">//</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">104</span>:                         
      
       
      <span style="">this</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">105</span>:                     
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">106</span>:                 
      
       
      <span style="">case</span> ONEWAY:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">107</span>:                 
      
       
      <span style="">case</span> SYNC:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">108</span>:                     sendResult = 
      
       
      <span style="">this</span>.mQClientFactory.getMQClientAPIImpl().sendMessage(
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">109</span>:                         brokerAddr,
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">110</span>:                         mq.getBrokerName(),
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">111</span>:                         msg,
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">112</span>:                         requestHeader,
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">113</span>:                         timeout,
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">114</span>:                         communicationMode,
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">115</span>:                         context,
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">116</span>:                         
      
       
      <span style="">this</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">117</span>:                     
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">118</span>:                 
      
       
      <span style="">default</span>:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">119</span>:                     
      
       
      <span style="">assert</span> 
      
       
      <span style="">false</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">120</span>:                     
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">121</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">122</span>:             
      
       
      <span style="">// hook：发送消息后逻辑</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">123</span>:             
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.hasSendMessageHook()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">124</span>:                 context.setSendResult(sendResult);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">125</span>:                 
      
       
      <span style="">this</span>.executeSendMessageHookAfter(context);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">126</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">127</span>:             
      
       
      <span style="">// 返回发送结果</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">128</span>:             
      
       
      <span style="">return</span> sendResult;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">129</span>:         } 
      
       
      <span style="">catch</span> (RemotingException e) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">130</span>:             
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.hasSendMessageHook()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">131</span>:                 context.setException(e);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">132</span>:                 
      
       
      <span style="">this</span>.executeSendMessageHookAfter(context);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">133</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">134</span>:             
      
       
      <span style="">throw</span> e;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">135</span>:         } 
      
       
      <span style="">catch</span> (MQBrokerException e) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">136</span>:             
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.hasSendMessageHook()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">137</span>:                 context.setException(e);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">138</span>:                 
      
       
      <span style="">this</span>.executeSendMessageHookAfter(context);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">139</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">140</span>:             
      
       
      <span style="">throw</span> e;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">141</span>:         } 
      
       
      <span style="">catch</span> (InterruptedException e) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">142</span>:             
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.hasSendMessageHook()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">143</span>:                 context.setException(e);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">144</span>:                 
      
       
      <span style="">this</span>.executeSendMessageHookAfter(context);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">145</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">146</span>:             
      
       
      <span style="">throw</span> e;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">147</span>:         } 
      
       
      <span style="">finally</span> {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">148</span>:             msg.setBody(prevBody);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">149</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">150</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">151</span>:     
      
       
      <span style="">// broker为空抛出异常</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">152</span>:     
      
       
      <span style="">throw</span> 
      
       
      <span style="">new</span> MQClientException(
      
       
      <span style="">"The broker["</span> + mq.getBrokerName() + 
      
       
      <span style="">"] not exist"</span>, 
      
       
      <span style="">null</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">153</span>: }
     
      
     </div></pre> </td> 
  </tr> 
 </tbody> 
</table>

*  说明 ：发送消息核心方法。该方法真正发起网络请求，发送消息给 `Broker`。
 *  第 21 行 ：生产消息编号，详细解析见[《RocketMQ 源码分析 —— Message 基础》][RocketMQ _ _ Message 1]。
 *  第 64 至 121 行 ：构建发送消息请求`SendMessageRequestHeader`。
 *  第 107 至 117 行 ：执行 `MQClientInstance#sendMessage(...)` 发起网络请求。

# 3、Broker 接收消息 #

> ![接收发送消息API顺序图][API]

## SendMessageProcessor\#sendMessage ##

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style=""> <pre style="">
     
      
     <div style="">  
      
       
      <span style="">1</span>: 
      
       
      <span style="">@Override</span>
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">2</span>: 
      
       
      <span style=""><span style="">public</span> RemotingCommand <span style="">processRequest</span><span style="">(ChannelHandlerContext ctx, RemotingCommand request)</span> <span style="">throws</span> RemotingCommandException </span>{
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">3</span>:     SendMessageContext mqtraceContext;
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">4</span>:     
      
       
      <span style="">switch</span> (request.getCode()) {
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">5</span>:         
      
       
      <span style="">case</span> RequestCode.CONSUMER_SEND_MSG_BACK:
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">6</span>:             
      
       
      <span style="">return</span> 
      
       
      <span style="">this</span>.consumerSendMsgBack(ctx, request);
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">7</span>:         
      
       
      <span style="">default</span>:
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">8</span>:             
      
       
      <span style="">// 解析请求</span>
     
      
     </div>
     
      
     <div style="">  
      
       
      <span style="">9</span>:             SendMessageRequestHeader requestHeader = parseRequestHeader(request);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">10</span>:             
      
       
      <span style="">if</span> (requestHeader == 
      
       
      <span style="">null</span>) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">11</span>:                 
      
       
      <span style="">return</span> 
      
       
      <span style="">null</span>;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">12</span>:             }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">13</span>:             
      
       
      <span style="">// 发送请求Context。在 hook 场景下使用</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">14</span>:             mqtraceContext = buildMsgContext(ctx, requestHeader);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">15</span>:             
      
       
      <span style="">// hook：处理发送消息前逻辑</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">16</span>:             
      
       
      <span style="">this</span>.executeSendMessageHookBefore(ctx, request, mqtraceContext);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">17</span>:             
      
       
      <span style="">// 处理发送消息逻辑</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">18</span>:             
      
       
      <span style="">final</span> RemotingCommand response = 
      
       
      <span style="">this</span>.sendMessage(ctx, request, mqtraceContext, requestHeader);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">19</span>:             
      
       
      <span style="">// hook：处理发送消息后逻辑</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">20</span>:             
      
       
      <span style="">this</span>.executeSendMessageHookAfter(response, mqtraceContext);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">21</span>:             
      
       
      <span style="">return</span> response;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">22</span>:     }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">23</span>: }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">24</span>: 
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">25</span>: 
      
       
      <span style=""><span style="">private</span> RemotingCommand <span style="">sendMessage</span><span style="">(<span style="">final</span> ChannelHandlerContext ctx, //</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">26</span>: <span style="">final</span> RemotingCommand request, //</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">27</span>: <span style="">final</span> SendMessageContext sendMessageContext, //</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">28</span>: <span style="">final</span> SendMessageRequestHeader requestHeader)</span> <span style="">throws</span> RemotingCommandException </span>{
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">29</span>: 
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">30</span>:     
      
       
      <span style="">// 初始化响应</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">31</span>:     
      
       
      <span style="">final</span> RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">32</span>:     
      
       
      <span style="">final</span> SendMessageResponseHeader responseHeader = (SendMessageResponseHeader) response.readCustomHeader();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">33</span>:     response.setOpaque(request.getOpaque());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">34</span>:     response.addExtField(MessageConst.PROPERTY_MSG_REGION, 
      
       
      <span style="">this</span>.brokerController.getBrokerConfig().getRegionId());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">35</span>:     response.addExtField(MessageConst.PROPERTY_TRACE_SWITCH, String.valueOf(
      
       
      <span style="">this</span>.brokerController.getBrokerConfig().isTraceOn()));
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">36</span>: 
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">37</span>:     
      
       
      <span style="">if</span> (log.isDebugEnabled()) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">38</span>:         log.debug(
      
       
      <span style="">"receive SendMessage request command, {}"</span>, request);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">39</span>:     }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">40</span>: 
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">41</span>:     
      
       
      <span style="">// 如果未开始接收消息，抛出系统异常</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">42</span>:     
      
       
      <span style="">@SuppressWarnings</span>(
      
       
      <span style="">"SpellCheckingInspection"</span>)
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">43</span>:     
      
       
      <span style="">final</span> 
      
       
      <span style="">long</span> startTimstamp = 
      
       
      <span style="">this</span>.brokerController.getBrokerConfig().getStartAcceptSendRequestTimeStamp();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">44</span>:     
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.brokerController.getMessageStore().now() &lt; startTimstamp) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">45</span>:         response.setCode(ResponseCode.SYSTEM_ERROR);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">46</span>:         response.setRemark(String.format(
      
       
      <span style="">"broker unable to service, until %s"</span>, UtilAll.timeMillisToHumanString2(startTimstamp)));
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">47</span>:         
      
       
      <span style="">return</span> response;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">48</span>:     }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">49</span>: 
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">50</span>:     
      
       
      <span style="">// 消息配置(Topic配置）校验</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">51</span>:     response.setCode(-
      
       
      <span style="">1</span>);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">52</span>:     
      
       
      <span style="">super</span>.msgCheck(ctx, requestHeader, response);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">53</span>:     
      
       
      <span style="">if</span> (response.getCode() != -
      
       
      <span style="">1</span>) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">54</span>:         
      
       
      <span style="">return</span> response;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">55</span>:     }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">56</span>: 
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">57</span>:     
      
       
      <span style="">final</span> 
      
       
      <span style="">byte</span>[] body = request.getBody();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">58</span>: 
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">59</span>:     
      
       
      <span style="">// 如果队列小于0，从可用队列随机选择</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">60</span>:     
      
       
      <span style="">int</span> queueIdInt = requestHeader.getQueueId();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">61</span>:     TopicConfig topicConfig = 
      
       
      <span style="">this</span>.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">62</span>:     
      
       
      <span style="">if</span> (queueIdInt &lt; 
      
       
      <span style="">0</span>) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">63</span>:         queueIdInt = Math.abs(
      
       
      <span style="">this</span>.random.nextInt() % 
      
       
      <span style="">99999999</span>) % topicConfig.getWriteQueueNums();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">64</span>:     }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">65</span>: 
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">66</span>:     
      
       
      <span style="">//</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">67</span>:     
      
       
      <span style="">int</span> sysFlag = requestHeader.getSysFlag();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">68</span>:     
      
       
      <span style="">if</span> (TopicFilterType.MULTI_TAG == topicConfig.getTopicFilterType()) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">69</span>:         sysFlag |= MessageSysFlag.MULTI_TAGS_FLAG;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">70</span>:     }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">71</span>: 
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">72</span>:     
      
       
      <span style="">// 对RETRY类型的消息处理。如果超过最大消费次数，则topic修改成"%DLQ%" + 分组名，即加入 死信队列(Dead Letter Queue)</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">73</span>:     String newTopic = requestHeader.getTopic();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">74</span>:     
      
       
      <span style="">if</span> (
      
       
      <span style="">null</span> != newTopic &amp;&amp; newTopic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">75</span>:         
      
       
      <span style="">// 获取订阅分组配置</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">76</span>:         String groupName = newTopic.substring(MixAll.RETRY_GROUP_TOPIC_PREFIX.length());
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">77</span>:         SubscriptionGroupConfig subscriptionGroupConfig =
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">78</span>:             
      
       
      <span style="">this</span>.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(groupName);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">79</span>:         
      
       
      <span style="">if</span> (
      
       
      <span style="">null</span> == subscriptionGroupConfig) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">80</span>:             response.setCode(ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">81</span>:             response.setRemark(
      
       
      <span style="">"subscription group not exist, "</span> + groupName + 
      
       
      <span style="">" "</span> + FAQUrl.suggestTodo(FAQUrl.SUBSCRIPTION_GROUP_NOT_EXIST));
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">82</span>:             
      
       
      <span style="">return</span> response;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">83</span>:         }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">84</span>:         
      
       
      <span style="">// 计算最大可消费次数</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">85</span>:         
      
       
      <span style="">int</span> maxReconsumeTimes = subscriptionGroupConfig.getRetryMaxTimes();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">86</span>:         
      
       
      <span style="">if</span> (request.getVersion() &gt;= MQVersion.Version.V3_4_9.ordinal()) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">87</span>:             maxReconsumeTimes = requestHeader.getMaxReconsumeTimes();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">88</span>:         }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">89</span>:         
      
       
      <span style="">int</span> reconsumeTimes = requestHeader.getReconsumeTimes() == 
      
       
      <span style="">null</span> ? 
      
       
      <span style="">0</span> : requestHeader.getReconsumeTimes();
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">90</span>:         
      
       
      <span style="">if</span> (reconsumeTimes &gt;= maxReconsumeTimes) { 
      
       
      <span style="">// 超过最大消费次数</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">91</span>:             newTopic = MixAll.getDLQTopic(groupName);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">92</span>:             queueIdInt = Math.abs(
      
       
      <span style="">this</span>.random.nextInt() % 
      
       
      <span style="">99999999</span>) % DLQ_NUMS_PER_GROUP;
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">93</span>:             topicConfig = 
      
       
      <span style="">this</span>.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(newTopic, 
      
       
      <span style="">//</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">94</span>:                 DLQ_NUMS_PER_GROUP, 
      
       
      <span style="">//</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">95</span>:                 PermName.PERM_WRITE, 
      
       
      <span style="">0</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">96</span>:             );
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">97</span>:             
      
       
      <span style="">if</span> (
      
       
      <span style="">null</span> == topicConfig) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">98</span>:                 response.setCode(ResponseCode.SYSTEM_ERROR);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">99</span>:                 response.setRemark(
      
       
      <span style="">"topic["</span> + newTopic + 
      
       
      <span style="">"] not exist"</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">100</span>:                 
      
       
      <span style="">return</span> response;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">101</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">102</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">103</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">104</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">105</span>:     
      
       
      <span style="">// 创建MessageExtBrokerInner</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">106</span>:     MessageExtBrokerInner msgInner = 
      
       
      <span style="">new</span> MessageExtBrokerInner();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">107</span>:     msgInner.setTopic(newTopic);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">108</span>:     msgInner.setBody(body);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">109</span>:     msgInner.setFlag(requestHeader.getFlag());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">110</span>:     MessageAccessor.setProperties(msgInner, MessageDecoder.string2messageProperties(requestHeader.getProperties()));
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">111</span>:     msgInner.setPropertiesString(requestHeader.getProperties());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">112</span>:     msgInner.setTagsCode(MessageExtBrokerInner.tagsString2tagsCode(topicConfig.getTopicFilterType(), msgInner.getTags()));
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">113</span>:     msgInner.setQueueId(queueIdInt);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">114</span>:     msgInner.setSysFlag(sysFlag);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">115</span>:     msgInner.setBornTimestamp(requestHeader.getBornTimestamp());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">116</span>:     msgInner.setBornHost(ctx.channel().remoteAddress());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">117</span>:     msgInner.setStoreHost(
      
       
      <span style="">this</span>.getStoreHost());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">118</span>:     msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == 
      
       
      <span style="">null</span> ? 
      
       
      <span style="">0</span> : requestHeader.getReconsumeTimes());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">119</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">120</span>:     
      
       
      <span style="">// 校验是否不允许发送事务消息</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">121</span>:     
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.brokerController.getBrokerConfig().isRejectTransactionMessage()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">122</span>:         String traFlag = msgInner.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">123</span>:         
      
       
      <span style="">if</span> (traFlag != 
      
       
      <span style="">null</span>) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">124</span>:             response.setCode(ResponseCode.NO_PERMISSION);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">125</span>:             response.setRemark(
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">126</span>:                 
      
       
      <span style="">"the broker["</span> + 
      
       
      <span style="">this</span>.brokerController.getBrokerConfig().getBrokerIP1() + 
      
       
      <span style="">"] sending transaction message is forbidden"</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">127</span>:             
      
       
      <span style="">return</span> response;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">128</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">129</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">130</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">131</span>:     
      
       
      <span style="">// 添加消息</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">132</span>:     PutMessageResult putMessageResult = 
      
       
      <span style="">this</span>.brokerController.getMessageStore().putMessage(msgInner);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">133</span>:     
      
       
      <span style="">if</span> (putMessageResult != 
      
       
      <span style="">null</span>) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">134</span>:         
      
       
      <span style="">boolean</span> sendOK = 
      
       
      <span style="">false</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">135</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">136</span>:         
      
       
      <span style="">switch</span> (putMessageResult.getPutMessageStatus()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">137</span>:             
      
       
      <span style="">// Success</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">138</span>:             
      
       
      <span style="">case</span> PUT_OK:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">139</span>:                 sendOK = 
      
       
      <span style="">true</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">140</span>:                 response.setCode(ResponseCode.SUCCESS);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">141</span>:                 
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">142</span>:             
      
       
      <span style="">case</span> FLUSH_DISK_TIMEOUT:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">143</span>:                 response.setCode(ResponseCode.FLUSH_DISK_TIMEOUT);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">144</span>:                 sendOK = 
      
       
      <span style="">true</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">145</span>:                 
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">146</span>:             
      
       
      <span style="">case</span> FLUSH_SLAVE_TIMEOUT:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">147</span>:                 response.setCode(ResponseCode.FLUSH_SLAVE_TIMEOUT);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">148</span>:                 sendOK = 
      
       
      <span style="">true</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">149</span>:                 
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">150</span>:             
      
       
      <span style="">case</span> SLAVE_NOT_AVAILABLE:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">151</span>:                 response.setCode(ResponseCode.SLAVE_NOT_AVAILABLE);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">152</span>:                 sendOK = 
      
       
      <span style="">true</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">153</span>:                 
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">154</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">155</span>:             
      
       
      <span style="">// Failed</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">156</span>:             
      
       
      <span style="">case</span> CREATE_MAPEDFILE_FAILED:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">157</span>:                 response.setCode(ResponseCode.SYSTEM_ERROR);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">158</span>:                 response.setRemark(
      
       
      <span style="">"create mapped file failed, server is busy or broken."</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">159</span>:                 
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">160</span>:             
      
       
      <span style="">case</span> MESSAGE_ILLEGAL:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">161</span>:             
      
       
      <span style="">case</span> PROPERTIES_SIZE_EXCEEDED:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">162</span>:                 response.setCode(ResponseCode.MESSAGE_ILLEGAL);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">163</span>:                 response.setRemark(
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">164</span>:                     
      
       
      <span style="">"the message is illegal, maybe msg body or properties length not matched. msg body length limit 128k, msg properties length limit 32k."</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">165</span>:                 
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">166</span>:             
      
       
      <span style="">case</span> SERVICE_NOT_AVAILABLE:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">167</span>:                 response.setCode(ResponseCode.SERVICE_NOT_AVAILABLE);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">168</span>:                 response.setRemark(
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">169</span>:                     
      
       
      <span style="">"service not available now, maybe disk full, "</span> + diskUtil() + 
      
       
      <span style="">", maybe your broker machine memory too small."</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">170</span>:                 
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">171</span>:             
      
       
      <span style="">case</span> OS_PAGECACHE_BUSY:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">172</span>:                 response.setCode(ResponseCode.SYSTEM_ERROR);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">173</span>:                 response.setRemark(
      
       
      <span style="">"[PC_SYNCHRONIZED]broker busy, start flow control for a while"</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">174</span>:                 
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">175</span>:             
      
       
      <span style="">case</span> UNKNOWN_ERROR:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">176</span>:                 response.setCode(ResponseCode.SYSTEM_ERROR);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">177</span>:                 response.setRemark(
      
       
      <span style="">"UNKNOWN_ERROR"</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">178</span>:                 
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">179</span>:             
      
       
      <span style="">default</span>:
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">180</span>:                 response.setCode(ResponseCode.SYSTEM_ERROR);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">181</span>:                 response.setRemark(
      
       
      <span style="">"UNKNOWN_ERROR DEFAULT"</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">182</span>:                 
      
       
      <span style="">break</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">183</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">184</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">185</span>:         String owner = request.getExtFields().get(BrokerStatsManager.COMMERCIAL_OWNER);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">186</span>:         
      
       
      <span style="">if</span> (sendOK) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">187</span>:             
      
       
      <span style="">// 统计</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">188</span>:             
      
       
      <span style="">this</span>.brokerController.getBrokerStatsManager().incTopicPutNums(msgInner.getTopic());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">189</span>:             
      
       
      <span style="">this</span>.brokerController.getBrokerStatsManager().incTopicPutSize(msgInner.getTopic(), putMessageResult.getAppendMessageResult().getWroteBytes());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">190</span>:             
      
       
      <span style="">this</span>.brokerController.getBrokerStatsManager().incBrokerPutNums();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">191</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">192</span>:             
      
       
      <span style="">// 响应</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">193</span>:             response.setRemark(
      
       
      <span style="">null</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">194</span>:             responseHeader.setMsgId(putMessageResult.getAppendMessageResult().getMsgId());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">195</span>:             responseHeader.setQueueId(queueIdInt);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">196</span>:             responseHeader.setQueueOffset(putMessageResult.getAppendMessageResult().getLogicsOffset());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">197</span>:             doResponse(ctx, request, response);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">198</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">199</span>:             
      
       
      <span style="">// hook：设置发送成功到context</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">200</span>:             
      
       
      <span style="">if</span> (hasSendMessageHook()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">201</span>:                 sendMessageContext.setMsgId(responseHeader.getMsgId());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">202</span>:                 sendMessageContext.setQueueId(responseHeader.getQueueId());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">203</span>:                 sendMessageContext.setQueueOffset(responseHeader.getQueueOffset());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">204</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">205</span>:                 
      
       
      <span style="">int</span> commercialBaseCount = brokerController.getBrokerConfig().getCommercialBaseCount();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">206</span>:                 
      
       
      <span style="">int</span> wroteSize = putMessageResult.getAppendMessageResult().getWroteBytes();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">207</span>:                 
      
       
      <span style="">int</span> incValue = (
      
       
      <span style="">int</span>) Math.ceil(wroteSize / BrokerStatsManager.SIZE_PER_COUNT) * commercialBaseCount;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">208</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">209</span>:                 sendMessageContext.setCommercialSendStats(BrokerStatsManager.StatsType.SEND_SUCCESS);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">210</span>:                 sendMessageContext.setCommercialSendTimes(incValue);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">211</span>:                 sendMessageContext.setCommercialSendSize(wroteSize);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">212</span>:                 sendMessageContext.setCommercialOwner(owner);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">213</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">214</span>:             
      
       
      <span style="">return</span> 
      
       
      <span style="">null</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">215</span>:         } 
      
       
      <span style="">else</span> {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">216</span>:             
      
       
      <span style="">// hook：设置发送失败到context</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">217</span>:             
      
       
      <span style="">if</span> (hasSendMessageHook()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">218</span>:                 
      
       
      <span style="">int</span> wroteSize = request.getBody().length;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">219</span>:                 
      
       
      <span style="">int</span> incValue = (
      
       
      <span style="">int</span>) Math.ceil(wroteSize / BrokerStatsManager.SIZE_PER_COUNT);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">220</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">221</span>:                 sendMessageContext.setCommercialSendStats(BrokerStatsManager.StatsType.SEND_FAILURE);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">222</span>:                 sendMessageContext.setCommercialSendTimes(incValue);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">223</span>:                 sendMessageContext.setCommercialSendSize(wroteSize);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">224</span>:                 sendMessageContext.setCommercialOwner(owner);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">225</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">226</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">227</span>:     } 
      
       
      <span style="">else</span> {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">228</span>:         response.setCode(ResponseCode.SYSTEM_ERROR);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">229</span>:         response.setRemark(
      
       
      <span style="">"store putMessage return null"</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">230</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">231</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">232</span>:     
      
       
      <span style="">return</span> response;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">233</span>: }
     
      
     </div></pre> </td> 
  </tr> 
 </tbody> 
</table>

*  `#processRequest()` 说明 ：处理消息请求。
 *  `#sendMessage()` 说明 ：发送消息，并返回发送消息结果。
 *  第 51 至 55 行 ：消息配置(Topic配置）校验，详细解析见：[AbstractSendMessageProcessor\#msgCheck()][AbstractSendMessageProcessor_msgCheck]。
 *  第 60 至 64 行 ：消息队列编号小于0时，`Broker` 可以设置随机选择一个消息队列。
 *  第 72 至 103 行 ：对RETRY类型的消息处理。如果超过最大消费次数，则topic修改成”%DLQ%” + 分组名， 即加 死信队 (Dead Letter Queue)，详细解析见：[《RocketMQ 源码分析 —— Topic》][RocketMQ _ _ Topic]。
 *  第 105 至 118 行 ：创建`MessageExtBrokerInner`。
 *  第 132 ：存储消息，详细解析见：[DefaultMessageStore\#putMessage()][DefaultMessageStore_putMessage]。
 *  第 133 至 183 行 ：处理消息发送结果，设置响应结果和提示。
 *  第 186 至 214 行 ：发送成功，响应。这里`doResponse(ctx, request, response)`进行响应，最后`return null`，原因是：响应给 `Producer` 可能发生异常，`#doResponse(ctx, request, response)`捕捉了该异常并输出日志。这样做的话，我们进行排查 `Broker` 接收消息成功后响应是否存在异常会方便很多。

### AbstractSendMessageProcessor\#msgCheck ###

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style=""> <pre style="">
     
      
     <div style=""> 
      
       
      <span style="">1</span>: 
      
       
      <span style=""><span style="">protected</span> RemotingCommand <span style="">msgCheck</span><span style="">(<span style="">final</span> ChannelHandlerContext ctx,</span></span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style=""><span style=""> <span style="">2</span>: <span style="">final</span> SendMessageRequestHeader requestHeader, <span style="">final</span> RemotingCommand response)</span> </span>{
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">3</span>:     
      
       
      <span style="">// 检查 broker 是否有写入权限</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">4</span>:     
      
       
      <span style="">if</span> (!PermName.isWriteable(
      
       
      <span style="">this</span>.brokerController.getBrokerConfig().getBrokerPermission())
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">5</span>:         &amp;&amp; 
      
       
      <span style="">this</span>.brokerController.getTopicConfigManager().isOrderTopic(requestHeader.getTopic())) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">6</span>:         response.setCode(ResponseCode.NO_PERMISSION);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">7</span>:         response.setRemark(
      
       
      <span style="">"the broker["</span> + 
      
       
      <span style="">this</span>.brokerController.getBrokerConfig().getBrokerIP1()
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">8</span>:             + 
      
       
      <span style="">"] sending message is forbidden"</span>);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">9</span>:         
      
       
      <span style="">return</span> response;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">10</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">11</span>:     
      
       
      <span style="">// 检查topic是否可以被发送。目前是{@link MixAll.DEFAULT_TOPIC}不被允许发送</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">12</span>:     
      
       
      <span style="">if</span> (!
      
       
      <span style="">this</span>.brokerController.getTopicConfigManager().isTopicCanSendMessage(requestHeader.getTopic())) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">13</span>:         String errorMsg = 
      
       
      <span style="">"the topic["</span> + requestHeader.getTopic() + 
      
       
      <span style="">"] is conflict with system reserved words."</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">14</span>:         log.warn(errorMsg);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">15</span>:         response.setCode(ResponseCode.SYSTEM_ERROR);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">16</span>:         response.setRemark(errorMsg);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">17</span>:         
      
       
      <span style="">return</span> response;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">18</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">19</span>:     TopicConfig topicConfig = 
      
       
      <span style="">this</span>.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">20</span>:     
      
       
      <span style="">if</span> (
      
       
      <span style="">null</span> == topicConfig) { 
      
       
      <span style="">// 不能存在topicConfig，则进行创建</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">21</span>:         
      
       
      <span style="">int</span> topicSysFlag = 
      
       
      <span style="">0</span>;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">22</span>:         
      
       
      <span style="">if</span> (requestHeader.isUnitMode()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">23</span>:             
      
       
      <span style="">if</span> (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">24</span>:                 topicSysFlag = TopicSysFlag.buildSysFlag(
      
       
      <span style="">false</span>, 
      
       
      <span style="">true</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">25</span>:             } 
      
       
      <span style="">else</span> {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">26</span>:                 topicSysFlag = TopicSysFlag.buildSysFlag(
      
       
      <span style="">true</span>, 
      
       
      <span style="">false</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">27</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">28</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">29</span>:         
      
       
      <span style="">// 创建topic配置</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">30</span>:         log.warn(
      
       
      <span style="">"the topic {} not exist, producer: {}"</span>, requestHeader.getTopic(), ctx.channel().remoteAddress());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">31</span>:         topicConfig = 
      
       
      <span style="">this</span>.brokerController.getTopicConfigManager().createTopicInSendMessageMethod(
      
       
      <span style="">//</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">32</span>:             requestHeader.getTopic(), 
      
       
      <span style="">//</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">33</span>:             requestHeader.getDefaultTopic(), 
      
       
      <span style="">//</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">34</span>:             RemotingHelper.parseChannelRemoteAddr(ctx.channel()), 
      
       
      <span style="">//</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">35</span>:             requestHeader.getDefaultTopicQueueNums(), topicSysFlag);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">36</span>:         
      
       
      <span style="">if</span> (
      
       
      <span style="">null</span> == topicConfig) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">37</span>:             
      
       
      <span style="">if</span> (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">38</span>:                 topicConfig =
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">39</span>:                     
      
       
      <span style="">this</span>.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">40</span>:                         requestHeader.getTopic(), 
      
       
      <span style="">1</span>, PermName.PERM_WRITE | PermName.PERM_READ,
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">41</span>:                         topicSysFlag);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">42</span>:             }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">43</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">44</span>:         
      
       
      <span style="">// 如果没配置</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">45</span>:         
      
       
      <span style="">if</span> (
      
       
      <span style="">null</span> == topicConfig) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">46</span>:             response.setCode(ResponseCode.TOPIC_NOT_EXIST);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">47</span>:             response.setRemark(
      
       
      <span style="">"topic["</span> + requestHeader.getTopic() + 
      
       
      <span style="">"] not exist, apply first please!"</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">48</span>:                 + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">49</span>:             
      
       
      <span style="">return</span> response;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">50</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">51</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">52</span>:     
      
       
      <span style="">// 队列编号是否正确</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">53</span>:     
      
       
      <span style="">int</span> queueIdInt = requestHeader.getQueueId();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">54</span>:     
      
       
      <span style="">int</span> idValid = Math.max(topicConfig.getWriteQueueNums(), topicConfig.getReadQueueNums());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">55</span>:     
      
       
      <span style="">if</span> (queueIdInt &gt;= idValid) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">56</span>:         String errorInfo = String.format(
      
       
      <span style="">"request queueId[%d] is illegal, %s Producer: %s"</span>,
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">57</span>:             queueIdInt,
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">58</span>:             topicConfig.toString(),
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">59</span>:             RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">60</span>:         log.warn(errorInfo);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">61</span>:         response.setCode(ResponseCode.SYSTEM_ERROR);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">62</span>:         response.setRemark(errorInfo);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">63</span>:         
      
       
      <span style="">return</span> response;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">64</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">65</span>:     
      
       
      <span style="">return</span> response;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">66</span>: }
     
      
     </div></pre> </td> 
  </tr> 
 </tbody> 
</table>

*  说明：校验消息是否正确，主要是Topic配置方面，例如：`Broker` 是否有写入权限，topic配置是否存在，队列编号是否正确。
 *  第 11 至 18 行 ：检查Topic是否可以被发送。目前是 `{@link MixAll.DEFAULT_TOPIC}` 不被允许发送。
 *  第 20 至 51 行 ：当找不到Topic配置，则进行创建。当然，创建会存在不成功的情况，例如说：`defaultTopic` 的Topic配置不存在，又或者是 存在但是不允许继承，详细解析见[《RocketMQ 源码分析 —— Topic》][RocketMQ _ _ Topic]。

## DefaultMessageStore\#putMessage ##

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style=""> <pre style="">
     
      
     <div style=""> 
      
       
      <span style="">1</span>: 
      
       
      <span style=""><span style="">public</span> PutMessageResult <span style="">putMessage</span><span style="">(MessageExtBrokerInner msg)</span> </span>{
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">2</span>:     
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.shutdown) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">3</span>:         log.warn(
      
       
      <span style="">"message store has shutdown, so putMessage is forbidden"</span>);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">4</span>:         
      
       
      <span style="">return</span> 
      
       
      <span style="">new</span> PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, 
      
       
      <span style="">null</span>);
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">5</span>:     }
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">6</span>: 
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">7</span>:     
      
       
      <span style="">// 从节点不允许写入</span>
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">8</span>:     
      
       
      <span style="">if</span> (BrokerRole.SLAVE == 
      
       
      <span style="">this</span>.messageStoreConfig.getBrokerRole()) {
     
      
     </div>
     
      
     <div style=""> 
      
       
      <span style="">9</span>:         
      
       
      <span style="">long</span> value = 
      
       
      <span style="">this</span>.printTimes.getAndIncrement();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">10</span>:         
      
       
      <span style="">if</span> ((value % 
      
       
      <span style="">50000</span>) == 
      
       
      <span style="">0</span>) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">11</span>:             log.warn(
      
       
      <span style="">"message store is slave mode, so putMessage is forbidden "</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">12</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">13</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">14</span>:         
      
       
      <span style="">return</span> 
      
       
      <span style="">new</span> PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, 
      
       
      <span style="">null</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">15</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">16</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">17</span>:     
      
       
      <span style="">// store是否允许写入</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">18</span>:     
      
       
      <span style="">if</span> (!
      
       
      <span style="">this</span>.runningFlags.isWriteable()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">19</span>:         
      
       
      <span style="">long</span> value = 
      
       
      <span style="">this</span>.printTimes.getAndIncrement();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">20</span>:         
      
       
      <span style="">if</span> ((value % 
      
       
      <span style="">50000</span>) == 
      
       
      <span style="">0</span>) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">21</span>:             log.warn(
      
       
      <span style="">"message store is not writeable, so putMessage is forbidden "</span> + 
      
       
      <span style="">this</span>.runningFlags.getFlagBits());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">22</span>:         }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">23</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">24</span>:         
      
       
      <span style="">return</span> 
      
       
      <span style="">new</span> PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, 
      
       
      <span style="">null</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">25</span>:     } 
      
       
      <span style="">else</span> {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">26</span>:         
      
       
      <span style="">this</span>.printTimes.set(
      
       
      <span style="">0</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">27</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">28</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">29</span>:     
      
       
      <span style="">// 消息过长</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">30</span>:     
      
       
      <span style="">if</span> (msg.getTopic().length() &gt; Byte.MAX_VALUE) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">31</span>:         log.warn(
      
       
      <span style="">"putMessage message topic length too long "</span> + msg.getTopic().length());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">32</span>:         
      
       
      <span style="">return</span> 
      
       
      <span style="">new</span> PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, 
      
       
      <span style="">null</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">33</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">34</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">35</span>:     
      
       
      <span style="">// 消息附加属性过长</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">36</span>:     
      
       
      <span style="">if</span> (msg.getPropertiesString() != 
      
       
      <span style="">null</span> &amp;&amp; msg.getPropertiesString().length() &gt; Short.MAX_VALUE) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">37</span>:         log.warn(
      
       
      <span style="">"putMessage message properties length too long "</span> + msg.getPropertiesString().length());
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">38</span>:         
      
       
      <span style="">return</span> 
      
       
      <span style="">new</span> PutMessageResult(PutMessageStatus.PROPERTIES_SIZE_EXCEEDED, 
      
       
      <span style="">null</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">39</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">40</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">41</span>:     
      
       
      <span style="">if</span> (
      
       
      <span style="">this</span>.isOSPageCacheBusy()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">42</span>:         
      
       
      <span style="">return</span> 
      
       
      <span style="">new</span> PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, 
      
       
      <span style="">null</span>);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">43</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">44</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">45</span>:     
      
       
      <span style="">long</span> beginTime = 
      
       
      <span style="">this</span>.getSystemClock().now();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">46</span>:     
      
       
      <span style="">// 添加消息到commitLog</span>
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">47</span>:     PutMessageResult result = 
      
       
      <span style="">this</span>.commitLog.putMessage(msg);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">48</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">49</span>:     
      
       
      <span style="">long</span> eclipseTime = 
      
       
      <span style="">this</span>.getSystemClock().now() - beginTime;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">50</span>:     
      
       
      <span style="">if</span> (eclipseTime &gt; 
      
       
      <span style="">500</span>) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">51</span>:         log.warn(
      
       
      <span style="">"putMessage not in lock eclipse time(ms)={}, bodyLength={}"</span>, eclipseTime, msg.getBody().length);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">52</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">53</span>:     
      
       
      <span style="">this</span>.storeStatsService.setPutMessageEntireTimeMax(eclipseTime);
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">54</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">55</span>:     
      
       
      <span style="">if</span> (
      
       
      <span style="">null</span> == result || !result.isOk()) {
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">56</span>:         
      
       
      <span style="">this</span>.storeStatsService.getPutMessageFailedTimes().incrementAndGet();
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">57</span>:     }
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">58</span>: 
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">59</span>:     
      
       
      <span style="">return</span> result;
     
      
     </div>
     
      
     <div style="">
      
       
      <span style="">60</span>: }
     
      
     </div></pre> </td> 
  </tr> 
 </tbody> 
</table>

*  说明：存储消息封装，最终存储需要 `CommitLog` 实现。
 *  第 7 至 27 行 ：校验 `Broker` 是否可以写入。
 *  第 29 至 39 行 ：消息格式与大小校验。
 *  第 47 行 ：调用 `CommitLong` 进行存储，详细逻辑见：[《RocketMQ 源码分析 —— Message 存储》][RocketMQ _ _ Message]

[1]: http://www.iocoder.cn/RocketMQ/message-send-and-receive/#
[RocketMQ _ _ Message]: http://www.iocoder.cn/RocketMQ/message-store/
[Producer]: /images/20220608/3d1873155742421e81a4a6a017cfe267.png
[Producer 1]: /images/20220608/46476f2cf54b423bbcd6f75f9850d253.png
[DefaultMQProducerImpl_tryToFindTopicPublishInfo]: http://www.iocoder.cn/RocketMQ/message-send-and-receive/#defaultmqproducerimpltrytofindtopicpublishinfo
[MQFaultStrategy]: http://www.iocoder.cn/RocketMQ/message-send-and-receive/#mqfaultstrategy
[DefaultMQProducerImpl_sendKernelImpl]: http://www.iocoder.cn/RocketMQ/message-send-and-receive/#defaultmqproducerimplsendkernelimpl
[RocketMQ _ _ Topic]: http://www.iocoder.cn/RocketMQ/topic/
[Latency]: /images/20220608/b509265fb5214742b7e67f92510f3a0f.png
[RocketMQ _ _ Message 1]: http://www.iocoder.cn/RocketMQ/message/
[API]: /images/20220608/0b969763f28e44cdaa551131be612d1e.png
[AbstractSendMessageProcessor_msgCheck]: http://www.iocoder.cn/RocketMQ/message-send-and-receive/#abstractsendmessageprocessormsgcheck
[DefaultMessageStore_putMessage]: http://www.iocoder.cn/RocketMQ/message-send-and-receive/defaultmessagestoreputmessage