devbox
我家自动化
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RocketMq源码解读（四）消息发送_send [6] times, still failed, cost [619]ms, topic:

作者：我家自动化 | 2024-03-17 16:06:25

赞

踩

send [6] times, still failed, cost [619]ms, topic: galaxy-sms-cache-sync, br

前两章内容中，介绍了Producer的启动过程，然后是如何创建topic，本章内容将主要集中在如何发送消息到broker 

 一、消息的投递

消息的投递分为普通消息的投递和事务消息的投递

  1. public class DefaultMQProducer extends ClientConfig implements MQProducer{
  2.     // 所有的实际操作委托给这个实现
  3.     protected final transient DefaultMQProducerImpl defaultMQProducerImpl;
  4.     // 默认topic队列数
  5.     private volatile int defaultTopicQueueNums = 4;
  6.     private int sendMsgTimeout = 3000;
  7.     private int compressMsgBodyOverHowmuch = 1024 * 4;
  8.     private int retryTimesWhenSendFailed = 2;
  9.     private int retryTimesWhenSendAsyncFailed = 2;
  10.     private boolean retryAnotherBrokerWhenNotStoreOK = false;
  11.     private int maxMessageSize = 1024 * 1024 * 4;
  12.     ...
  13.     @Override
  14.     public SendResult send(Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  15.         return this.defaultMQProducerImpl.send(msg);
  16.     }
  17.     @Override
  18.     public SendResult send(Message msg, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  19.         return this.defaultMQProducerImpl.send(msg, timeout);
  20.     }
  21.     @Override
  22.     public void send(Message msg, SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {
  23.         this.defaultMQProducerImpl.send(msg, sendCallback);
  24.     }
  25.     @Override
  26.     public SendResult send(Message msg, MessageQueue mq, long timeout)
  27.         throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  28.         return this.defaultMQProducerImpl.send(msg, mq, timeout);
  29.     }
  30.     @Override
  31.     public void send(Message msg, MessageQueue mq, SendCallback sendCallback, long timeout)
  32.         throws MQClientException, RemotingException, InterruptedException {
  33.         this.defaultMQProducerImpl.send(msg, mq, sendCallback, timeout);
  34.     }
  35.     @Override
  36.     public void send(Message msg, MessageQueueSelector selector, Object arg, SendCallback sendCallback, long timeout)
  37.         throws MQClientException, RemotingException, InterruptedException {
  38.         this.defaultMQProducerImpl.send(msg, selector, arg, sendCallback, timeout);
  39.     }
  40.  
  41.     @Override
  42.     public void sendOneway(Message msg, MessageQueueSelector selector, Object arg)
  43.         throws MQClientException, RemotingException, InterruptedException {
  44.         this.defaultMQProducerImpl.sendOneway(msg, selector, arg);
  45.     }
  46.     @Override
  47.     public TransactionSendResult sendMessageInTransaction(Message msg, LocalTransactionExecuter tranExecuter, final Object arg)
  48.         throws MQClientException {
  49.         throw new RuntimeException("sendMessageInTransaction not implement, please use TransactionMQProducer class");
  50.     }
  51.     ...
  52. }

二、消息投递的具体实现过程

1.、同样的方式，消息的具体投递是委托给DefaultMQProducer内部的DefaultMQProducerImpl

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

2、关键函数获取topic的路由信息tryToFindTopicPublishInfo，因为在创建topic时，会将topic和broker的信息同步到NameSrv，所以这里可以从NameSrv获取发送消息的topic理由信息。

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

