1. 消息发送介绍

RocketMQ提供三种消息发送方式，分别是同步（Syn）、异步（Asyn）和单向（one way）

  1. 同步：生产者发送消息时，同步等待消息服务器发送结果
  2. 异步：生产者发送消息时，指定发送成功后的回调函数，生产者调用消息发送接口时立即返回结果，消息发送线程不阻塞。消息发送成功或者失败时回调函数通过一个新的线程来执行

  3. 单向：生产者发送消息时直接返回，不等待消息服务器的结果，也不注册回调函数。简单概括为只管消息发送，不管消息发送结果

    1.1.Topic 路由机制

    我们所说的消息发送，其实就是将消息发送至某个topic，在初次发送时会根据topic的名称向Nameserver 集群查询topic的路由信息，然后将路由信息存储在生产者本地缓存中，然后每隔30s遍历本地缓存中的topic，向Nameserver 查询最新的路由信息。如果成功查到路由信息，会将这些信息更新至本地缓存，实现topic 路由信息的动态感知
    RocketMQ 还提供自动创建主题（Topic） 机制，如果topic不存在，并且开启了自动创建主题机制配置，那么会使用默认的路由信息进行路由 消息发送 - 图1

    1.2. 消息发送高可用

    生产者在自动发现主题的路由信息后，RocketMQ 默认使用轮询算法进行路由的负载均衡。RocketMQ 在消息发送时也支持自定义队列负载均衡，但是使用自定义路由负债时重试机制会失效
    RocketMQ 引入两个重要的特点来保证消息的高可用

  4. 消息发送重试机制

RocketMQ 在消息发送时如果出现失败，默认会重试两次

  1. 故障规避机制

当消息第一次发送失败时，如果下一次消息还是发送到刚刚失败的Broker上，其消息发送失败的概率会很大，为了保证重试的可靠性，在重试时会尽量避开刚刚接收失败的Broker，而是选择其他Broker上的队列进行发送来提高消息发送的成功率
RocketMQ-第 3 页.jpg

2. 生产者启动

消息生产者代码都在client 模块中，对于RocketMQ来说它既是消息客户端，也是消息生产者。

2.1. 时序图

消息发送 - 图3

2.2. 源码分析

生产者启动入口在org.apache.rocketmq.client.producer.DefaultMQProducer.start() 方法

  1.  @Override
  2. public void start() throws MQClientException {
  3.     this.setProducerGroup(withNamespace(this.producerGroup));
  4.     // 调用DefaultMQProducerImpl 对象的start()方法启动生产者
  5.     this.defaultMQProducerImpl.start();
  6.     if (null != traceDispatcher) {
  7.         try {
  8.             traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
  9.         } catch (MQClientException e) {
  10.             log.warn("trace dispatcher start failed ", e);
  11.         }
  12.     }
  13. }

进入DefaultMQProducerImpl 类的start()方法，启动主要分为四个部分

  1. 检查producerGroup 是否符合要求
  2. 创建MQClientInstance 实例
  3. 向MQClientInstance 注册服务

  4. 启动MQClientInstance ```java public void start(final boolean startFactory) throws MQClientException { switch (this.serviceState) {

    1.  case CREATE_JUST:
    2.      this.serviceState = ServiceState.START_FAILED;
    3.      // Step1: 检查producerGroup 是否符合要求，改变生产者的instanceName为进程ID
    4.      this.checkConfig();
    5.      // 如果生产者的实例名称为默认DEFAULT，那么将实例名称替换为进程ID
    6.      // 这么做的目的是解决在同一个物理机上部署多个生产者应用导致clientId 混乱（重复）问题
    7.      if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
    8.          this.defaultMQProducer.changeInstanceNameToPID();
    9.      }
    10.      // Step2:创建MQClientInstance 实例
    11.      // 在一个JVM中，MQClientInstance 实例只存在一个
    12.      this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);
    13.      // Step3:向MQClientInstance 注册服务，将当前生产者加入MQClientInstance 管理，便于后续调用网络请求、心跳检测等
    14.      boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
    15.      if (!registerOK) {
    16.          this.serviceState = ServiceState.CREATE_JUST;
    17.          throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
    18.                                      + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
    19.                                      null);
    20.      }
    21.      // 默认TopicKey
    22.      this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
    23.      // Step4: 启动MQClientInstance，如果MQClientInstanec 已经启动，本次启动不会去执行
    24.      if (startFactory) {
    25.          mQClientFactory.start();
    26.      }
    28.      log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
    29.               this.defaultMQProducer.isSendMessageWithVIPChannel());
    30.      this.serviceState = ServiceState.RUNNING;
    31.      break;
    32.  case RUNNING:
    33.  case START_FAILED:
    34.  case SHUTDOWN_ALREADY:
    35.      throw new MQClientException("The producer service state not OK, maybe started once, "
    36.                                  + this.serviceState
    37.                                  + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
    38.                                  null);
    39.  default:
    40.      break;

    }

    this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();

    RequestFutureHolder.getInstance().startScheduledTask(this);

}

  2. <a name="mBl2v"></a>
  3. ### 2.2.1. 检查producerGroup
  4. 检查producerGroup 是否符合要求，改变生产者的instanceName为进程ID
  5. ```java
  6. private void checkConfig() throws MQClientException {
  7.     // 检查producerGroup
  8.     Validators.checkGroup(this.defaultMQProducer.getProducerGroup());
  10.     if (null == this.defaultMQProducer.getProducerGroup()) {
  11.         throw new MQClientException("producerGroup is null", null);
  12.     }
  14.     if (this.defaultMQProducer.getProducerGroup().equals(MixAll.DEFAULT_PRODUCER_GROUP)) {
  15.         throw new MQClientException("producerGroup can not equal " + MixAll.DEFAULT_PRODUCER_GROUP + ", please specify another one.",
  16.                                     null);
  17.     }
  18. }

producerGroup 检查最终交给Validators.checkGroup()方法校验，主要对producerGroup做了非空和长度校验

  1. public static void checkGroup(String group) throws MQClientException {
  2.     // 非空校验
  3.     if (UtilAll.isBlank(group)) {
  4.         throw new MQClientException("the specified group is blank", null);
  5.     }
  6.     // 长度校验
  7.     if (group.length() > CHARACTER_MAX_LENGTH) {
  8.         throw new MQClientException("the specified group is longer than group max length 255.", null);
  9.     }
  12.     if (isTopicOrGroupIllegal(group)) {
  13.         throw new MQClientException(String.format(
  14.             "the specified group[%s] contains illegal characters, allowing only %s", group,
  15.             "^[%|a-zA-Z0-9_-]+$"), null);
  16.     }
  17. }

如果生产者的实例名称为默认DEFAULT，那么将实例名称替换为进程ID，这么做的目的是解决在同一个物理机上部署多个生产者应用导致clientId 混乱（重复）问题

  1.  if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
  2.      this.defaultMQProducer.changeInstanceNameToPID();
  3.  }

替换实例名称为进程ID

  1. public void changeInstanceNameToPID() {
  2.     if (this.instanceName.equals("DEFAULT")) {
  3.         this.instanceName = UtilAll.getPid() + "#" + System.nanoTime();
  4.     }
  5. }

2.2.2. 创建MQClientInstance实例

MQClientInstance 实例的创建在MQClientManager.getOrCreateMQClientInstance()方法中创建
在一个JVM中，RocketMQ只允许一个MQClientInstance 实例存在。在MQClientManager类中维护一个缓存表Map结构，名称为factoryTable ，key=clientId，value=MQClientInstance。

  1. public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) {
  2.     // 创建clientId
  3.     String clientId = clientConfig.buildMQClientId();
  4.     // 根据clientId从本地缓存factoryTable中获取MQClientInstance
  5.     // 确保一个clientId 只会创建一个实例
  6.     MQClientInstance instance = this.factoryTable.get(clientId);
  7.     // 为空说明还没有创建MQClientInstance 实例
  8.     if (null == instance) {
  9.         // 创建MqClientInstance 实例
  10.         instance =
  11.             new MQClientInstance(clientConfig.cloneClientConfig(),
  12.                                  this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
  13.         // 将MQClientInstance 实例加入factoryTable 缓存中
  14.         MQClientInstance prev = this.factoryTable.putIfAbsent(clientId, instance);
  15.         if (prev != null) {
  16.             instance = prev;
  17.             log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId);
  18.         } else {
  19.             log.info("Created new MQClientInstance for clientId:[{}]", clientId);
  20.         }
  21.     }
  23.     return instance;
  24. }

