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

RocketMQ原理学习---生产者普通消息发送_.setmaxreconsumetimes(0)

作者：Cpp五条 | 2024-05-21 18:20:09

赞

踩

.setmaxreconsumetimes(0)

        上一篇博客RocketMQ原理学习-- Name Server中我们介绍了Name Server提供的相关功能，这篇博客我们来介绍一下生产者消息发送相关的内容。

消息发送示例：

  1. public class Producer {
  2.  
  3.     public static void main(String[] args) throws MQClientException, InterruptedException {
  4.         DefaultMQProducer producer = new DefaultMQProducer("rmq-group");
  5.         producer.setNamesrvAddr("localhost:9876");
  6.         producer.start();
  7.         try {
  8.             for (int i = 0; i < 3; i++) {
  9.                 Message msg = new Message("TopicA-test",// topic
  10.                         "TagA",// tag
  11.                         (new Date() + "Hello RocketMQ ,QuickStart 11" + i)
  12.                                 .getBytes()// body
  13.                 );
  14.                 SendResult sendResult = producer.send(msg);
  15.                 System.out.println(sendResult);
  16.             }
  17.         } catch (Exception e) {
  18.             e.printStackTrace();
  19.         }
  20.        producer.shutdown();
  21.     }
  22.  
  23. }

在DefaultMQProducer中调用send方法发送消息

  1. public SendResult send(Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  2.         return this.defaultMQProducerImpl.send(msg);
  3.     }

最终是调用DefaultMQProducerImpl的send方法来发送消息

  1. public SendResult send(Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  2.         return this.send(msg, (long)this.defaultMQProducer.getSendMsgTimeout());
  3.     }

最终会调用sendDefaultImpl方法来发送消息：

（1）在发送消息时会根据消息的Topic名称从Name Server中根据Topic名称来获取Broker相关的地址信息TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());

（2）由于一个Broker中对于每个Topic会有多个MessageQueue，生产者默认选择MessageQueue策略为轮询MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, info);