接着看更新topic的路由信息

  1. public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault, DefaultMQProducer defaultMQProducer) {
  2.     try {
  3.         if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
  4.            try {
  5.                TopicRouteData topicRouteData;
  6.                if (isDefault && defaultMQProducer != null) { 
  7.                 // 使用默认TopicKey获取TopicRouteData。
  8.                 // 当broker开启自动创建topic开关时，会使用MixAll.DEFAULT_TOPIC进行创建。
  9.                 // 当producer的createTopic为MixAll.DEFAULT_TOPIC时，则可以获得TopicRouteData。
  10.                // 目的：用于新的topic，发送消息时，未创建路由信息，先使用createTopic的路由信息，等到发送到broker时，进行自动创建。
  11.                                                                   
  12.                topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(),1000 * 3);
  13.               if (topicRouteData != null) {
  14.                   for (QueueData data : topicRouteData.getQueueDatas()) {
  15.                       int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
  16.                       data.setReadQueueNums(queueNums);
  17.                       data.setWriteQueueNums(queueNums);
  18.                    }
  19.                }
  20.                } else {
  21.                    topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);
  22.                }
  23.                if (topicRouteData != null) {
  24.                     TopicRouteData old = this.topicRouteTable.get(topic);
  25.                     boolean changed = topicRouteDataIsChange(old, topicRouteData);
  26.                     if (!changed) {
  27.                         changed = this.isNeedUpdateTopicRouteInfo(topic);
  28.                     } else {
  29.                        log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);
  30.                     }
  31.                     if (changed) {
  32.                       TopicRouteData cloneTopicRouteData = topicRouteData.cloneTopicRouteData(); // 克隆对象的原因：topicRouteData会被设置到下面的publishInfo/subscribeInfo
  33.  
  34.                        // 更新 Broker 地址相关信息
  35.                        for (BrokerData bd : topicRouteData.getBrokerDatas()) {
  36.                            this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
  37.                        }
  38.                        // 更新发布topic的消息，这里可能涉及到创建MessageQueue
  39.                        {
  40.                           TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);
  41.                           publishInfo.setHaveTopicRouterInfo(true);
  42.                           for (Entry<String, MQProducerInner> entry : this.producerTable.entrySet()) {
  43.                              MQProducerInner impl = entry.getValue();
  44.                              if (impl != null) {
  45.                                 impl.updateTopicPublishInfo(topic, publishInfo);
  46.                              }
  47.                            }
  48.                        }
  49.                        // 这里会更新一次consumer的信息
  50.                        {
  51.                           Set<MessageQueue> subscribeInfo = topicRouteData2TopicSubscribeInfo(topic, topicRouteData);
  52.                           for (Entry<String, MQConsumerInner> entry : this.consumerTable.entrySet()) {
  53.                              MQConsumerInner impl = entry.getValue();
  54.                              if (impl != null) {
  55.                                  impl.updateTopicSubscribeInfo(topic, subscribeInfo);
  56.                              }
  57.                            }
  58.                         }
  59.                         log.info("topicRouteTable.put TopicRouteData[{}]", cloneTopicRouteData);
  60.                         this.topicRouteTable.put(topic, cloneTopicRouteData);
  61.                         return true;
  62.                    }
  63.              } else {
  64.                     log.warn("updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {}", topic);
  65.              }
  66.          } catch (Exception e) {
  67.              if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) && !topic.equals(MixAll.DEFAULT_TOPIC)) {
  68.                  log.warn("updateTopicRouteInfoFromNameServer Exception", e);
  69.               }
  70.           } finally {
  71.               this.lockNamesrv.unlock();
  72.           }
  73.        } else {
  74.            log.warn("updateTopicRouteInfoFromNameServer tryLock timeout {}ms", LOCK_TIMEOUT_MILLIS);
  75.        }
  76.    } catch (InterruptedException e) {
  77.        log.warn("updateTopicRouteInfoFromNameServer Exception", e);
  78.    }
  79.  
  80.    return false;
  81. }