clientid 的生成规则，ip+@+生产者实例名称+unitname(可选）。
注意：如果在同一个物理机上部署两个生产者应用程序，clientId 相同，为了避免造成这样的混乱
解决方案如果实例名称（instance）为默认值DEFAULT，那么会将instance设置为进程ID来避免不同进程相互影响的问题。

  1. public String buildMQClientId() {
  2.     StringBuilder sb = new StringBuilder();
  3.     sb.append(this.getClientIP());
  5.     sb.append("@");
  6.     sb.append(this.getInstanceName());
  7.     if (!UtilAll.isBlank(this.unitName)) {
  8.         sb.append("@");
  9.         sb.append(this.unitName);
  10.     }
  12.     return sb.toString();
  13. }

2.2.3. 向MQClientInstance 注册服务

创建号MQClientInstance 实例后，接下来就是将当前生产者加入MQClientInstance 来管理，方便后续调用网络请求、心跳检测等

  1. public synchronized boolean registerProducer(final String group, final DefaultMQProducerImpl producer) {
  2.     if (null == group || null == producer) {
  3.         return false;
  4.     }
  5.     // 将生产者对象加入producerTable缓存表中（Map结构），key=生产者名称，value=生产者实例
  6.     MQProducerInner prev = this.producerTable.putIfAbsent(group, producer);
  7.     if (prev != null) {
  8.         log.warn("the producer group[{}] exist already.", group);
  9.         return false;
  10.     }
  12.     return true;
  13. }

2.2.4. 启动MQClientInstance

  1. public void start() throws MQClientException {
  3.     synchronized (this) {
  4.         // 只有serviceState没有启动时才创建，否则不会创建
  5.         switch (this.serviceState) {
  6.             case CREATE_JUST:
  7.                 this.serviceState = ServiceState.START_FAILED;
  8.                 // If not specified,looking address from name server
  9.                 if (null == this.clientConfig.getNamesrvAddr()) {
  10.                     this.mQClientAPIImpl.fetchNameServerAddr();
  11.                 }
  12.                 // Start request-response channel
  13.                 this.mQClientAPIImpl.start();
  14.                 // Start various schedule tasks
  15.                 this.startScheduledTask();
  16.                 // Start pull service
  17.                 this.pullMessageService.start();
  18.                 // Start rebalance service
  19.                 this.rebalanceService.start();
  20.                 // Start push service
  21.                 this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
  22.                 log.info("the client factory [{}] start OK", this.clientId);
  23.                 this.serviceState = ServiceState.RUNNING;
  24.                 break;
  25.             case START_FAILED:
  26.                 throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
  27.             default:
  28.                 break;
  29.         }
  30.     }
  31. }

2.2.5. 生产者状态

消息生产者一共有四个状态，分别是创建、运行、关闭、失败四个状态，默认为创建状态，在调用org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.start()方法启动时，状态首先改变为START_FAILED状态，当生产者启动完成之后状态改变为RUNNING，在调用org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.shutdown()方法时，状态改变为SHUTDOWN_ALREADY 状态

  1. public enum ServiceState {
  2.     /**
  3.      * Service just created,not start
  4.      */
  5.     CREATE_JUST,
  6.     /**
  7.      * Service Running
  8.      */
  9.     RUNNING,
  10.     /**
  11.      * Service shutdown
  12.      */
  13.     SHUTDOWN_ALREADY,
  14.     /**
  15.      * Service Start failure
  16.      */
  17.     START_FAILED;
  18. }

3. 消息发送流程

创建完DefaultMQProducer对象后，接着我们就可以调用send()方法来发送消息，在MQProducer接口中提供了很多send()方法来发送消息

  1. lt send(final Message msg, final long timeout) throws MQClientException,
  2.         RemotingException, MQBrokerException, InterruptedException;
  4.     void send(final Message msg, final SendCallback sendCallback) throws MQClientException,
  5.         RemotingException, InterruptedException;
  7.     void send(final Message msg, final SendCallback sendCallback, final long timeout)
  8.         throws MQClientException, RemotingException, InterruptedException;
  10.     void sendOneway(final Message msg) throws MQClientException, RemotingException,
  11.         InterruptedException;
  13.     SendResult send(final Message msg, final MessageQueue mq) throws MQClientException,
  14.         RemotingException, MQBrokerException, InterruptedException;
  16.     SendResult send(final Message msg, final MessageQueue mq, final long timeout)
  17.         throws MQClientException, RemotingException, MQBrokerException, InterruptedException;
  19.     void send(final Message msg, final MessageQueue mq, final SendCallback sendCallback)
  20.         throws MQClientException, RemotingException, InterruptedException;
  22.     void send(final Message msg, final MessageQueue mq, final SendCallback sendCallback, long timeout)
  23.         throws MQClientException, RemotingException, InterruptedException;
  25.     SendResult send(final Message msg, final MessageQueueSelector selector, final Object arg)
  26.         throws MQClientException, RemotingException, MQBrokerException, InterruptedException;
  28.     SendResult send(final Message msg, final MessageQueueSelector selector, final Object arg,
  29.         final long timeout) throws MQClientException, RemotingException, MQBrokerException,
  30.         InterruptedException;
  32.     void send(final Message msg, final MessageQueueSelector selector, final Object arg,
  33.         final SendCallback sendCallback) throws MQClientException, RemotingException,
  34.         InterruptedException;
  36.     void send(final Message msg, final MessageQueueSelector selector, final Object arg,
  37.         final SendCallback sendCallback, final long timeout) throws MQClientException, RemotingException,
  38.         InterruptedException;
  40.     //for batch
  41.     SendResult send(final Collection<Message> msgs) throws MQClientException, RemotingException, MQBrokerException,
  42.         InterruptedException;
  44.     SendResult send(final Collection<Message> msgs, final long timeout) throws MQClientException,
  45.         RemotingException, MQBrokerException, InterruptedException;
  47.     SendResult send(final Collection<Message> msgs, final MessageQueue mq) throws MQClientException,
  48.         RemotingException, MQBrokerException, InterruptedException;
  50.     SendResult send(final Collection<Message> msgs, final MessageQueue mq, final long timeout)
  51.         throws MQClientException, RemotingException, MQBrokerException, InterruptedException;
  53.     void send(final Collection<Message> msgs, final SendCallback sendCallback) throws MQClientException, RemotingException, MQBrokerException,
  54.         InterruptedException;
  56.     void send(final Collection<Message> msgs, final SendCallback sendCallback, final long timeout) throws MQClientException, RemotingException,
  57.         MQBrokerException, InterruptedException;
  59.     void send(final Collection<Message> msgs, final MessageQueue mq, final SendCallback sendCallback) throws MQClientException, RemotingException,
  60.         MQBrokerException, InterruptedException;
  62.     void send(final Collection<Message> msgs, final MessageQueue mq, final SendCallback sendCallback, final long timeout) throws MQClientException,
  63.         RemotingException, MQBrokerException, InterruptedException;
  65.     //for rpc
  66.     Message request(final Message msg, final long timeout) throws RequestTimeoutException, MQClientException,
  67.         RemotingException, MQBrokerException, InterruptedException;
  69.     void request(final Message msg, final RequestCallback requestCallback, final long timeout)
  70.         throws MQClientException, RemotingException, InterruptedException, MQBrokerException;
  72.     Message request(final Message msg, final MessageQueueSelector selector, final Object arg,
  73.         final long timeout) throws RequestTimeoutException, MQClientException, RemotingException, MQBrokerException,
  74.         InterruptedException;
  76.     void request(final Message msg, final MessageQueueSelector selector, final Object arg,
  77.         final RequestCallback requestCallback,
  78.         final long timeout) throws MQClientException, RemotingException,
  79.         InterruptedException, MQBrokerException;
  81.     Message request(final Message msg, final MessageQueue mq, final long timeout)
  82.         throws RequestTimeoutException, MQClientException, RemotingException, MQBrokerException, InterruptedException;
  84.     void request(final Message msg, final MessageQueue mq, final RequestCallback requestCallback, long timeout)
  85.         throws MQClientException, RemotingException, MQBrokerException, InterruptedException;
  86. }

3.1. 时序图

消息发送 - 图4

3.2. 消息发送入口