（3）选择了MessageQueue后，会根据MessageQueue中Broker信息调用sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout);发送消息。

  1.  private SendResult sendDefaultImpl(Message msg, CommunicationMode communicationMode, SendCallback sendCallback, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  2.         this.makeSureStateOK();
  3.         Validators.checkMessage(msg, this.defaultMQProducer);
  4.         long invokeID = this.random.nextLong();
  5.         long beginTimestampFirst = System.currentTimeMillis();
  6.         long beginTimestampPrev = beginTimestampFirst;
  7.         TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
  8.         if (topicPublishInfo != null && topicPublishInfo.ok()) {
  9.             MessageQueue mq = null;
  10.             Exception exception = null;
  11.             SendResult sendResult = null;
  12.             int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
  13.             int times = 0;
  14.             String[] brokersSent = new String[timesTotal];
  15.  
  16.             while(true) {
  17.                 String info;
  18.                 if (times < timesTotal) {
  19.                     info = null == mq ? null : mq.getBrokerName();
  20.                     MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, info);
  21.                     if (mqSelected != null) {
  22.                         mq = mqSelected;
  23.                         brokersSent[times] = mqSelected.getBrokerName();
  24.  
  25.                         long endTimestamp;
  26.                         try {
  27.                             beginTimestampPrev = System.currentTimeMillis();
  28.                             sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout);
  29.                             endTimestamp = System.currentTimeMillis();
  30.                             this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
  31.                             switch(communicationMode) {
  32.                             case ASYNC:
  33.                                 return null;
  34.                             case ONEWAY:
  35.                                 return null;
  36.                             case SYNC:
  37.                                 if (sendResult.getSendStatus() == SendStatus.SEND_OK || !this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
  38.                                     return sendResult;
  39.                                 }
  40.                             }
  41.                         } catch (RemotingException var24) {
  42.                             endTimestamp = System.currentTimeMillis();
  43.                             this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, true);
  44.                             this.log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var24);
  45.                             this.log.warn(msg.toString());
  46.                             exception = var24;
  47.                         } catch (MQClientException var25) {
  48.                             endTimestamp = System.currentTimeMillis();
  49.                             this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, true);
  50.                             this.log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var25);
  51.                             this.log.warn(msg.toString());
  52.                             exception = var25;
  53.                         } catch (MQBrokerException var26) {
  54.                             endTimestamp = System.currentTimeMillis();
  55.                             this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, true);
  56.                             this.log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var26);
  57.                             this.log.warn(msg.toString());
  58.                             exception = var26;
  59.                             switch(var26.getResponseCode()) {
  60.                             case 1:
  61.                             case 14:
  62.                             case 16:
  63.                             case 17:
  64.                             case 204:
  65.                             case 205:
  66.                                 break;
  67.                             default:
  68.                                 if (sendResult != null) {
  69.                                     return sendResult;
  70.                                 }
  71.  
  72.                                 throw var26;
  73.                             }
  74.                         } catch (InterruptedException var27) {
  75.                             endTimestamp = System.currentTimeMillis();
  76.                             this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, false);
  77.                             this.log.warn(String.format("sendKernelImpl exception, throw exception, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var27);
  78.                             this.log.warn(msg.toString());
  79.                             this.log.warn("sendKernelImpl exception", var27);
  80.                             this.log.warn(msg.toString());
  81.                             throw var27;
  82.                         }
  83.  
  84.                         ++times;
  85.                         continue;
  86.                     }
  87.                 }
  88.  
  89.                 if (sendResult != null) {
  90.                     return sendResult;
  91.                 }
  92.  
  93.                 info = String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s", times, System.currentTimeMillis() - beginTimestampFirst, msg.getTopic(), Arrays.toString(brokersSent));
  94.                 info = info + FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/");
  95.                 MQClientException mqClientException = new MQClientException(info, (Throwable)exception);
  96.                 if (exception instanceof MQBrokerException) {
  97.                     mqClientException.setResponseCode(((MQBrokerException)exception).getResponseCode());
  98.                 } else if (exception instanceof RemotingConnectException) {
  99.                     mqClientException.setResponseCode(10001);
  100.                 } else if (exception instanceof RemotingTimeoutException) {
  101.                     mqClientException.setResponseCode(10002);
  102.                 } else if (exception instanceof MQClientException) {
  103.                     mqClientException.setResponseCode(10003);
  104.                 }
  105.  
  106.                 throw mqClientException;
  107.             }
  108.         } else {
  109.             List<String> nsList = this.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList();
  110.             if (null != nsList && !nsList.isEmpty()) {
  111.                 throw (new MQClientException("No route info of this topic, " + msg.getTopic() + FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/"), (Throwable)null)).setResponseCode(10005);
  112.             } else {
  113.                 throw (new MQClientException("No name server address, please set it." + FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/"), (Throwable)null)).setResponseCode(10004);
  114.             }
  115.         }
  116.     }

在sendKernelImpl中会获取Broker的地址信息，然后组装消息信息，调用MQClientAPIImpl的sendMessage方法将消息发送到Broker中。

  1. private SendResult sendKernelImpl(Message msg, MessageQueue mq, CommunicationMode communicationMode, SendCallback sendCallback, TopicPublishInfo topicPublishInfo, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  2.         //获取Broker的地址信息
  3.         String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
  4.         if (null == brokerAddr) {
  5.             this.tryToFindTopicPublishInfo(mq.getTopic());
  6.             brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
  7.         }
  8.  
  9.         SendMessageContext context = null;
  10.         if (brokerAddr != null) {
  11.             brokerAddr = MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
  12.             byte[] prevBody = msg.getBody();
  13.  
  14.             SendResult var28;
  15.             try {
  16.                 if (!(msg instanceof MessageBatch)) {
  17.                     MessageClientIDSetter.setUniqID(msg);
  18.                 }
  19.  
  20.                 int sysFlag = 0;
  21.                 if (this.tryToCompressMessage(msg)) {
  22.                     sysFlag |= 1;
  23.                 }
  24.  
  25.                 String tranMsg = msg.getProperty("TRAN_MSG");
  26.                 if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
  27.                     sysFlag |= 4;
  28.                 }
  29.  
  30.                 if (this.hasCheckForbiddenHook()) {
  31.                     CheckForbiddenContext checkForbiddenContext = new CheckForbiddenContext();
  32.                     checkForbiddenContext.setNameSrvAddr(this.defaultMQProducer.getNamesrvAddr());
  33.                     checkForbiddenContext.setGroup(this.defaultMQProducer.getProducerGroup());
  34.                     checkForbiddenContext.setCommunicationMode(communicationMode);
  35.                     checkForbiddenContext.setBrokerAddr(brokerAddr);
  36.                     checkForbiddenContext.setMessage(msg);
  37.                     checkForbiddenContext.setMq(mq);
  38.                     checkForbiddenContext.setUnitMode(this.isUnitMode());
  39.                     this.executeCheckForbiddenHook(checkForbiddenContext);
  40.                 }
  41.  
  42.                 if (this.hasSendMessageHook()) {
  43.                     context = new SendMessageContext();
  44.                     context.setProducer(this);
  45.                     context.setProducerGroup(this.defaultMQProducer.getProducerGroup());
  46.                     context.setCommunicationMode(communicationMode);
  47.                     context.setBornHost(this.defaultMQProducer.getClientIP());
  48.                     context.setBrokerAddr(brokerAddr);
  49.                     context.setMessage(msg);
  50.                     context.setMq(mq);
  51.                     String isTrans = msg.getProperty("TRAN_MSG");
  52.                     if (isTrans != null && isTrans.equals("true")) {
  53.                         context.setMsgType(MessageType.Trans_Msg_Half);
  54.                     }
  55.  
  56.                     if (msg.getProperty("__STARTDELIVERTIME") != null || msg.getProperty("DELAY") != null) {
  57.                         context.setMsgType(MessageType.Delay_Msg);
  58.                     }
  59.  
  60.                     this.executeSendMessageHookBefore(context);
  61.                 }
  62.                 //组装消息请求头信息
  63.                 SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
  64.                 requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
  65.                 requestHeader.setTopic(msg.getTopic());
  66.                 requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
  67.                 requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
  68.                 requestHeader.setQueueId(mq.getQueueId());
  69.                 requestHeader.setSysFlag(sysFlag);
  70.                 requestHeader.setBornTimestamp(System.currentTimeMillis());
  71.                 requestHeader.setFlag(msg.getFlag());
  72.                 requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
  73.                 requestHeader.setReconsumeTimes(0);
  74.                 requestHeader.setUnitMode(this.isUnitMode());
  75.                 requestHeader.setBatch(msg instanceof MessageBatch);
  76.                 if (requestHeader.getTopic().startsWith("%RETRY%")) {
  77.                     String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
  78.                     if (reconsumeTimes != null) {
  79.                         requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
  80.                         MessageAccessor.clearProperty(msg, "RECONSUME_TIME");
  81.                     }
  82.  
  83.                     String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(msg);
  84.                     if (maxReconsumeTimes != null) {
  85.                         requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
  86.                         MessageAccessor.clearProperty(msg, "MAX_RECONSUME_TIMES");
  87.                     }
  88.                 }
  89.  
  90.                 SendResult sendResult = null;
  91.                 switch(communicationMode) {
  92.                 //异步消息
  93.                 case ASYNC:
  94.                     sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(brokerAddr, mq.getBrokerName(), msg, requestHeader, timeout, communicationMode, sendCallback, topicPublishInfo, this.mQClientFactory, this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(), context, this);
  95.                     break;
  96.                 case ONEWAY:
  97.                 //同步消息
  98.                 case SYNC:
  99.                     sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(brokerAddr, mq.getBrokerName(), msg, requestHeader, timeout, communicationMode, context, this);
  100.                     break;
  101.                 default:
  102.                     assert false;
  103.                 }
  104.  
  105.                 if (this.hasSendMessageHook()) {
  106.                     context.setSendResult(sendResult);
  107.                     this.executeSendMessageHookAfter(context);
  108.                 }
  109.  
  110.                 var28 = sendResult;
  111.             } catch (RemotingException var21) {
  112.                 if (this.hasSendMessageHook()) {
  113.                     context.setException(var21);
  114.                     this.executeSendMessageHookAfter(context);
  115.                 }
  116.  
  117.                 throw var21;
  118.             } catch (MQBrokerException var22) {
  119.                 if (this.hasSendMessageHook()) {
  120.                     context.setException(var22);
  121.                     this.executeSendMessageHookAfter(context);
  122.                 }
  123.  
  124.                 throw var22;
  125.             } catch (InterruptedException var23) {
  126.                 if (this.hasSendMessageHook()) {
  127.                     context.setException(var23);
  128.                     this.executeSendMessageHookAfter(context);
  129.                 }
  130.  
  131.                 throw var23;
  132.             } finally {
  133.                 msg.setBody(prevBody);
  134.             }
  135.  
  136.             return var28;
  137.         } else {
  138.             throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", (Throwable)null);
  139.         }
  140.     }

在MQClientAPIImpl中会调用底层通信模块NettyRemotingClient的invokeSync方法将消息信息发送到Broker中。

  1. public SendResult sendMessage(String addr, String brokerName, Message msg, SendMessageRequestHeader requestHeader, long timeoutMillis, CommunicationMode communicationMode, SendCallback sendCallback, TopicPublishInfo topicPublishInfo, MQClientInstance instance, int retryTimesWhenSendFailed, SendMessageContext context, DefaultMQProducerImpl producer) throws RemotingException, MQBrokerException, InterruptedException {
  2.         RemotingCommand request = null;
  3. 		//创建消息
  4.         if (!sendSmartMsg && !(msg instanceof MessageBatch)) {
  5.             request = RemotingCommand.createRequestCommand(10, requestHeader);
  6.         } else {
  7.             SendMessageRequestHeaderV2 requestHeaderV2 = SendMessageRequestHeaderV2.createSendMessageRequestHeaderV2(requestHeader);
  8.             request = RemotingCommand.createRequestCommand(msg instanceof MessageBatch ? 320 : 310, requestHeaderV2);
  9.         }
  10. 		//添加消息请求体
  11.         request.setBody(msg.getBody());
  12.         switch(communicationMode) {
  13.         case ONEWAY:
  14.             this.remotingClient.invokeOneway(addr, request, timeoutMillis);
  15.             return null;
  16.         case ASYNC:
  17.             AtomicInteger times = new AtomicInteger();
  18.             this.sendMessageAsync(addr, brokerName, msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, retryTimesWhenSendFailed, times, context, producer);
  19.             return null;
  20.         case SYNC:
  21.             return this.sendMessageSync(addr, brokerName, msg, timeoutMillis, request);
  22.         default:
  23.             assert false;
  24.  
  25.             return null;
  26.         }
  27. }

在NettyRemotingClient中调用invokeSync方法根据Broker地址信息创建Channel，然后根据request请求信息发起消息，并获取返回值RemotingCommand。

  1. public RemotingCommand invokeSync(String addr, RemotingCommand request, long timeoutMillis) throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException {
  2.         Channel channel = this.getAndCreateChannel(addr);
  3.         if (channel != null && channel.isActive()) {
  4.             try {
  5.                 if (this.rpcHook != null) {
  6.                     this.rpcHook.doBeforeRequest(addr, request);
  7.                 }
  8.  
  9.                 RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis);
  10.                 if (this.rpcHook != null) {
  11.                     this.rpcHook.doAfterResponse(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
  12.                 }
  13.  
  14.                 return response;
  15.             } catch (RemotingSendRequestException var7) {
  16.                 log.warn("invokeSync: send request exception, so close the channel[{}]", addr);
  17.                 this.closeChannel(addr, channel);
  18.                 throw var7;
  19.             } catch (RemotingTimeoutException var8) {
  20.                 if (this.nettyClientConfig.isClientCloseSocketIfTimeout()) {
  21.                     this.closeChannel(addr, channel);
  22.                     log.warn("invokeSync: close socket because of timeout, {}ms, {}", timeoutMillis, addr);
  23.                 }
  24.  
  25.                 log.warn("invokeSync: wait response timeout exception, the channel[{}]", addr);
  26.                 throw var8;
  27.             }
  28.         } else {
  29.             this.closeChannel(addr, channel);
  30.             throw new RemotingConnectException(addr);
  31.         }
  32.     }

总结：

（1）生产者根据Topic名称从Name Server中获取相关Broker信息，根据Broker的地址信息将消息发送到对应Broker地址中

（2）对于同一个Topic，一个Broker对应多个MessageQueue(Topic可以在多个Broker中)，生产者默认通过取模轮询方式将消息发送到对应的MessageQueue中。

（3）生产者提供了接口send(Message msg, MessageQueueSelector selector, Object arg)，我们可以通过实现MessageQueueSelector接口可以实现选取MessageQueue的方法，例如实现有序消息就可以将需要有序的消息发送到同一个MessageQueue即可。

本文内容由网友自发贡献，转载请注明出处：https://www.wpsshop.cn/w/Cpp五条/article/detail/604010
推荐阅读
相关标签

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

  

闽ICP备14008679号