3、因为一个topic可能有很多的消息队列MessageQueue，如何选取MessageQueue路由呢？这里会通过一种容错机制选取出最优的MessageQueue。默认的MQFaultStrategy实现

  1. // MQ 故障策略
  2. public class MQFaultStrategy {
  3.     /**
  4.      * 延迟故障容错，维护每个Broker的发送消息的延迟
  5.      * key：brokerName
  6.      */
  7.     private final LatencyFaultTolerance<String> latencyFaultTolerance = new LatencyFaultToleranceImpl();
  8.     /**
  9.      * 发送消息延迟容错开关
  10.      */
  11.     private boolean sendLatencyFaultEnable = false;
  12.     /**
  13.      * 延迟级别数组
  14.      */
  15.     private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
  16.     /**
  17.      * 不可用时长数组
  18.      */
  19.     private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};  
  20.     public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
  21.         if (this.sendLatencyFaultEnable) {
  22.             try {
  23.                 // 感觉是在MessageQueue中轮流选取
  24.                 int index = tpInfo.getSendWhichQueue().getAndIncrement();
  25.                 for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
  26.                     int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
  27.                     if (pos < 0)
  28.                         pos = 0;
  29.                     MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
  30.                     if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
  31.                         if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
  32.                             return mq;
  33.                     }
  34.                 }
  35.                 // 选择一个相对好的broker，并获得其对应的一个消息队列，不考虑该队列的可用性
  36.                 final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
  37.                 int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
  38.                 if (writeQueueNums > 0) {
  39.                     final MessageQueue mq = tpInfo.selectOneMessageQueue();
  40.                     if (notBestBroker != null) {
  41.                         mq.setBrokerName(notBestBroker);
  42.                         mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
  43.                     }
  44.                     return mq;
  45.                 } else {
  46.                     latencyFaultTolerance.remove(notBestBroker);
  47.                 }
  48.             } catch (Exception e) {
  49.                 log.error("Error occurred when selecting message queue", e);
  50.             }
  51.             // 选择一个消息队列，不考虑队列的可用性
  52.             return tpInfo.selectOneMessageQueue();
  53.         }
  54.         // 获得 lastBrokerName 对应的一个消息队列，不考虑该队列的可用性
  55.         return tpInfo.selectOneMessageQueue(lastBrokerName);
  56.     }
  57. }