org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl()方法作为消息发送的入口，生产者不管调用哪个消息发送方法，最终都会执行到sendDefaultImpl()方法，所以我们叫sendDefaultImpl()为消息发送入口方法

  1. private SendResult sendDefaultImpl(
  2.     Message msg,
  3.     final CommunicationMode communicationMode,
  4.     final SendCallback sendCallback,
  5.     final long timeout
  6.     ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  7.         // 生产者状态验证，状态一定要是RUNING状态
  8.         this.makeSureStateOK();
  9.         // Step1: 消息长度验证，要求主题名称、消息提不能为空。
  10.         // 消息长度要大于0且不能超过允许发送的最大长度1024 * 1024 * 4 (4MB)
  11.         Validators.checkMessage(msg, this.defaultMQProducer);
  12.         final long invokeID = random.nextLong();
  13.         long beginTimestampFirst = System.currentTimeMillis();
  14.         long beginTimestampPrev = beginTimestampFirst;
  15.         long endTimestamp = beginTimestampFirst;
  16.         // Step2: 查找主题的路由信息
  17.         TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
  19.         // Step3: 主题信息没有问题，继续下一步选择队列
  20.         if (topicPublishInfo != null && topicPublishInfo.ok()) {
  21.             boolean callTimeout = false;
  22.             MessageQueue mq = null;
  23.             Exception exception = null;
  24.             SendResult sendResult = null;
  25.             // 最大重试次数
  26.             int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
  27.             // 重试计数器
  28.             int times = 0;
  29.             String[] brokersSent = new String[timesTotal];
  31.             // 循环，实现消息重试，最大重试次数不能超过timesTotal
  32.             for (; times < timesTotal; times++) {
  33.                 String lastBrokerName = null == mq ? null : mq.getBrokerName();
  34.                 // 选择消息队列，轮询策略，通过生成一个随机数，然后递增和消息队列长度取模，根据取模位置返回消息队列
  35.                 // 消息队列选择有两种方式，通过sendLatencyFaultEnable 属性设置：
  36.                 // 1. 启用Broker 故障延迟机制
  37.                 // 2. 不启用Broker 故障延迟机制
  38.                 MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
  39.                 if (mqSelected != null) {
  40.                     mq = mqSelected;
  41.                     // 记录每一次发送消息的broker名称
  42.                     brokersSent[times] = mq.getBrokerName();
  43.                     try {
  44.                         beginTimestampPrev = System.currentTimeMillis();
  45.                         if (times > 0) {
  46.                             //Reset topic with namespace during resend.
  47.                             msg.setTopic(this.defaultMQProducer.withNamespace(msg.getTopic()));
  48.                         }
  49.                         long costTime = beginTimestampPrev - beginTimestampFirst;
  50.                         if (timeout < costTime) {
  51.                             callTimeout = true;
  52.                             break;
  53.                         }
  54.                         // Step4: 调用消息发送核心方法
  55.                         sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
  56.                         endTimestamp = System.currentTimeMillis();
  57.                         this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
  58.                         switch (communicationMode) {
  59.                             case ASYNC:
  60.                                 return null;
  61.                             case ONEWAY:
  62.                                 return null;
  63.                             case SYNC:
  64.                                 if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
  65.                                     if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
  66.                                         continue;
  67.                                     }
  68.                                 }
  70.                                 return sendResult;
  71.                             default:
  72.                                 break;
  73.                         }
  74.                     } catch (RemotingException e) {
  75.                         endTimestamp = System.currentTimeMillis();
  76.                         this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
  77.                         log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
  78.                         log.warn(msg.toString());
  79.                         exception = e;
  80.                         continue;
  81.                     } catch (MQClientException e) {
  82.                         endTimestamp = System.currentTimeMillis();
  83.                         this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
  84.                         log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
  85.                         log.warn(msg.toString());
  86.                         exception = e;
  87.                         continue;
  88.                     } catch (MQBrokerException e) {
  89.                         endTimestamp = System.currentTimeMillis();
  90.                         this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
  91.                         log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
  92.                         log.warn(msg.toString());
  93.                         exception = e;
  94.                         if (this.defaultMQProducer.getRetryResponseCodes().contains(e.getResponseCode())) {
  95.                             continue;
  96.                         } else {
  97.                             if (sendResult != null) {
  98.                                 return sendResult;
  99.                             }
  101.                             throw e;
  102.                         }
  103.                     } catch (InterruptedException e) {
  104.                         endTimestamp = System.currentTimeMillis();
  105.                         this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
  106.                         log.warn(String.format("sendKernelImpl exception, throw exception, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
  107.                         log.warn(msg.toString());
  109.                         log.warn("sendKernelImpl exception", e);
  110.                         log.warn(msg.toString());
  111.                         throw e;
  112.                     }
  113.                 } else {
  114.                     break;
  115.                 }
  116.             }
  118.             if (sendResult != null) {
  119.                 return sendResult;
  120.             }
  122.             String info = String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s",
  123.                 times,
  124.                 System.currentTimeMillis() - beginTimestampFirst,
  125.                 msg.getTopic(),
  126.                 Arrays.toString(brokersSent));
  128.             info += FAQUrl.suggestTodo(FAQUrl.SEND_MSG_FAILED);
  130.             MQClientException mqClientException = new MQClientException(info, exception);
  131.             if (callTimeout) {
  132.                 throw new RemotingTooMuchRequestException("sendDefaultImpl call timeout");
  133.             }
  135.             if (exception instanceof MQBrokerException) {
  136.                 mqClientException.setResponseCode(((MQBrokerException) exception).getResponseCode());
  137.             } else if (exception instanceof RemotingConnectException) {
  138.                 mqClientException.setResponseCode(ClientErrorCode.CONNECT_BROKER_EXCEPTION);
  139.             } else if (exception instanceof RemotingTimeoutException) {
  140.                 mqClientException.setResponseCode(ClientErrorCode.ACCESS_BROKER_TIMEOUT);
  141.             } else if (exception instanceof MQClientException) {
  142.                 mqClientException.setResponseCode(ClientErrorCode.BROKER_NOT_EXIST_EXCEPTION);
  143.             }
  145.             throw mqClientException;
  146.         }
  148.         validateNameServerSetting();
  150.         throw new MQClientException("No route info of this topic: " + msg.getTopic() + FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO),
  151.             null).setResponseCode(ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION);
  152.     }

通过分析消息发送入口方法，消息发送大致分为4个步骤

  1. 消息长度验证
  2. 查找主题路由信息
  3. 选择消息队列
  4. 调用消息发送核心方法

我们依次分析这4步具体做了什么事情

3.2. 消息长度验证

消息的合法性进行检查的逻辑很简单，主要要求主题名称、消息提不能为空，消息长度要大于0且不能超过允许发送的最大长度1024 1024 4 (4MB)

  1. public static void checkMessage(Message msg, DefaultMQProducer defaultMQProducer) throws MQClientException {
  2.     if (null == msg) {
  3.         throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message is null");
  4.     }
  5.     // 消息主题的长度和非空验证
  6.     Validators.checkTopic(msg.getTopic());
  7.     Validators.isNotAllowedSendTopic(msg.getTopic());
  9.     // 消息非空验证
  10.     if (null == msg.getBody()) {
  11.         throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body is null");
  12.     }
  13.     // 消息最小长度验证，长度一定要大于0
  14.     if (0 == msg.getBody().length) {
  15.         throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body length is zero");
  16.     }
  17.     // 消息最大长度验证
  18.     if (msg.getBody().length > defaultMQProducer.getMaxMessageSize()) {
  19.         throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL,
  20.                                     "the message body size over max value, MAX: " + defaultMQProducer.getMaxMessageSize());
  21.     }
  22. }

3.2. 查找主题路由信息

3.2.1. 时序图

消息发送 - 图5

3.2.2. 源码分析

在消息发送之前，需要获取主题的路由信息，只有获得路由信息我们才知道消息具体要发送到哪个Broker节点上，如果生产者中缓存了Topic的路由信息，且该路由信息包含消息队列，则直接返回该路由信息。如果缓存或者没有包含消息队列，就从Nameserver查询该topic的路由信息，如果最终未找到路由信息，就抛出无法找到主题的异常

  1. private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
  2.     // 从生产者本地缓存获取topic的路由信息
  3.     TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
  4.     // 如果本地缓存，向Nameserver查询该topic的路由信息
  5.     if (null == topicPublishInfo ，|| !topicPublishInfo.ok()) {
  6.         this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
  7.         // 从Nameserver 查询topic的路由信息
  8.         this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
  9.         topicPublishInfo = this.topicPublishInfoTable.get(topic);
  10.     }
  11.     // 是否查询到Topic的路由信息校验， 查询到直接返回
  12.     if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
  13.         return topicPublishInfo;
  14.     } else {
  15.         // 没有查询到Topic的路由信息，查询默认主题的路由信息
  16.         // 注意：只有autoCreateTopicEnable=true时，Topic没有路由信息时才会使用默认路由，否则抛出异常
  17.         this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
  18.         topicPublishInfo = this.topicPublishInfoTable.get(topic);
  19.         return topicPublishInfo;
  20.     }
  21. }

