• RocketMQ消息发送源码解析

    Producer作为生产者是RocketMQ的重要组成部分，下面我们通过一个消息发送的例子来说明Producer的工作原理。

    一个消息发送的例子：

    1. // 使用GroupName初始化Producer
    2. DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    3. // 指定NameSrv的地址: 也可以通过环境变量NAMESRV_ADDR来指定，则不需要下面这一行。
    4. producer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
    5. // 启动实例
    6. producer.start();
    7. try {
    8.     // 创建消息实例,指定 topic, tag, message body.
    9.     Message msg = new Message("TopicTest"/* Topic */,
    10.         "TagA"/* Tag */,
    11.         ("Hello RocketMQ !!").getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
    12.     );
    13.     // 发送消息给Broker
    14.     SendResult sendResult = producer.send(msg);
    15.     System.out.printf("%s%n", sendResult);
    16. } catch (Exception e) {
    17.     e.printStackTrace();
    18.     Thread.sleep(1000);
    19. }
    20. // 关闭生产者
    21. producer.shutdown();

    通过上面的例子可以看出，Producer的消息发送流程如下：

       1、首先创建一个DefaultMQProducer 实例，然后设置NameSrv的地址，再启动实例。

       2、当需要发送消息时，创建消息实例，然后发送消息给Broker。

    可见，Producer主要是通过类DefaultMQProducer来实现发送流程。DefaultMQProducer的构造函数如下：

    1. public DefaultMQProducer(final String producerGroup) {
    2.         this(producerGroup, null);
    3.     }
    4. public DefaultMQProducer(final String producerGroup, RPCHook rpcHook) {
    5.         this.producerGroup = producerGroup;
    6.         defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
    7.     }

    在DefaultMQProducer的构造函数中，主要是创建了类DefaultMQProducerImpl的实例，通过类DefaultMQProducerImpl的实例来实现各种逻辑。

    DefaultMQProducerImpl#start()方法分析

    1. public void start(final boolean startFactory) throws MQClientException {
    2.         switch (this.serviceState) {
    3.             case CREATE_JUST:
    4.                 // 标记初始化失败，这个技巧不错。
    5.                 this.serviceState = ServiceState.START_FAILED;
    6.  
    7.                 this.checkConfig();
    8.  
    9.                 if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
    10.                     this.defaultMQProducer.changeInstanceNameToPID();
    11.                 }
    12.  
    13.                 // 获取MQClient 对象
    14.                 this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);
    15.  
    16.                 // 注册Producer
    17.                 boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
    18.                 if (!registerOK) {
    19.                     this.serviceState = ServiceState.CREATE_JUST;
    20.                     throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
    21.                         + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
    22.                         null);
    23.                 }
    24.  
    25.                 //把 MixAll.DEFAULT_TOPIC 放入其中
    26.                 this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
    27.  
    28.                 // 启动MQClient对象
    29.                 if (startFactory) {
    30.                     mQClientFactory.start();
    31.                 }
    32.  
    33.                 log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
    34.                     this.defaultMQProducer.isSendMessageWithVIPChannel());
    35.  
    36.                 // 标记初始化成功
    37.                 this.serviceState = ServiceState.RUNNING;
    38.                 break;
    39.             case RUNNING:
    40.             case START_FAILED:
    41.             case SHUTDOWN_ALREADY:
    42.                 throw new MQClientException("The producer service state not OK, maybe started once, "//
    43.                     + this.serviceState//
    44.                     + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
    45.                     null);
    46.             default:
    47.                 break;
    48.         }
    49.  
    50.         this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
    51.     }

    逻辑流程：

    1. 初始化mQClientFactory为MQClientInstance，并将该实例加入factoryTable
    2. 将producer注册到MQClientInstance.producerTable
    3. 保存topic对应的路由信息
    4. 启动MQClientInstance

    关于启动MQClientInstance有如下逻辑：

    1. 启动Netty客户端，注意这里并没有创建连接，在producer发送消息的时候创建连接
    2. 启动一系列定时任务
    3. 在while循环里不间断拉取消息，以后台线程方式运行
    4. 给consumer分配队列，以后台线程运行

    关于启动一系列定时任务有：

    1. 2分钟获取一次NameServer地址
    2. 默认30S更新一次topic的路由信息，频率可配置
    3. 30秒对Broker发送一次心跳检测，并将下线的broker删除
    4. 5秒持久化一次consumer的offset
    5. 每分钟调整线程池大小，不过里面代码注释掉了

    定时任务使用的是scheduleAtFixedRate，如果上一次任务超时则一下次任务会立即执行。

    DefaultMQProducer消息发送

    DefaultMQProducer的消息发送主要是通过DefaultMQProducerImpl的send方法来实现的，其流程如下：

    1. /**
    2.      * 发送消息。
    3.      * 1. 获取消息路由信息
    4.      * 2. 选择要发送到的消息队列
    5.      * 3. 执行消息发送核心方法
    6.      * 4. 对发送结果进行封装返回
    7.      *
    8.      * @param msg               消息
    9.      * @param communicationMode 通信模式
    10.      * @param sendCallback      发送回调
    11.      * @param timeout           发送消息请求超时时间
    12.      * @return 发送结果
    13.      * @throws MQClientException 当Client发生异常
    14.      * @throws RemotingException 当请求发生异常
    15.      * @throws MQBrokerException 当Broker发生异常
    16.      * @throws InterruptedException 当线程被打断
    17.      */
    18.     private SendResult sendDefaultImpl(Message msg,
    19.                                        final CommunicationMode communicationMode,
    20.                                        final SendCallback sendCallback,
    21.                                        final long timeout
    22.     ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
    23.         // 校验 Producer 处于运行状态
    24.         this.makeSureStateOK();
    25.         // 校验消息格式
    26.         Validators.checkMessage(msg, this.defaultMQProducer);
    27.         //
    28.         final long invokeID = random.nextLong(); // 调用编号；用于下面打印日志，标记为同一次发送消息
    29.         long beginTimestampFirst = System.currentTimeMillis();
    30.         long beginTimestampPrev = beginTimestampFirst;
    31.         @SuppressWarnings("UnusedAssignment")
    32.         long endTimestamp = beginTimestampFirst;
    33.         // 获取 Topic路由信息
    34.         TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
    35.         if (topicPublishInfo != null && topicPublishInfo.ok()) {
    36.             MessageQueue mq = null; // 最后选择消息要发送到的队列
    37.             Exception exception = null;
    38.             SendResult sendResult = null; // 最后一次发送结果
    39.             int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1; // 同步3次调用
    40.             int times = 0; // 第几次发送
    41.             String[] brokersSent = new String[timesTotal]; // 存储每次发送消息选择的broker名
    42.             // 循环调用发送消息，直到成功
    43.             for (; times < timesTotal; times++) {
    44.                 String lastBrokerName = null == mq ? null : mq.getBrokerName();
    45.                 @SuppressWarnings("SpellCheckingInspection")
    46.                 // 选择消息要发送到的队列,默认策略下,按顺序轮流发送,当一次发送失败时,按顺序选择下一个Broker的MessageQueue
    47.                 MessageQueue tmpmq = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
    48.                 if (tmpmq != null) {
    49.                     mq = tmpmq;
    50.                     brokersSent[times] = mq.getBrokerName();
    51.                     try {
    52.                         beginTimestampPrev = System.currentTimeMillis();
    53.                         // 调用发送消息核心方法
    54.                         sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout);
    55.                         endTimestamp = System.currentTimeMillis();
    56.                         // 更新Broker可用性信息,发送时间超过550ms后会有不可用时长，至少30S,不可用时间只有在开启了延迟容错机制才有效果
    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.                                     // 同步发送成功但存储有问题时 && 配置存储异常时重新发送开关 时，进行重试
    66.                                     if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
    67.                                         continue;
    68.                                     }
    69.                                 }
    70.                                 return sendResult;
    71.                             default:
    72.                                 break;
    73.                         }
    74.                     } catch (RemotingException e) { // 打印异常，更新Broker可用性信息,停用10M，更新继续循环
    75.                         endTimestamp = System.currentTimeMillis();
    76.                         this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
    77.                         log.warn(String
    78.                             .format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev,
    79.                                 mq), e);
    80.                         log.warn(msg.toString());
    81.                         exception = e;
    82.                         continue;
    83.                     } catch (MQClientException e) { // 打印异常，更新Broker可用性信息,停用10M，继续循环
    84.                         endTimestamp = System.currentTimeMillis();
    85.                         this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
    86.                         log.warn(String
    87.                             .format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev,
    88.                                 mq), e);
    89.                         log.warn(msg.toString());
    90.                         exception = e;
    91.                         continue;
    92.                     } catch (MQBrokerException e) { // 打印异常，更新Broker可用性信息，部分情况下的异常，直接返回，结束循环
    93.                         endTimestamp = System.currentTimeMillis();
    94.                         this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
    95.                         log.warn(String
    96.                             .format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev,
    97.                                 mq), e);
    98.                         log.warn(msg.toString());
    99.                         exception = e;
    100.                         switch (e.getResponseCode()) {
    101.                             // 如下异常continue，进行发送消息重试
    102.                             case ResponseCode.TOPIC_NOT_EXIST:
    103.                             case ResponseCode.SERVICE_NOT_AVAILABLE:
    104.                             case ResponseCode.SYSTEM_ERROR:
    105.                             case ResponseCode.NO_PERMISSION:
    106.                             case ResponseCode.NO_BUYER_ID:
    107.                             case ResponseCode.NOT_IN_CURRENT_UNIT:
    108.                                 continue;
    109.                                 // 如果有发送结果，进行返回，否则，抛出异常；
    110.                             default:
    111.                                 if (sendResult != null) {
    112.                                     return sendResult;
    113.                                 }
    114.                                 throw e;
    115.                         }
    116.                     } catch (InterruptedException e) {
    117.                         endTimestamp = System.currentTimeMillis();
    118.                         this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
    119.                         log.warn(String.format("sendKernelImpl exception, throw exception, InvokeID: %s, RT: %sms, Broker: %s", invokeID,
    120.                             endTimestamp - beginTimestampPrev, mq), e);
    121.                         log.warn(msg.toString());
    122.                         throw e;
    123.                     }
    124.                 } else {
    125.                     break;
    126.                 }
    127.             }
    128.             // 返回发送结果
    129.             if (sendResult != null) {
    130.                 return sendResult;
    131.             }
    132.             // 根据不同情况，抛出不同的异常
    133.             String info = String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s", times,
    134.                 System.currentTimeMillis() - beginTimestampFirst,
    135.                 msg.getTopic(), Arrays.toString(brokersSent)) + FAQUrl.suggestTodo(FAQUrl.SEND_MSG_FAILED);
    136.             MQClientException mqClientException = new MQClientException(info, exception);
    137.             if (exception instanceof MQBrokerException) {
    138.                 mqClientException.setResponseCode(((MQBrokerException)exception).getResponseCode());
    139.             } else if (exception instanceof RemotingConnectException) {
    140.                 mqClientException.setResponseCode(ClientErrorCode.CONNECT_BROKER_EXCEPTION);
    141.             } else if (exception instanceof RemotingTimeoutException) {
    142.                 mqClientException.setResponseCode(ClientErrorCode.ACCESS_BROKER_TIMEOUT);
    143.             } else if (exception instanceof MQClientException) {
    144.                 mqClientException.setResponseCode(ClientErrorCode.BROKER_NOT_EXIST_EXCEPTION);
    145.             }
    146.             throw mqClientException;
    147.         }
    148.         // Namesrv找不到异常
    149.         List nsList = this.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList();
    150.         if (null == nsList || nsList.isEmpty()) {
    151.             throw new MQClientException(
    152.                 "No name server address, please set it." + FAQUrl.suggestTodo(FAQUrl.NAME_SERVER_ADDR_NOT_EXIST_URL), null).setResponseCode(
    153.                 ClientErrorCode.NO_NAME_SERVER_EXCEPTION);
    154.         }
    155.         // 消息路由找不到异常
    156.         throw new MQClientException("No route info of this topic, " + msg.getTopic() + FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO),
    157.             null).setResponseCode(ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION);
    158.     }

    同步发送消息的逻辑还是比较简单的：

    1. 如果没指定超时时间，则使用默认的3S超时，需要注意下这里的超时时间指是超时时间而非单次重试的时间(4.3.0版本为单次超时时间)
    2. 校验Producer服务状态是否正确，topic是否有效，消息体长度是否合法
    3. 获取topic路由信息，先在本地查找，本地没有查寻NameServer
    4. 根据每个broker处理的时间选择发送队列，每次发送完消息后会记录broker与每次发送消息所花时间
    5. 发送消息。如果是同步发送最多会重试3次，也可通过配置进行调整
    6. 记录此broker与发送消息所花时间的对应关系

    总结一下Producer发送消息流程：

    1. 获取消息路由信息