4、发送消息的核心：非常主要一点，因为MessageQueue可能在不同的broker上，每一个brokerName可能有一组broker，所以这个时候会向这组broker的主broker投递消息。

  1. //发送消息的实际操作者
  2. private SendResult sendKernelImpl(final Message msg,
  3.                                   final MessageQueue mq,
  4.                                   final CommunicationMode communicationMode,
  5.                                   final SendCallback sendCallback,
  6.                                   final TopicPublishInfo topicPublishInfo,
  7.                                   final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  8.     long beginStartTime = System.currentTimeMillis();
  9.      //获得需要发送的broker地址，主要是发送给master
  10.     String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
  11.     if (null == brokerAddr) {
  12.         //重新拉取topic信息
  13.         tryToFindTopicPublishInfo(mq.getTopic());
  14.         brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
  15.     }
  16.     SendMessageContext context = null;
  17.     if (brokerAddr != null) {
  18.         //是否是vip的操作，主要是端口-2的操作，然后重新拼接addr
  19.         brokerAddr = MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
  20.         //发送的消息体
  21.         byte[] prevBody = msg.getBody();
  22.         try {
  23.             //for MessageBatch,ID has been set in the generating process
  24.             //是否是批量消息体
  25.             if (!(msg instanceof MessageBatch)) {
  26.                 MessageClientIDSetter.setUniqID(msg);
  27.             }
  28.  
  29.             int sysFlag = 0;
  30.             boolean msgBodyCompressed = false;
  31.             //压缩消息体
  32.             if (this.tryToCompressMessage(msg)) {
  33.                 sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
  34.                 msgBodyCompressed = true;
  35.             }
  36.  
  37.             //判断是否是事务消息，设置事务属性
  38.             final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
  39.             if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
  40.                 sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
  41.             }
  42.  
  43.             //验证自定义的hook
  44.             if (hasCheckForbiddenHook()) {
  45.                 CheckForbiddenContext checkForbiddenContext = new CheckForbiddenContext();
  46.                 checkForbiddenContext.setNameSrvAddr(this.defaultMQProducer.getNamesrvAddr());
  47.                 checkForbiddenContext.setGroup(this.defaultMQProducer.getProducerGroup());
  48.                 checkForbiddenContext.setCommunicationMode(communicationMode);
  49.                 checkForbiddenContext.setBrokerAddr(brokerAddr);
  50.                 checkForbiddenContext.setMessage(msg);
  51.                 checkForbiddenContext.setMq(mq);
  52.                 checkForbiddenContext.setUnitMode(this.isUnitMode());
  53.                 this.executeCheckForbiddenHook(checkForbiddenContext);
  54.             }
  55.  
  56.             //验证发送消息的hook
  57.             if (this.hasSendMessageHook()) {
  58.                 context = new SendMessageContext();
  59.                 context.setProducer(this);
  60.                 context.setProducerGroup(this.defaultMQProducer.getProducerGroup());
  61.                 context.setCommunicationMode(communicationMode);
  62.                 context.setBornHost(this.defaultMQProducer.getClientIP());
  63.                 context.setBrokerAddr(brokerAddr);
  64.                 context.setMessage(msg);
  65.                 context.setMq(mq);
  66.                 String isTrans = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
  67.                 if (isTrans != null && isTrans.equals("true")) {
  68.                     context.setMsgType(MessageType.Trans_Msg_Half);
  69.                 }
  70.  
  71.                 if (msg.getProperty("__STARTDELIVERTIME") != null || msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != null) {
  72.                     context.setMsgType(MessageType.Delay_Msg);
  73.                 }
  74.                 this.executeSendMessageHookBefore(context);
  75.             }
  76.  
  77.             //消息头的封装
  78.             SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
  79.             //消息发送组
  80.             requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
  81.             //消息的topic
  82.             requestHeader.setTopic(msg.getTopic());
  83.             //消息的默认topic
  84.             requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
  85.             //消息的默认queue数量
  86.            requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
  87.             //消息发送的queueid
  88.             requestHeader.setQueueId(mq.getQueueId());
  89.             //特殊标识
  90.             requestHeader.setSysFlag(sysFlag);
  91.             //消息创建的时间戳
  92.             requestHeader.setBornTimestamp(System.currentTimeMillis());
  93.             //标识
  94.             requestHeader.setFlag(msg.getFlag());
  95.             //消息的扩展属性
  96.             requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
  97.             //消息的消费
  98.             requestHeader.setReconsumeTimes(0);
  99.             //模型
  100.             requestHeader.setUnitMode(this.isUnitMode());
  101.             //批量
  102.             requestHeader.setBatch(msg instanceof MessageBatch);
  103.             //特殊标记的处理
  104.             if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
  105.                 String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
  106.                 if (reconsumeTimes != null) {
  107.                     requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
  108.                     MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
  109.                 }
  110.  
  111.                 String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(msg);
  112.                 if (maxReconsumeTimes != null) {
  113.                     requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
  114.                     MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_MAX_RECONSUME_TIMES);
  115.                 }
  116.             }
  117.  
  118.             SendResult sendResult = null;
  119.             //消息发送的模型
  120.             switch (communicationMode) {
  121.                 //异步发送
  122.                 case ASYNC:
  123.                     Message tmpMessage = msg;
  124.                     if (msgBodyCompressed) {
  125.                         //If msg body was compressed, msgbody should be reset using prevBody.
  126.                         //Clone new message using commpressed message body and recover origin massage.
  127.                         //Fix bug:https://github.com/apache/rocketmq-externals/issues/66
  128.                         tmpMessage = MessageAccessor.cloneMessage(msg);
  129.                         msg.setBody(prevBody);
  130.                     }
  131.                     long costTimeAsync = System.currentTimeMillis() - beginStartTime;
  132.                     if (timeout < costTimeAsync) {
  133.                         throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
  134.                     }
  135.                     //参数区别主要是回调相关的配置
  136.                     sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
  137.                         brokerAddr,
  138.                         mq.getBrokerName(),
  139.                         tmpMessage,
  140.                         requestHeader,
  141.                         timeout - costTimeAsync,
  142.                         communicationMode,
  143.                         sendCallback,
  144.                         topicPublishInfo,
  145.                         this.mQClientFactory,
  146.                         this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(),
  147.                         context,
  148.                         this);
  149.                     break;
  150.                 //同步及单发
  151.                 case ONEWAY:
  152.                 case SYNC:
  153.                     long costTimeSync = System.currentTimeMillis() - beginStartTime;
  154.                     if (timeout < costTimeSync) {
  155.                         throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
  156.                     }
  157.                     //消息的最终发送
  158.                     sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
  159.                         brokerAddr,
  160.                         mq.getBrokerName(),
  161.                         msg,
  162.                         requestHeader,
  163.                         timeout - costTimeSync,
  164.                         communicationMode,
  165.                         context,
  166.                         this);
  167.                     break;
  168.                 default:
  169.                     assert false;
  170.                     break;
  171.             }
  172.  
  173.             //是否有发送消息hook
  174.             if (this.hasSendMessageHook()) {
  175.                 context.setSendResult(sendResult);
  176.                 this.executeSendMessageHookAfter(context);
  177.             }
  178.  
  179.             return sendResult;
  180.         } catch (RemotingException e) {
  181.             if (this.hasSendMessageHook()) {
  182.                 context.setException(e);
  183.                 this.executeSendMessageHookAfter(context);
  184.             }
  185.             throw e;
  186.         } catch (MQBrokerException e) {
  187.             if (this.hasSendMessageHook()) {
  188.                 context.setException(e);
  189.                 this.executeSendMessageHookAfter(context);
  190.             }
  191.             throw e;
  192.         } catch (InterruptedException e) {
  193.             if (this.hasSendMessageHook()) {
  194.                 context.setException(e);
  195.                 this.executeSendMessageHookAfter(context);
  196.             }
  197.             throw e;
  198.         } finally {
  199.             msg.setBody(prevBody);
  200.         }
  201.     }
  202.  
  203.     throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
  204. }