主题路由信息描述在TopicPublishInfo类中，这个类的属性是用来描述主题的路由信息

  1. public class TopicPublishInfo {
  2.     /**
  3.      * 是否顺序消息
  4.      */
  5.     private boolean orderTopic = false;
  6.     /**
  7.      * 是否有主题路由信息标识
  8.      */
  9.     private boolean haveTopicRouterInfo = false;
  10.     /**
  11.      * 当前主题的消息队列
  12.      */
  13.     private List<MessageQueue> messageQueueList = new ArrayList<MessageQueue>();
  14.     /**
  15.      * 每选择依次消息队列，该值会自增1，如果超过Integer.MAX_VALUE，重置为0，用于选择消息队列
  16.      */
  17.     private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();
  18.     /**
  19.      * Topic 路由信息
  20.      */
  21.     private TopicRouteData topicRouteData;
  23.     public boolean isOrderTopic() {
  24.         return orderTopic;
  25.     }
  27.     public void setOrderTopic(boolean orderTopic) {
  28.         this.orderTopic = orderTopic;
  29.     }
  31.     public boolean ok() {
  32.         return null != this.messageQueueList && !this.messageQueueList.isEmpty();
  33.     }
  35.     public List<MessageQueue> getMessageQueueList() {
  36.         return messageQueueList;
  37.     }
  39.     public void setMessageQueueList(List<MessageQueue> messageQueueList) {
  40.         this.messageQueueList = messageQueueList;
  41.     }
  43.     public ThreadLocalIndex getSendWhichQueue() {
  44.         return sendWhichQueue;
  45.     }
  47.     public void setSendWhichQueue(ThreadLocalIndex sendWhichQueue) {
  48.         this.sendWhichQueue = sendWhichQueue;
  49.     }
  51.     public boolean isHaveTopicRouterInfo() {
  52.         return haveTopicRouterInfo;
  53.     }
  55.     public void setHaveTopicRouterInfo(boolean haveTopicRouterInfo) {
  56.         this.haveTopicRouterInfo = haveTopicRouterInfo;
  57.     }
  59.     public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
  60.         if (lastBrokerName == null) {
  61.             return selectOneMessageQueue();
  62.         } else {
  63.             for (int i = 0; i < this.messageQueueList.size(); i++) {
  64.                 int index = this.sendWhichQueue.incrementAndGet();
  65.                 int pos = Math.abs(index) % this.messageQueueList.size();
  66.                 if (pos < 0)
  67.                     pos = 0;
  68.                 MessageQueue mq = this.messageQueueList.get(pos);
  69.                 if (!mq.getBrokerName().equals(lastBrokerName)) {
  70.                     return mq;
  71.                 }
  72.             }
  73.             return selectOneMessageQueue();
  74.         }
  75.     }
  77.     public MessageQueue selectOneMessageQueue() {
  78.         int index = this.sendWhichQueue.incrementAndGet();
  79.         int pos = Math.abs(index) % this.messageQueueList.size();
  80.         if (pos < 0)
  81.             pos = 0;
  82.         return this.messageQueueList.get(pos);
  83.     }
  85.     public int getQueueIdByBroker(final String brokerName) {
  86.         for (int i = 0; i < topicRouteData.getQueueDatas().size(); i++) {
  87.             final QueueData queueData = this.topicRouteData.getQueueDatas().get(i);
  88.             if (queueData.getBrokerName().equals(brokerName)) {
  89.                 return queueData.getWriteQueueNums();
  90.             }
  91.         }
  93.         return -1;
  94.     }
  96.     @Override
  97.     public String toString() {
  98.         return "TopicPublishInfo [orderTopic=" + orderTopic + ", messageQueueList=" + messageQueueList
  99.             + ", sendWhichQueue=" + sendWhichQueue + ", haveTopicRouterInfo=" + haveTopicRouterInfo + "]";
  100.     }
  102.     public TopicRouteData getTopicRouteData() {
  103.         return topicRouteData;
  104.     }
  106.     public void setTopicRouteData(final TopicRouteData topicRouteData) {
  107.         this.topicRouteData = topicRouteData;
  108.     }
  109. }

Topic 元数据信息TopicRouteData 的属性字段描述

  1. public class TopicRouteData extends RemotingSerializable {
  2.     private String orderTopicConf;
  3.     /**
  4.      * topic 的队列信息
  5.      */
  6.     private List<QueueData> queueDatas;
  7.     /**
  8.      * topic 所在的Broker 信息
  9.      */
  10.     private List<BrokerData> brokerDatas;
  11.     /**
  12.      * broker 上过滤服务器的地址列表
  13.      */
  14.     private HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
  16. }

当根据topic 名称从生产者本地获取不到主题路由信息时，从Nameserver 尝试获取路由信息，然后对比Nameserver中获取的路由信息和本地缓存表中的路由信息，如果不一致更新本地缓存表中的路由信息，通过updateTopicRouteInfoFromNameServer()获取路由信息

  1. /**
  2. * 更新消息生产者和维护路由缓存
  3. *
  4. * @param topic             主题名称
  5. * @param isDefault         是否查询默认主题路由信息
  6. * @param defaultMQProducer 生产者实例
  7. * @return
  8. */
  9. public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
  10.                                                   DefaultMQProducer defaultMQProducer) {
  11.     try {
  12.         // 上锁
  13.         if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
  14.             try {
  15.                 TopicRouteData topicRouteData;
  16.                 // 处理默认主题的路由信息
  17.                 if (isDefault && defaultMQProducer != null) {
  18.                     // 查询默认主题路由信息，默认3秒超时
  19.                     topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(),
  20.                                                                                                  clientConfig.getMqClientApiTimeout());
  21.                     // 如果查询到默认路由信息
  22.                     // 将路由信息中读写队列的个数替换为生产者默认队列个数
  23.                     if (topicRouteData != null) {
  24.                         for (QueueData data : topicRouteData.getQueueDatas()) {
  25.                             // 默认队列数量和Nameserver中队列数量，取最小数量
  26.                             int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
  27.                             data.setReadQueueNums(queueNums);
  28.                             data.setWriteQueueNums(queueNums);
  29.                         }
  30.                     }
  31.                 } else {
  32.                     // 查询自定义主题路由信息
  33.                     topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, clientConfig.getMqClientApiTimeout());
  34.                 }
  35.                 // 非空表示找到Topic的路由信息，与本地缓存中的路由信息进行对比
  36.                 // 判断路由信息是否发生了改变，如果未发生变化直接返回false
  37.                 if (topicRouteData != null) {
  38.                     // 本地缓存中取路由信息表
  39.                     TopicRouteData old = this.topicRouteTable.get(topic);
  40.                     // 对比Topic本地缓存和Nameserver 中的路由信息表
  41.                     boolean changed = topicRouteDataIsChange(old, topicRouteData);
  42.                     // 本地缓存和Nameserver中路由信息不一致，更新
  43.                     if (!changed) {
  44.                         changed = this.isNeedUpdateTopicRouteInfo(topic);
  45.                     } else {
  46.                         log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);
  47.                     }
  49.                     if (changed) {
  50.                         TopicRouteData cloneTopicRouteData = topicRouteData.cloneTopicRouteData();
  52.                         for (BrokerData bd : topicRouteData.getBrokerDatas()) {
  53.                             this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
  54.                         }
  56.                         // Update Pub info
  57.                         if (!producerTable.isEmpty()) {
  58.                             TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);
  59.                             // 有主题路由标识
  60.                             publishInfo.setHaveTopicRouterInfo(true);
  61.                             Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
  62.                             while (it.hasNext()) {
  63.                                 Entry<String, MQProducerInner> entry = it.next();
  64.                                 MQProducerInner impl = entry.getValue();
  65.                                 if (impl != null) {
  66.                                     impl.updateTopicPublishInfo(topic, publishInfo);
  67.                                 }
  68.                             }
  69.                         }
  71.                         // Update sub info
  72.                         if (!consumerTable.isEmpty()) {
  73.                             Set<MessageQueue> subscribeInfo = topicRouteData2TopicSubscribeInfo(topic, topicRouteData);
  74.                             Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
  75.                             while (it.hasNext()) {
  76.                                 Entry<String, MQConsumerInner> entry = it.next();
  77.                                 MQConsumerInner impl = entry.getValue();
  78.                                 if (impl != null) {
  79.                                     impl.updateTopicSubscribeInfo(topic, subscribeInfo);
  80.                                 }
  81.                             }
  82.                         }
  83.                         log.info("topicRouteTable.put. Topic = {}, TopicRouteData[{}]", topic, cloneTopicRouteData);
  84.                         this.topicRouteTable.put(topic, cloneTopicRouteData);
  85.                         return true;
  86.                     }
  87.                 } else {
  88.                     log.warn("updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {}. [{}]", topic, this.clientId);
  89.                 }
  90.             } catch (MQClientException e) {
  91.                 if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) && !topic.equals(TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC)) {
  92.                     log.warn("updateTopicRouteInfoFromNameServer Exception", e);
  93.                 }
  94.             } catch (RemotingException e) {
  95.                 log.error("updateTopicRouteInfoFromNameServer Exception", e);
  96.                 throw new IllegalStateException(e);
  97.             } finally {
  98.                 this.lockNamesrv.unlock();
  99.             }
  100.         } else {
  101.             log.warn("updateTopicRouteInfoFromNameServer tryLock timeout {}ms. [{}]", LOCK_TIMEOUT_MILLIS, this.clientId);
  102.         }
  103.     } catch (InterruptedException e) {
  104.         log.warn("updateTopicRouteInfoFromNameServer Exception", e);
  105.     }
  107.     return false;
  108. }