2. 选择要发送到的消息队列
3. 执行消息发送核心方法
4. 对发送结果进行封装返回

    DefaultMQProducerImpl#tryToFindTopicPublishInfo

    DefaultMQProducerImpl查找路由的流程如下：

    1. /**
    2.      * 获取 Topic发布信息
    3.      * 如果获取不到，或者状态不正确，则从 Namesrv获取一次
    4.      *
    5.      * @param topic Topic
    6.      * @return topic 信息
    7.      */
    8.     private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
    9.         // 缓存中获取 Topic发布信息
    10.         TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
    11.         // 当无可用的 Topic发布信息时，从Namesrv获取一次
    12.         if (null == topicPublishInfo || !topicPublishInfo.ok()) {
    13.             this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
    14.             this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
    15.             topicPublishInfo = this.topicPublishInfoTable.get(topic);
    16.         }
    17.         // 若获取的 Topic发布信息时候可用，则返回
    18.         if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
    19.             return topicPublishInfo;
    20.         } else { // 使用 {@link DefaultMQProducer#createTopicKey} 对应的 Topic发布信息。用于 Topic发布信息不存在 && Broker支持自动创建Topic
    21.             this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
    22.             topicPublishInfo = this.topicPublishInfoTable.get(topic);
    23.             return topicPublishInfo;
    24.         }
    25.     }

    TopicPublishInfo的定义如下：

    1. /**
    2.  * Topic发布信息
    3.  */
    4. public class TopicPublishInfo {
    5.  
    6.     /**
    7.      * 是否顺序消息
    8.      */
    9.     private boolean orderTopic = false;
    10.     /**
    11.      * 是否有路由信息
    12.      */
    13.     private boolean haveTopicRouterInfo = false;
    14.     /**
    15.      * 消息队列数组
    16.      */
    17.     private List messageQueueList = new ArrayList();
    18.     /**
    19.      * 线程变量（Index）
    20.      */
    21.     private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();
    22.     /**
    23.      * Topic消息路由信息
    24.      */
    25.     private TopicRouteData topicRouteData;
    26. }

    使用函数updateTopicRouteInfoFromNameServer来获取路由信息

    1. /**
    2.      * 更新单个 Topic 路由信息
    3.      * 若 isDefault=true && defaultMQProducer!=null 时，使用{@link DefaultMQProducer#createTopicKey}
    4.      *
    5.      * @param topic             Topic
    6.      * @param isDefault         是否默认
    7.      * @param defaultMQProducer producer
    8.      * @return 是否更新成功
    9.      */
    10.     public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault, DefaultMQProducer defaultMQProducer) {
    11.         try {
    12.             if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
    13.                 try {
    14.                     TopicRouteData topicRouteData;
    15.                     // 使用默认TopicKey获取TopicRouteData。
    16.                     // 当broker开启自动创建topic开关时，会使用MixAll.DEFAULT_TOPIC进行创建。
    17.                     // 当producer的createTopic为MixAll.DEFAULT_TOPIC时，则可以获得TopicRouteData。
    18.                     // 目的：用于新的topic，发送消息时，未创建路由信息，先使用createTopic的路由信息，等到发送到broker时，进行自动创建。
    19.                     // @see TopicConfigManager
    20.                     if (isDefault && defaultMQProducer != null) {
    21.                         topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(), 1000 * 3);
    22.                         if (topicRouteData != null) {
    23.                             for (QueueData data : topicRouteData.getQueueDatas()) {
    24.                                 int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
    25.                                 data.setReadQueueNums(queueNums);
    26.                                 data.setWriteQueueNums(queueNums);
    27.                             }
    28.                         }
    29.                     } else {
    30.                         topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);
    31.                     }
    32.                     if (topicRouteData != null) {
    33.                         TopicRouteData old = this.topicRouteTable.get(topic);
    34.                         boolean changed = topicRouteDataIsChange(old, topicRouteData);
    35.                         if (!changed) {
    36.                             changed = this.isNeedUpdateTopicRouteInfo(topic);
    37.                         } else {
    38.                             log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);
    39.                         }
    40.  
    41.                         if (changed) {
    42.                             // 克隆对象的原因：topicRouteData会被设置到下面的publishInfo/subscribeInfo
    43.                             TopicRouteData cloneTopicRouteData = topicRouteData.cloneTopicRouteData();
    44.  
    45.                             // 更新 Broker 地址相关信息,当某个Broker心跳超时后,会被从BrokerData的brokerAddrs中移除(由Namesrv定时操作)
    46.                             // Namesrv存在Slave的BrokerData,所以brokerAddrTable含有Slave的brokerAddr
    47.                             for (BrokerData bd : topicRouteData.getBrokerDatas()) {
    48.                                 this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
    49.                             }
    50.  
    51.                             // 更新生产者里的TopicPublishInfo,Slave在注册Broker时不会生成QueueData,但会生成BrokerData
    52.                             TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);
    53.                             publishInfo.setHaveTopicRouterInfo(true);
    54.                             for (Entry entry : this.producerTable.entrySet()) {
    55.                                 MQProducerInner impl = entry.getValue();
    56.                                 if (impl != null) {
    57.                                     impl.updateTopicPublishInfo(topic, publishInfo);
    58.                                 }
    59.                             }
    60.  
    61.                             // 更新订阅者(消费者)里的队列信息,Slave在注册Broker时不会生成QueueData,但会生成BrokerData
    62.                             Set subscribeInfo = topicRouteData2TopicSubscribeInfo(topic, topicRouteData);
    63.                             for (Entry entry : this.consumerTable.entrySet()) {
    64.                                 MQConsumerInner impl = entry.getValue();
    65.                                 if (impl != null) {
    66.                                     impl.updateTopicSubscribeInfo(topic, subscribeInfo);
    67.                                 }
    68.                             }
    69.                             log.info("topicRouteTable.put TopicRouteData[{}]", cloneTopicRouteData);
    70.                             this.topicRouteTable.put(topic, cloneTopicRouteData);
    71.                             return true;
    72.                         }
    73.                     } else {
    74.                         log.warn("updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {}", topic);
    75.                     }
    76.                 } catch (Exception e) {
    77.                     if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) && !topic.equals(MixAll.DEFAULT_TOPIC)) {
    78.                         log.warn("updateTopicRouteInfoFromNameServer Exception", e);
    79.                     }
    80.                 } finally {
    81.                     this.lockNamesrv.unlock();
    82.                 }
    83.             } else {
    84.                 log.warn("updateTopicRouteInfoFromNameServer tryLock timeout {}ms", LOCK_TIMEOUT_MILLIS);
    85.             }
    86.         } catch (InterruptedException e) {
    87.             log.warn("updateTopicRouteInfoFromNameServer Exception", e);
    88.         }
    89.  
    90.         return false;
    91.     }

    选择队列

    1. /**
    2.      * 根据 Topic发布信息 选择一个消息队列
    3.      * 默认情形下向所有Broker的MessageQueue按顺序轮流发送
    4.      *
    5.      * @param tpInfo         Topic发布信息
    6.      * @param lastBrokerName
    7.      * @return 消息队列
    8.      */
    9.     public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
    10.         if (this.sendLatencyFaultEnable) {  //如果开启了延迟容错机制，默认未开启
    11.             try {
    12.                 //循环所有MessageQueue
    13.                 // 当 lastBrokerName == null 时，获取第一个可用的MessageQueue
    14.                 // 当 lastBrokerName != null 时, 获取 brokerName=lastBrokerName && 可用的MessageQueue
    15.                 int index = tpInfo.getSendWhichQueue().getAndIncrement();
    16.                 for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
    17.                     int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
    18.                     if (pos < 0) {
    19.                         pos = 0;
    20.                     }
    21.                     MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
    22.                     if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
    23.                         if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName)) {
    24.                             return mq;
    25.                         }
    26.                     }
    27.                 }
    28.                 // 选择一个相对好的broker，并获得其对应的一个消息队列，按 可用性 > 延迟 > 开始可用时间 选择
    29.                 final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
    30.                 int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
    31.                 if (writeQueueNums > 0) {
    32.                     final MessageQueue mq = tpInfo.selectOneMessageQueue();
    33.                     if (notBestBroker != null) {
    34.                         mq.setBrokerName(notBestBroker);
    35.                         mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
    36.                     }
    37.                     return mq;
    38.                 } else {
    39.                     latencyFaultTolerance.remove(notBestBroker);
    40.                 }
    41.             } catch (Exception e) {
    42.                 log.error("Error occurred when selecting message queue", e);
    43.             }
    44.             // 选择一个消息队列，不考虑队列的可用性
    45.             return tpInfo.selectOneMessageQueue();
    46.         }
    47.         // 默认情况下,获得 lastBrokerName 对应的一个消息队列，不考虑该队列的可用性
    48.         return tpInfo.selectOneMessageQueue(lastBrokerName);
    49.     }
     

    DefaultMQProducerImpl#sendKernelImpl：发送消息核心方法
     

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

    最终通过MQClientAPIImpl的发送函数sendMessage将消息发送给Broker。
     

    
                
    
                    
    
                
    • 

                
  • 

                    相关阅读:
    
                    
Zigbee开发笔记- IAR的使用                            
    
java获取文件编码方式                            
    
单片机对比：选择最适合你的单片机                            
    
了解《单链表》看这篇就好了（内含动图）！！！                            
    
Vue语法                            
    
OpenCode 对接实践：从独立进程到共享 Runtime 的架构演进                            
    
GeoTools的AStar算法实现，自定义Node及Edge                            
    
累加出整个范围所有的数最少还需要几个数                            
    
灵芝酸大鼠血清白蛋白纳米粒|茯苓酸小麦麦清白蛋白纳米粒雷公藤内酯醇-牛血清白蛋白纳米粒(TRD-BSA NPs)                            
    
Python tkinter -- 第15章 Combobox                            
    
                        
                
    • 
                
  • 
                    原文地址：https://blog.csdn.net/bao2901203013/article/details/126091368
                
    •