6、组装netty的发送指令，开始消息的最终发送：

  1. //发送消息的核心操作实现
  2. public SendResult sendMessage(
  3.     final String addr,//地址
  4.     final String brokerName,//broker名称
  5.     final Message msg,//消息
  6.     final SendMessageRequestHeader requestHeader,//消息头
  7.     final long timeoutMillis,//超时
  8.     final CommunicationMode communicationMode,//发送模式
  9.     final SendCallback sendCallback,//发送后的回调
  10.     final TopicPublishInfo topicPublishInfo,//topic的配置
  11.     final MQClientInstance instance,//基于client的发送实例
  12.     final int retryTimesWhenSendFailed,//重试次数
  13.     final SendMessageContext context,//发送消息的上下文信息
  14.     final DefaultMQProducerImpl producer//发送消息者
  15. ) throws RemotingException, MQBrokerException, InterruptedException {
  16.     long beginStartTime = System.currentTimeMillis();
  17.     RemotingCommand request = null;
  18.     //协议转换
  19.     if (sendSmartMsg || msg instanceof MessageBatch) {
  20.         SendMessageRequestHeaderV2 requestHeaderV2 = SendMessageRequestHeaderV2.createSendMessageRequestHeaderV2(requestHeader);
  21.         request = RemotingCommand.createRequestCommand(msg instanceof MessageBatch ? RequestCode.SEND_BATCH_MESSAGE : RequestCode.SEND_MESSAGE_V2, requestHeaderV2);
  22.     } else {
  23.         request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, requestHeader);
  24.     }
  25.  
  26.     request.setBody(msg.getBody());
  27.  
  28.     switch (communicationMode) {
  29.         case ONEWAY:
  30.             this.remotingClient.invokeOneway(addr, request, timeoutMillis);
  31.             return null;
  32.         case ASYNC:
  33.             final AtomicInteger times = new AtomicInteger();
  34.             long costTimeAsync = System.currentTimeMillis() - beginStartTime;
  35.             if (timeoutMillis < costTimeAsync) {
  36.                 throw new RemotingTooMuchRequestException("sendMessage call timeout");
  37.             }
  38.             this.sendMessageAsync(addr, brokerName, msg, timeoutMillis - costTimeAsync, request, sendCallback, topicPublishInfo, instance,
  39.                 retryTimesWhenSendFailed, times, context, producer);
  40.             return null;
  41.         case SYNC:
  42.             long costTimeSync = System.currentTimeMillis() - beginStartTime;
  43.             if (timeoutMillis < costTimeSync) {
  44.                 throw new RemotingTooMuchRequestException("sendMessage call timeout");
  45.             }
  46.             return this.sendMessageSync(addr, brokerName, msg, timeoutMillis - costTimeSync, request);
  47.         default:
  48.             assert false;
  49.             break;
  50.     }
  51.  
  52.     return null;
  53. }
  54.  
  55. // 同步调用发送方法
  56. private SendResult sendMessageSync(
  57.     final String addr,
  58.     final String brokerName,
  59.     final Message msg,
  60.     final long timeoutMillis,
  61.     final RemotingCommand request
  62. ) throws RemotingException, MQBrokerException, InterruptedException {
  63.     //同步发送消息
  64.     RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
  65.     assert response != null;
  66.  
  67.     //处理请求的相应结果
  68.     return this.processSendResponse(brokerName, msg, response);
  69. }
  70.  
  71. //网络通信的核心方法实现，同步
  72. @Override
  73. public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)
  74.     throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException {
  75.     long beginStartTime = System.currentTimeMillis();
  76.  
  77.     //获得通信的channel
  78.     final Channel channel = this.getAndCreateChannel(addr);
  79.  
  80.     //验证channel的有效性
  81.     if (channel != null && channel.isActive()) {
  82.         try {
  83.             //前置处理
  84.             doBeforeRpcHooks(addr, request);
  85.  
  86.             long costTime = System.currentTimeMillis() - beginStartTime;
  87.             if (timeoutMillis < costTime) {
  88.                 throw new RemotingTimeoutException("invokeSync call timeout");
  89.             }
  90.  
  91.             //执行方法调用
  92.             RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
  93.  
  94.             //后置处理
  95.             doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
  96.             return response;
  97.         } catch (RemotingSendRequestException e) {
  98.             log.warn("invokeSync: send request exception, so close the channel[{}]", addr);
  99.             this.closeChannel(addr, channel);
  100.             throw e;
  101.         } catch (RemotingTimeoutException e) {
  102.             if (nettyClientConfig.isClientCloseSocketIfTimeout()) {
  103.                 this.closeChannel(addr, channel);
  104.                 log.warn("invokeSync: close socket because of timeout, {}ms, {}", timeoutMillis, addr);
  105.             }
  106.             log.warn("invokeSync: wait response timeout exception, the channel[{}]", addr);
  107.             throw e;
  108.         }
  109.     } else {
  110.         //如果异常则删除执行的channel的操作
  111.         this.closeChannel(addr, channel);
  112.         throw new RemotingConnectException(addr);
  113.     }
  114. }
  115.  
  116. //同步操作的核心
  117. public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
  118.     final long timeoutMillis)
  119.     throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
  120.     final int opaque = request.getOpaque();
  121.  
  122.     try {
  123.         //异步调用封装，处理netty通信后的结果
  124.         final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis, null, null);
  125.  
  126.         //结果的并发处理及通知，及后续的异常处理收集
  127.         this.responseTable.put(opaque, responseFuture);
  128.         final SocketAddress addr = channel.remoteAddress();
  129.         //netty的消息发送，只负责发送
  130.         channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
  131.             @Override
  132.             public void operationComplete(ChannelFuture f) throws Exception {
  133.                 //操作成功后马上返回
  134.                 if (f.isSuccess()) {
  135.                     responseFuture.setSendRequestOK(true);
  136.                     return;
  137.                 } else {
  138.                     responseFuture.setSendRequestOK(false);
  139.                 }
  140.                 //同步处理后删除对应的请求对象
  141.                 responseTable.remove(opaque);
  142.                 responseFuture.setCause(f.cause());
  143.                 responseFuture.putResponse(null);
  144.                 log.warn("send a request command to channel <" + addr + "> failed.");
  145.             }
  146.         });
  147.  
  148.         //相应结果的处理，实际的操作在NettyClientHandler中实现，根据opaque来更新结果
  149.         RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
  150.         if (null == responseCommand) {
  151.             if (responseFuture.isSendRequestOK()) {
  152.                 throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
  153.                     responseFuture.getCause());
  154.             } else {
  155.                 throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
  156.             }
  157.         }
  158.  
  159.         return responseCommand;
  160.     } finally {
  161.         this.responseTable.remove(opaque);
  162.     }
  163. }

如果消息是异步发送机制，会通过sendCallBack进行回调处理。

7、broker接收到消息发送的指令后，会做消息的持久化处理，后面在broker环节将重要介绍。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家自动化/article/detail/256942
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号