3.3. 选择消息队列

提供两种消息队列选择机制，默认机制为不启用Broker 故障延迟机制，还有一种启用故障延迟机制。
通过sendLatencyFaultEnable 参数设置，是boolean值， false 为默认机制，true 为Broker 故障延迟机制

注意：如果sendLatencyFaultEnable 设置为false也能规避Broker，它们的区别在于如果sendLatencyFaultEnable设置为true，当消息发送消息失败时，就会悲观的认为Broker不可用，那么在接下来的一段时间内不在向其发送消息，直接略过该Broker。sendLatencyFaultEnable设置为true时，就只会在本次消息发送的重试过程中略过Broker，下一次消息发送还是会继续尝试使用该Broker

  1. public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
  2.     // 是否启用Broker 故障延迟机制，默认不启用，可以通过DefaultMQProducerImpl对象设置
  3.     // Broker 故障延迟机制
  4.     if (this.sendLatencyFaultEnable) {
  5.         try {
  6.             // Step1: 轮询获取一个消息队列
  7.             int index = tpInfo.getSendWhichQueue().incrementAndGet();
  8.             for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
  9.                 int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
  10.                 if (pos < 0)
  11.                     pos = 0;
  12.                 MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
  13.                 // 验证消息队列是否可用
  14.                 if (latencyFaultTolerance.isAvailable(mq.getBrokerName()))
  15.                     return mq;
  16.             }
  17.             // 尝试从避规的Broker中选择一个可用的Broker，如果没有找到返回null
  18.             final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
  19.             // 一个主题的队列写数量
  20.             int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
  21.             if (writeQueueNums > 0) {
  22.                 final MessageQueue mq = tpInfo.selectOneMessageQueue();
  23.                 if (notBestBroker != null) {
  24.                     mq.setBrokerName(notBestBroker);
  25.                     mq.setQueueId(tpInfo.getSendWhichQueue().incrementAndGet() % writeQueueNums);
  26.                 }
  27.                 return mq;
  28.             } else {
  29.                 latencyFaultTolerance.remove(notBestBroker);
  30.             }
  31.         } catch (Exception e) {
  32.             log.error("Error occurred when selecting message queue", e);
  33.         }
  35.         return tpInfo.selectOneMessageQueue();
  36.     }
  37.     // 没有启动Broker 故障延迟机制选择消息队列
  38.     // 默认机制
  39.     return tpInfo.selectOneMessageQueue(lastBrokerName);
  40. }

3.3.1. 默认机制

消息队列选择策略在当org.apache.rocketmq.client.latency.MQFaultStrategy.selectOneMessageQueue()中，如果sendLatencyFaultEnable=false 使用默认消息选择策略，调用TopicPublishInfo.selectOneMessageQueue()方法选择消息队列，默认消息队列选择采用轮询策略
这种机制的特点在于只针对当前发送的消息，在重试时规避掉不可用的Broker

  1. /**
  2. * 默认消息队列选择机制（默认机制）
  3. * 没有开启Broker 故障延迟机制
  4. * @param lastBrokerName  上一次选择执行发送失败的Broker
  5. * @return
  6. */
  7. public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
  8.     // 消息发送第一次选择Queue
  9.     if (lastBrokerName == null) {、
  10.         // 第一次选择QUeue时，直接用自增（sendWhichQueue）变量和消息路由表中的队列数量取模
  11.         // 根据取模下标获取队列
  12.         return selectOneMessageQueue();
  13.     } else {
  14.         // 消息发送不是第一次选择Queue，此时说明至少有一次消息发送失败了，发生重试，
  15.         // 此时lastBrokerName 为上一次发送消息选择的Broker名称
  16.         for (int i = 0; i < this.messageQueueList.size(); i++) {
  17.             int index = this.sendWhichQueue.incrementAndGet();
  18.             int pos = Math.abs(index) % this.messageQueueList.size();
  19.             if (pos < 0)
  20.                 pos = 0;
  21.             MessageQueue mq = this.messageQueueList.get(pos);
  22.             // 校验当前发送消息选择的Broker名称不能上一次发送消息选择的Broker名称是否相同
  23.             // 对于相同的消息队列，不做为本次消息队列选择的Broker
  24.             if (!mq.getBrokerName().equals(lastBrokerName)) {
  25.                 return mq;
  26.             }
  27.         }
  28.         return selectOneMessageQueue();
  29.     }
  30. }

敲重点：为什么选择消息队列的时候会有一个lastBrokerName参数，首先说一下这个参数是上一次选择执行发送消息失败的Broker名称，第一次执行消息队列选择时，lastBrokerName 为Null，当消息发送失败重试时，lastBrokerName为上一次选择的队列所属的Broker名称。这么做的目的是避免重试消息发送时还是选择相同的Broker进行发送，如果重试选择的Broker和上一次发送选择的Broker是一样，发生失败的概率会很高。
RocketMQ-第 4 页.jpg
lastBrokerName 变化图

3.3.2. Broker 故障延迟机制

sendLatencyFaultEnable 设置为true时，使用故障延迟机制来规避Broker的选择，故障延迟机制的特点在于当前消息发送失败时，会将发送的Broker在一段时间规避，也就时说在一段时间内下一个消息发送时，该Broker不能被选择

  1. /**
  2. * 选择消息队列
  3. * 提供两种消息队列选择机制，默认机制、Broker 故障延迟机制
  4. *
  5. * @param tpInfo
  6. * @param lastBrokerName
  7. * @return
  8. */
  9. public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
  10.     // 是否启用Broker 故障延迟机制，默认不启用，可以通过DefaultMQProducerImpl对象设置
  11.     if (this.sendLatencyFaultEnable) {
  12.         try {
  13.             // Step1: 轮询获取一个消息队列
  14.             int index = tpInfo.getSendWhichQueue().incrementAndGet();
  15.             for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
  16.                 int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
  17.                 if (pos < 0)
  18.                     pos = 0;
  19.                 MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
  20.                 // 验证消息队列是否可用
  21.                 if (latencyFaultTolerance.isAvailable(mq.getBrokerName()))
  22.                     return mq;
  23.             }
  24.             // 尝试从避规的Broker中选择一个可用的Broker，如果没有找到返回null
  25.             final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
  26.             // 一个主题的队列写数量
  27.             int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
  28.             if (writeQueueNums > 0) {
  29.                 final MessageQueue mq = tpInfo.selectOneMessageQueue();
  30.                 if (notBestBroker != null) {
  31.                     mq.setBrokerName(notBestBroker);
  32.                     mq.setQueueId(tpInfo.getSendWhichQueue().incrementAndGet() % writeQueueNums);
  33.                 }
  34.                 return mq;
  35.             } else {
  36.                 latencyFaultTolerance.remove(notBestBroker);
  37.             }
  38.         } catch (Exception e) {
  39.             log.error("Error occurred when selecting message queue", e);
  40.         }
  42.         return tpInfo.selectOneMessageQueue();
  43.     }
  44.     // 没有启动Broker故障延迟机制选择消息队列
  45.     return tpInfo.selectOneMessageQueue(lastBrokerName);
  46. }

当消息发送失败时，会调用org.apache.rocketmq.client.latency.updateFaultItem()方法来更新当前Broker 不可用的时间（规避时间）

  1. /**
  2. * 消息发送异常时，更新失败Broker条目的延迟信息
  3. *
  4. * @param brokerName     broker名称
  5. * @param currentLatency 本次消息发送的延迟时间
  6. * @param isolation      是否规避Broker {true:规避,false:不规避}
  7. */
  8. public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
  9.     // 校验是否启用Broker故障延迟机制
  10.     if (this.sendLatencyFaultEnable) {
  11.         // 如果规避Broker，使用默认30s作为Broker故障规避时间，否则使用消息发送延迟时间来计算Broker故障规避时间
  12.         long duration = computeNotAvailableDuration(isolation ? 30000 : currentLatency);
  13.         // 更新Broker 规避条目
  14.         this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
  15.     }
  16. }

计算Broker规避时长，在这个时间段内Broker将不在参与消息发送队列的负载。

  1. private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
  2. // 规避时长
  3. private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};
  1. /**
  2. * 计算本次消息发送故障需要规避Broker的时长，也就是接下来多长时间内Broker将不参与消息发送队列负债
  3. *
  4. * @param currentLatency
  5. * @return
  6. */
  7. private long computeNotAvailableDuration(final long currentLatency) {
  8.     // 从latencyMax尾部开始寻找，找到第一个比currentLatency小的下标
  9.     // 然后从notAvailableDuration数组中获取需要规避的时长
  10.     for (int i = latencyMax.length - 1; i >= 0; i--) {
  11.         if (currentLatency >= latencyMax[i])
  12.             return this.notAvailableDuration[i];
  13.     }
  15.     return 0;
  16. }

3.4. 消息发送核心

3.4.1. 时序图

消息发送 - 图7

3.4.2. 源码分析

在选择消息发送队列后，接下来执行消息发送核心方法，在org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendKernelImpl()

  1. /**
  2. * 消息发送
  3. * @param msg                 待发送消息
  4. * @param mq                  消息发送的消息队列
  5. * @param communicationMode   消息发送模式，SYNC\ASYNC\ONEWAY
  6. * @param sendCallback        异步消息回调函数
  7. * @param topicPublishInfo    主题路由信息
  8. * @param timeout             消息发送超时时间
  9. * @return
  10. * @throws MQClientException
  11. * @throws RemotingException
  12. * @throws MQBrokerException
  13. * @throws InterruptedException
  14. */
  15. private SendResult sendKernelImpl(final Message msg,
  16.                                   final MessageQueue mq,
  17.                                   final CommunicationMode communicationMode,
  18.                                   final SendCallback sendCallback,
  19.                                   final TopicPublishInfo topicPublishInfo,
  20.                                   final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  21.     long beginStartTime = System.currentTimeMillis();
  22.     // Step1:根据Broker名称获取Broker的网络地址
  23.     // 如果本地MQClientInstance实例中brokerAddrTable 为缓存Broker信息，从Nameserver主动更新topic的路由信息
  24.     // 如果路由信息更新后还是拿不到Broker信息就抛出异常（MQClientException）提示Broker不存在
  25.     String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
  26.     if (null == brokerAddr) {
  27.         tryToFindTopicPublishInfo(mq.getTopic());
  28.         brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
  29.     }
  30.     // 定义消息发送对象
  31.     SendMessageContext context = null;
  32.     if (brokerAddr != null) {
  33.         brokerAddr = MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
  34.         // 消息内容原始数据
  35.         byte[] prevBody = msg.getBody();
  36.         try {
  37.             //for MessageBatch,ID has been set in the generating process
  38.             // Step2: 为消息分配全局唯一ID
  39.             if (!(msg instanceof MessageBatch)) {
  40.                 MessageClientIDSetter.setUniqID(msg);
  41.             }
  43.             // 如果有命名空间，使用命名空间做为消息实例ID
  44.             boolean topicWithNamespace = false;
  45.             if (null != this.mQClientFactory.getClientConfig().getNamespace()) {
  46.                 msg.setInstanceId(this.mQClientFactory.getClientConfig().getNamespace());
  47.                 topicWithNamespace = true;
  48.             }
  50.             int sysFlag = 0;
  51.             boolean msgBodyCompressed = false;
  52.             // 对消息体采用zip压缩，默认大于4k的消息体会进行压缩
  53.             if (this.tryToCompressMessage(msg)) {
  54.                 sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
  55.                 msgBodyCompressed = true;
  56.             }
  57.             // 是否为消息系统标记
  58.             final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
  59.             if (Boolean.parseBoolean(tranMsg)) {
  60.                 sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
  61.             }
  64.             if (hasCheckForbiddenHook()) {
  65.                 CheckForbiddenContext checkForbiddenContext = new CheckForbiddenContext();
  66.                 checkForbiddenContext.setNameSrvAddr(this.defaultMQProducer.getNamesrvAddr());
  67.                 checkForbiddenContext.setGroup(this.defaultMQProducer.getProducerGroup());
  68.                 checkForbiddenContext.setCommunicationMode(communicationMode);
  69.                 checkForbiddenContext.setBrokerAddr(brokerAddr);
  70.                 checkForbiddenContext.setMessage(msg);
  71.                 checkForbiddenContext.setMq(mq);
  72.                 checkForbiddenContext.setUnitMode(this.isUnitMode());
  73.                 this.executeCheckForbiddenHook(checkForbiddenContext);
  74.             }
  75.             // Step3: 如果注册了消息发送钩子函数，则执行消息发送之前的钩子函数（前置执行）
  76.             // 钩子函数通过DefaultMQProducerImpl.registerSendMessageHook()方法注册，钩子函数需要实现SendMessageHook接口
  77.             // 可以注册多个，MQ默认为我们注册了SendMessageTraceHookImpl 钩子函数
  78.             if (this.hasSendMessageHook()) {
  79.                 context = new SendMessageContext();
  80.                 context.setProducer(this);
  81.                 context.setProducerGroup(this.defaultMQProducer.getProducerGroup());
  82.                 context.setCommunicationMode(communicationMode);
  83.                 context.setBornHost(this.defaultMQProducer.getClientIP());
  84.                 context.setBrokerAddr(brokerAddr);
  85.                 context.setMessage(msg);
  86.                 context.setMq(mq);
  87.                 context.setNamespace(this.defaultMQProducer.getNamespace());
  88.                 String isTrans = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
  89.                 if (isTrans != null && isTrans.equals("true")) {
  90.                     context.setMsgType(MessageType.Trans_Msg_Half);
  91.                 }
  93.                 if (msg.getProperty("__STARTDELIVERTIME") != null || msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != null) {
  94.                     context.setMsgType(MessageType.Delay_Msg);
  95.                 }
  96.                 // 执行钩子函数，既执行实现了SendMessageHook 接口的类，并且注册到DefaultMQProducerImpl.sendMessageHookList
  97.                 this.executeSendMessageHookBefore(context);
  98.             }
  100.             // Step4: 构建消息发送体，消息体信息主要包括：
  101.             // 生产者组、主题名称、默认创建主题key、该主题在单个Broker上的默认队列数、队列ID（队列序号）、消息系统标记（MessageSysFlag）、
  102.             // 消息发送时间、消息标记（RocketMQ对消息中的标记不做任何处理，仅提供应用程序使用）、消息扩展属性、消息重试次数、是否是批量消息
  103.             SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
  104.             // 生产者组
  105.             requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
  106.             // 主题名称
  107.             requestHeader.setTopic(msg.getTopic());
  108.             // 默认创建主题key
  109.             requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
  110.             // 该主题在单个Broker上的默认队列数
  111.             requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
  112.             // 队列ID（队列序号）
  113.             requestHeader.setQueueId(mq.getQueueId());
  114.             // 消息系统标记（MessageSysFlag）
  115.             requestHeader.setSysFlag(sysFlag);
  116.             // 消息发送时间
  117.             requestHeader.setBornTimestamp(System.currentTimeMillis());
  118.             // 消息标记（RocketMQ对消息中的标记不做任何处理，仅提供应用程序使用）
  119.             requestHeader.setFlag(msg.getFlag());
  120.             // 消息扩展属性
  121.             requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
  122.             // 消息重试次数
  123.             requestHeader.setReconsumeTimes(0);
  124.             requestHeader.setUnitMode(this.isUnitMode());
  125.             // 是否是批量消息
  126.             requestHeader.setBatch(msg instanceof MessageBatch);
  127.             // 处理重试主题
  128.             if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
  129.                 // 获取重试次数
  130.                 String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
  131.                 if (reconsumeTimes != null) {
  132.                     // 消息重试次数
  133.                     requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
  134.                     // 清除Message待发送消息重试次数
  135.                     MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
  136.                 }
  137.                 // 最大重试次数
  138.                 String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(msg);
  139.                 if (maxReconsumeTimes != null) {
  140.                     // 设置最大重试次数
  141.                     requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
  142.                     // 清除Message待发送消息最大重试次数
  143.                     MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_MAX_RECONSUME_TIMES);
  144.                 }
  145.             }
  147.             SendResult sendResult = null;
  148.             switch (communicationMode) {
  149.                     // 异步消息，异步消息发送支持重试
  150.                     // 注意异步发送仅在服务端收到消息发送响应请求时才会重试，如果出现网络异常、超时等情况不会重试
  151.                 case ASYNC:
  152.                     Message tmpMessage = msg;
  153.                     boolean messageCloned = false;
  154.                     // 如果消息内容被压缩，msg对象中的消息内容重置回原始（压缩之前）内容
  155.                     if (msgBodyCompressed) {
  156.                         //If msg body was compressed, msgbody should be reset using prevBody.
  157.                         //Clone new message using commpressed message body and recover origin massage.
  158.                         //Fix bug:https://github.com/apache/rocketmq-externals/issues/66
  159.                         tmpMessage = MessageAccessor.cloneMessage(msg);
  160.                         messageCloned = true;
  161.                         msg.setBody(prevBody);
  162.                     }
  164.                     if (topicWithNamespace) {
  165.                         if (!messageCloned) {
  166.                             tmpMessage = MessageAccessor.cloneMessage(msg);
  167.                             messageCloned = true;
  168.                         }
  169.                         msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQProducer.getNamespace()));
  170.                     }
  172.                     // 消息发送超时校验
  173.                     long costTimeAsync = System.currentTimeMillis() - beginStartTime;
  174.                     if (timeout < costTimeAsync) {
  175.                         throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
  176.                     }
  177.                     // 异步发送消息
  178.                     sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
  179.                         brokerAddr,
  180.                         mq.getBrokerName(),
  181.                         tmpMessage,
  182.                         requestHeader,
  183.                         timeout - costTimeAsync,
  184.                         communicationMode,
  185.                         sendCallback,
  186.                         topicPublishInfo,
  187.                         this.mQClientFactory,
  188.                         this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(),
  189.                         context,
  190.                         this);
  191.                     break;
  192.                 case ONEWAY:
  193.                     // 单项发送
  195.                     // 同步消息，同步发送消息不支持重试
  196.                 case SYNC:
  197.                     // 消息发送超时校验
  198.                     long costTimeSync = System.currentTimeMillis() - beginStartTime;
  199.                     if (timeout < costTimeSync) {
  200.                         throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
  201.                     }
  202.                     // 同步发送消息
  203.                     sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
  204.                         brokerAddr,
  205.                         mq.getBrokerName(),
  206.                         msg,
  207.                         requestHeader,
  208.                         timeout - costTimeSync,
  209.                         communicationMode,
  210.                         context,
  211.                         this);
  212.                     break;
  213.                 default:
  214.                     assert false;
  215.                     break;
  216.             }
  218.             // Step6: 执行钩子成函数（后置执行）
  219.             if (this.hasSendMessageHook()) {
  220.                 context.setSendResult(sendResult);
  221.                 this.executeSendMessageHookAfter(context);
  222.             }
  224.             return sendResult;
  225.         } catch (RemotingException e) {
  226.             if (this.hasSendMessageHook()) {
  227.                 context.setException(e);
  228.                 // 执行钩子成函数（后置执行）
  229.                 this.executeSendMessageHookAfter(context);
  230.             }
  231.             throw e;
  232.         } catch (MQBrokerException e) {
  233.             if (this.hasSendMessageHook()) {
  234.                 context.setException(e);
  235.                 // 执行钩子成函数（后置执行）
  236.                 this.executeSendMessageHookAfter(context);
  237.             }
  238.             throw e;
  239.         } catch (InterruptedException e) {
  240.             if (this.hasSendMessageHook()) {
  241.                 context.setException(e);
  242.                 // 执行钩子成函数（后置执行）
  243.                 this.executeSendMessageHookAfter(context);
  244.             }
  245.             throw e;
  246.         } finally {
  247.             msg.setBody(prevBody);
  248.             msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQProducer.getNamespace()));
  249.         }
  250.     }
  252.     throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
  253. }

3.4.2.1. 获取Broker的master节点网络地址

首先从本地缓存表中获取Broker的master节点的网络地址

  1. /**
  2. * 获取Broker的master节点网络地址
  3. * @param brokerName
  4. * @return
  5. */
  6. public String findBrokerAddressInPublish(final String brokerName) {
  7.     HashMap<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName);
  8.     if (map != null && !map.isEmpty()) {
  9.         return map.get(MixAll.MASTER_ID);
  10.     }
  12.     return null;
  13. }

如果本地缓存表中没有，从Nameserver获取Broker的路由信息，具体逻辑看3.2章节查找主题路由信息

  1. private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
  2.     // 从生产者本地缓存获取topic的路由信息
  3.     TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
  4.     // 如果本地缓存，向Nameserver查询该topic的路由信息
  5.     if (null == topicPublishInfo || !topicPublishInfo.ok()) {
  6.         this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
  7.         // 从Nameserver 查询topic的路由信息
  8.         // Nameserver和本地缓存的路由信息对比，如果不一致，替换本地缓存表中的路由信息
  9.         this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
  10.         topicPublishInfo = this.topicPublishInfoTable.get(topic);
  11.     }
  12.     // 是否查询到Topic的路由信息校验， 查询到直接返回
  13.     if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
  14.         return topicPublishInfo;
  15.     } else {
  16.         // 没有查询到Topic的路由信息，从Nameserver查询默认主题的路由信息
  17.         // 注意：只有autoCreateTopicEnable=true时，Topic没有路由信息时才会使用默认路由，否则抛出异常
  18.         // Nameserver默认路由信息和本地缓存的默认路由信息对比，如果不一致，替换本地缓存表中的路由信息
  19.         this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
  20.         topicPublishInfo = this.topicPublishInfoTable.get(topic);
  21.         return topicPublishInfo;
  22.     }
  23. }

3.4.2.2. 为消息分配唯一ID

  1. public static void setUniqID(final Message msg) {
  2.     if (msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null) {
  3.         msg.putProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, createUniqID());
  4.     }
  5. }

消息id生成规则

  1. public static String createUniqID() {
  2.     char[] sb = new char[LEN * 2];
  3.     System.arraycopy(FIX_STRING, 0, sb, 0, FIX_STRING.length);
  4.     long current = System.currentTimeMillis();
  5.     if (current >= nextStartTime) {
  6.         setStartTime(current);
  7.     }
  8.     int diff = (int)(current - startTime);
  9.     if (diff < 0 && diff > -1000_000) {
  10.         // may cause by NTP
  11.         diff = 0;
  12.     }
  13.     int pos = FIX_STRING.length;
  14.     UtilAll.writeInt(sb, pos, diff);
  15.     pos += 8;
  16.     UtilAll.writeShort(sb, pos, COUNTER.getAndIncrement());
  17.     return new String(sb);
  18. }

3.4.2.3. 消息发送之前的钩子函数

在消息发送之前，如果有注册钩子函数，先执行消息发送之前的钩子函数。
钩子函数通过DefaultMQProducerImpl.registerSendMessageHook()方法注册，钩子函数需要实现SendMessageHook接口， 可以注册多个，MQ默认为我们注册了SendMessageTraceHookImpl 钩子函数

  1. public void executeSendMessageHookBefore(final SendMessageContext context) {
  2.     if (!this.sendMessageHookList.isEmpty()) {
  3.         for (SendMessageHook hook : this.sendMessageHookList) {
  4.             try {
  5.                 hook.sendMessageBefore(context);
  6.             } catch (Throwable e) {
  7.                 log.warn("failed to executeSendMessageHookBefore", e);
  8.             }
  9.         }
  10.     }
  11. }

3.4.2.4. 消息发送

首先构建消息发送体，消息体内容包括：生产者组、主题名称、默认创建主题key、该主题在单个Broker上的默认队列数、队列ID（队列序号）、消息系统标记（MessageSysFlag）、
消息发送时间、消息标记（RocketMQ对消息中的标记不做任何处理，仅提供应用程序使用）、消息扩展属性、消息重试次数、是否是批量消息
然后根据消息发送方式向Broker 发送消息，消息发送方式分为同步、异步、单项。
同步和单项消息发送有返回结果。异步消息发送没有返回结果，通过回调函处理消息发送结果

  1. 消息同步发送

消息同步发送方法org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessageSync()

  1. private SendResult sendMessageSync(
  2.     final String addr,
  3.     final String brokerName,
  4.     final Message msg,
  5.     final long timeoutMillis,
  6.     final RemotingCommand request
  7.     ) throws RemotingException, MQBrokerException, InterruptedException {
  8.         // Netty 同步发送消息
  9.         RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
  10.         assert response != null;
  11.         // 处理消息发送结果
  12.         return this.processSendResponse(brokerName, msg, response, addr);
  13.     }

同步发送消息

  1. public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)
  2.     throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException {
  3.     long beginStartTime = System.currentTimeMillis();
  4.     final Channel channel = this.getAndCreateChannel(addr);
  5.     if (channel != null && channel.isActive()) {
  6.         try {
  7.             // 消息发送前置处理
  8.             doBeforeRpcHooks(addr, request);
  9.             long costTime = System.currentTimeMillis() - beginStartTime;
  10.             // 超时校验
  11.             if (timeoutMillis < costTime) {
  12.                 throw new RemotingTimeoutException("invokeSync call the addr[" + addr + "] timeout");
  13.             }
  14.             // 通过Netty调用Broker 发送消息
  15.             RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
  16.             // 消息发送后置处理
  17.             doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
  18.             return response;
  19.         } catch (RemotingSendRequestException e) {
  20.             log.warn("invokeSync: send request exception, so close the channel[{}]", addr);
  21.             // 关闭通道
  22.             this.closeChannel(addr, channel);
  23.             throw e;
  24.         } catch (RemotingTimeoutException e) {
  25.             if (nettyClientConfig.isClientCloseSocketIfTimeout()) {
  26.                 this.closeChannel(addr, channel);
  27.                 log.warn("invokeSync: close socket because of timeout, {}ms, {}", timeoutMillis, addr);
  28.             }
  29.             log.warn("invokeSync: wait response timeout exception, the channel[{}]", addr);
  30.             throw e;
  31.         }
  32.     } else {
  33.         // 关闭通道
  34.         this.closeChannel(addr, channel);
  35.         throw new RemotingConnectException(addr);
  36.     }
  37. }

处理同步发送消息结果

  1. private SendResult processSendResponse(
  2.     final String brokerName,
  3.     final Message msg,
  4.     final RemotingCommand response,
  5.     final String addr
  6.     ) throws MQBrokerException, RemotingCommandException {
  7.         SendStatus sendStatus;
  8.         switch (response.getCode()) {
  9.             case ResponseCode.FLUSH_DISK_TIMEOUT: {
  10.                 sendStatus = SendStatus.FLUSH_DISK_TIMEOUT;
  11.                 break;
  12.             }
  13.             case ResponseCode.FLUSH_SLAVE_TIMEOUT: {
  14.                 sendStatus = SendStatus.FLUSH_SLAVE_TIMEOUT;
  15.                 break;
  16.             }
  17.             case ResponseCode.SLAVE_NOT_AVAILABLE: {
  18.                 sendStatus = SendStatus.SLAVE_NOT_AVAILABLE;
  19.                 break;
  20.             }
  21.             case ResponseCode.SUCCESS: {
  22.                 sendStatus = SendStatus.SEND_OK;
  23.                 break;
  24.             }
  25.             default: {
  26.                 throw new MQBrokerException(response.getCode(), response.getRemark(), addr);
  27.             }
  28.         }
  29.         // 构建同步发送消息的响应结果对象
  30.         SendMessageResponseHeader responseHeader =
  31.             (SendMessageResponseHeader) response.decodeCommandCustomHeader(SendMessageResponseHeader.class);
  33.         //If namespace not null , reset Topic without namespace.
  34.         String topic = msg.getTopic();
  35.         if (StringUtils.isNotEmpty(this.clientConfig.getNamespace())) {
  36.             topic = NamespaceUtil.withoutNamespace(topic, this.clientConfig.getNamespace());
  37.         }
  39.         MessageQueue messageQueue = new MessageQueue(topic, brokerName, responseHeader.getQueueId());
  41.         String uniqMsgId = MessageClientIDSetter.getUniqID(msg);
  42.         if (msg instanceof MessageBatch) {
  43.             StringBuilder sb = new StringBuilder();
  44.             for (Message message : (MessageBatch) msg) {
  45.                 sb.append(sb.length() == 0 ? "" : ",").append(MessageClientIDSetter.getUniqID(message));
  46.             }
  47.             uniqMsgId = sb.toString();
  48.         }
  49.         // 同步发送消息结果
  50.         SendResult sendResult = new SendResult(sendStatus,
  51.             uniqMsgId,
  52.             responseHeader.getMsgId(), messageQueue, responseHeader.getQueueOffset());
  53.         sendResult.setTransactionId(responseHeader.getTransactionId());
  54.         String regionId = response.getExtFields().get(MessageConst.PROPERTY_MSG_REGION);
  55.         String traceOn = response.getExtFields().get(MessageConst.PROPERTY_TRACE_SWITCH);
  56.         if (regionId == null || regionId.isEmpty()) {
  57.             regionId = MixAll.DEFAULT_TRACE_REGION_ID;
  58.         }
  59.         if (traceOn != null && traceOn.equals("false")) {
  60.             sendResult.setTraceOn(false);
  61.         } else {
  62.             sendResult.setTraceOn(true);
  63.         }
  64.         sendResult.setRegionId(regionId);
  65.         return sendResult;
  66.     }

  1. 异步发送

    1. private void sendMessageAsync(
    2.  final String addr,
    3.  final String brokerName,
    4.  final Message msg,
    5.  final long timeoutMillis,
    6.  final RemotingCommand request,
    7.  final SendCallback sendCallback,
    8.  final TopicPublishInfo topicPublishInfo,
    9.  final MQClientInstance instance,
    10.  final int retryTimesWhenSendFailed,
    11.  final AtomicInteger times,
    12.  final SendMessageContext context,
    13.  final DefaultMQProducerImpl producer
    14.  ) throws InterruptedException, RemotingException {
    15.      final long beginStartTime = System.currentTimeMillis();
    16.      try {
    17.          // Netty 请求发送消息，接收该请求的类是org.apache.rocketmq.broker.processor.SendMessageProcessor
    18.          this.remotingClient.invokeAsync(addr, request, timeoutMillis, new InvokeCallback() {
    19.              @Override
    20.              public void operationComplete(ResponseFuture responseFuture) {
    21.                  long cost = System.currentTimeMillis() - beginStartTime;
    22.                  RemotingCommand response = responseFuture.getResponseCommand();
    23.                  if (null == sendCallback && response != null) {
    25.                      try {
    26.                          SendResult sendResult = MQClientAPIImpl.this.processSendResponse(brokerName, msg, response, addr);
    27.                          if (context != null && sendResult != null) {
    28.                              context.setSendResult(sendResult);
    29.                              context.getProducer().executeSendMessageHookAfter(context);
    30.                          }
    31.                      } catch (Throwable e) {
    32.                      }
    34.                      producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), false);
    35.                      return;
    36.                  }
    38.                  if (response != null) {
    39.                      try {
    40.                          SendResult sendResult = MQClientAPIImpl.this.processSendResponse(brokerName, msg, response, addr);
    41.                          assert sendResult != null;
    42.                          if (context != null) {
    43.                              context.setSendResult(sendResult);
    44.                              context.getProducer().executeSendMessageHookAfter(context);
    45.                          }
    47.                          try {
    48.                              sendCallback.onSuccess(sendResult);
    49.                          } catch (Throwable e) {
    50.                          }
    52.                          producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), false);
    53.                      } catch (Exception e) {
    54.                          producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), true);
    55.                          onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
    56.                              retryTimesWhenSendFailed, times, e, context, false, producer);
    57.                      }
    58.                  } else {
    59.                      producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), true);
    60.                      if (!responseFuture.isSendRequestOK()) {
    61.                          MQClientException ex = new MQClientException("send request failed", responseFuture.getCause());
    62.                          onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
    63.                              retryTimesWhenSendFailed, times, ex, context, true, producer);
    64.                      } else if (responseFuture.isTimeout()) {
    65.                          MQClientException ex = new MQClientException("wait response timeout " + responseFuture.getTimeoutMillis() + "ms",
    66.                              responseFuture.getCause());
    67.                          onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
    68.                              retryTimesWhenSendFailed, times, ex, context, true, producer);
    69.                      } else {
    70.                          MQClientException ex = new MQClientException("unknow reseaon", responseFuture.getCause());
    71.                          onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
    72.                              retryTimesWhenSendFailed, times, ex, context, true, producer);
    73.                      }
    74.                  }
    75.              }
    76.          });
    77.      } catch (Exception ex) {
    78.          long cost = System.currentTimeMillis() - beginStartTime;
    79.          producer.updateFaultItem(brokerName, cost, true);
    80.          onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
    81.                  retryTimesWhenSendFailed, times, ex, context, true, producer);
    82.      }
    83.  }

3.4.2.5. 消息接收

生产者将消息发送至Broker后，Netty的code是RequestCode._SEND_MESSAGE，在Broker中通过_org.apache.rocketmq.broker.processor.SendMessageProcessor.processRequest()方法接收消息发送请求

  1. @Override
  2. public RemotingCommand processRequest(ChannelHandlerContext ctx,
  3.                                       RemotingCommand request) throws RemotingCommandException {
  4.     RemotingCommand response = null;
  5.     try {
  6.         // 处理消息发送请求
  7.         response = asyncProcessRequest(ctx, request).get();
  8.     } catch (InterruptedException | ExecutionException e) {
  9.         log.error("process SendMessage error, request : " + request.toString(), e);
  10.     }
  11.     return response;
  12. }

消息接收核心入口方法

  1. public CompletableFuture<RemotingCommand> asyncProcessRequest(ChannelHandlerContext ctx,
  2.                                                                   RemotingCommand request) throws RemotingCommandException {
  3.     final SendMessageContext mqtraceContext;
  4.     switch (request.getCode()) {
  5.         case RequestCode.CONSUMER_SEND_MSG_BACK:
  6.             return this.asyncConsumerSendMsgBack(ctx, request);
  7.         default:
  8.             // 解析消息发送对象，转换为SendMessageRequestHeader 对象
  9.             SendMessageRequestHeader requestHeader = parseRequestHeader(request);
  10.             if (requestHeader == null) {
  11.                 return CompletableFuture.completedFuture(null);
  12.             }
  13.             // 构建消息对象
  14.             mqtraceContext = buildMsgContext(ctx, requestHeader);
  15.             // 执行钩子函数（前置处理）
  16.             this.executeSendMessageHookBefore(ctx, request, mqtraceContext);
  17.             if (requestHeader.isBatch()) {
  18.                 // 批量消息异步发送
  19.                 return this.asyncSendBatchMessage(ctx, request, mqtraceContext, requestHeader);
  20.             } else {
  21.                 // 单个消息异步发送
  22.                 return this.asyncSendMessage(ctx, request, mqtraceContext, requestHeader);
  23.             }
  24.     }
  25. }