同步消息发送

1）：同步消息发送代码执行

首先利用DefaultMQProducer实现类进行发送消息。

  1. //检查消息
  2. Validators.checkMessage(msg, this);
  3. //设置Topic
  4. msg.setTopic(withNamespace(msg.getTopic()));
  5. //调用MQProducer实现并返回结果
  6. return this.defaultMQProducerImpl.send(msg);

DefaultMQProducerImpl#send

  1. //加上发送默认超时时间
  2. send(msg, this.defaultMQProducer.getSendMsgTimeout());
  4. //默认为同步
  5. this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);

DefaultMQProducerImpl#sendDefaultImpl

  1. //确保当前状态为运行中，否则抛异常
  2. this.makeSureStateOK();
  3. //根据消息跟producer进行消息检查
  4. Validators.checkMessage(msg, this.defaultMQProducer);
  5. //生成随机invokeID
  6. final long invokeID = random.nextLong();
  7. //记录第一次开始时间
  8. long beginTimestampFirst = System.currentTimeMillis();
  9. //记录前一次开始时间
  10. long beginTimestampPrev = beginTimestampFirst;
  11. //结束时间
  12. long endTimestamp = beginTimestampFirst;
  13. //获取Topic订阅信息(主要为了确定将消息发到哪一个Broker上)
  14. TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
  15. if (topicPublishInfo != null && topicPublishInfo.ok()) {
  16.     boolean callTimeout = false;
  17.     MessageQueue mq = null;
  18.     Exception exception = null;
  19.     SendResult sendResult = null;
  20.     //发送次数总数,若为同步次数为1+2=3,否则为1
  21.     int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
  22.     //当前次数
  23.     int times = 0;
  24.     String[] brokersSent = new String[timesTotal];
  25.     //循环发送消息(用于重试)
  26.     for (; times < timesTotal; times++) {
  27.         //获取broker服务名称
  28.         String lastBrokerName = null == mq ? null : mq.getBrokerName();
  29.         //根据Topic信息+broker名称选择一个消息队列
  30.         MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
  31.         if (mqSelected != null) {
  32.             mq = mqSelected;
  33.             brokersSent[times] = mq.getBrokerName();
  34.             try {
  35.                 beginTimestampPrev = System.currentTimeMillis();
  36.                 //若times大于0，说明需要重试
  37.                 if (times > 0) {
  38.                     //重新发送时，重设Topic名称
  39.                     msg.setTopic(this.defaultMQProducer.withNamespace(msg.getTopic()));
  40.                 }
  41.                 //计算耗时
  42.                 long costTime = beginTimestampPrev - beginTimestampFirst;
  43.                 //若耗时已经超过发送时的超时时间，则退出
  44.                 if (timeout < costTime) {
  45.                     callTimeout = true;
  46.                     break;
  47.                 }
  48.                 //执行消息发送
  49.                 sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
  50.                 //计算结束时间
  51.                 endTimestamp = System.currentTimeMillis();
  52.                 //更新当前Broker状态。表示当前broker是好用的
  53.                 this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
  54.                 switch (communicationMode) {
  55.                     case ASYNC:
  56.                         return null;
  57.                     case ONEWAY:
  58.                         return null;
  59.                     case SYNC:
  60.                         //同步,判断发送不成功的情况下是否需要重试
  61.                         if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
  62.                             if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
  63.                                 continue;
  64.                             }
  65.                         }
  66.                         //返回结果
  67.                         return sendResult;
  68.                     default:
  69.                         break;
  70.                 }
  71.             } catch (RemotingException e) {
  72.                 //计算结束时间
  73.                 endTimestamp = System.currentTimeMillis();
  74.                 //更新当前Broker。表示当前broker是不好用的
  75.                 this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
  76.                 log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
  77.                 log.warn(msg.toString());
  78.                 exception = e;
  79.                 continue;
  80.             } catch (MQClientException e) {
  81.                 //同上
  82.                 endTimestamp = System.currentTimeMillis();
  83.                 this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
  84.                 log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
  85.                 log.warn(msg.toString());
  86.                 exception = e;
  87.                 continue;
  88.             } catch (MQBrokerException e) {
  89.                 //同上
  90.                 endTimestamp = System.currentTimeMillis();
  91.                 this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
  92.                 log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
  93.                 log.warn(msg.toString());
  94.                 exception = e;
  95.                 if (this.defaultMQProducer.getRetryResponseCodes().contains(e.getResponseCode())) {
  96.                     continue;
  97.                 } else {
  98.                     if (sendResult != null) {
  99.                         return sendResult;
  100.                     }
  102.                     throw e;
  103.                 }
  104.             } catch (InterruptedException e) {
  105.                 //同上
  106.                 endTimestamp = System.currentTimeMillis();
  107.                 this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
  108.                 log.warn(String.format("sendKernelImpl exception, throw exception, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
  109.                 log.warn(msg.toString());
  111.                 log.warn("sendKernelImpl exception", e);
  112.                 log.warn(msg.toString());
  113.                 throw e;
  114.             }
  115.         } else {
  116.             break;
  117.         }
  118.     }
  119.     //返回结果不为空，则返回
  120.     if (sendResult != null) {
  121.         return sendResult;
  122.     }
  124.     //报错
  125.     String info = String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s",
  126.                                 times,
  127.                                 System.currentTimeMillis() - beginTimestampFirst,
  128.                                 msg.getTopic(),
  129.                                 Arrays.toString(brokersSent));
  131.     info += FAQUrl.suggestTodo(FAQUrl.SEND_MSG_FAILED);
  133.     MQClientException mqClientException = new MQClientException(info, exception);
  134.     if (callTimeout) {
  135.         throw new RemotingTooMuchRequestException("sendDefaultImpl call timeout");
  136.     }
  138.     if (exception instanceof MQBrokerException) {
  139.         mqClientException.setResponseCode(((MQBrokerException) exception).getResponseCode());
  140.     } else if (exception instanceof RemotingConnectException) {
  141.         mqClientException.setResponseCode(ClientErrorCode.CONNECT_BROKER_EXCEPTION);
  142.     } else if (exception instanceof RemotingTimeoutException) {
  143.         mqClientException.setResponseCode(ClientErrorCode.ACCESS_BROKER_TIMEOUT);
  144.     } else if (exception instanceof MQClientException) {
  145.         mqClientException.setResponseCode(ClientErrorCode.BROKER_NOT_EXIST_EXCEPTION);
  146.     }
  148.     throw mqClientException;
  149. }
  150. //检查NameServer地址设置，地址为空就抛异常
  151. validateNameServerSetting();
  152. //抛异常
  153. throw new MQClientException("No route info of this topic: " + msg.getTopic() + FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO),
  154.                             null).setResponseCode(ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION);

DefaultMQProducerImpl#sendKernelImpl

  1. //计算开始时间
  2. long beginStartTime = System.currentTimeMillis();
  3. //根据Broker名称获取Broker地址
  4. String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
  5. //若Broker地址为空
  6. if (null == brokerAddr) {
  7.     //那就根据Topic重新检索一下Broker
  8.     tryToFindTopicPublishInfo(mq.getTopic());
  9.     brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
  10. }
  12. SendMessageContext context = null;
  13. if (brokerAddr != null) {
  14.     //Vip通道Broker
  15.     brokerAddr = MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
  17.     byte[] prevBody = msg.getBody();
  18.     try {
  19.         //for MessageBatch,ID has been set in the generating process
  20.         if (!(msg instanceof MessageBatch)) {
  21.             //设置唯一ID
  22.             MessageClientIDSetter.setUniqID(msg);
  23.         }
  25.         boolean topicWithNamespace = false;
  26.         //设置nameSpace
  27.         if (null != this.mQClientFactory.getClientConfig().getNamespace()) {
  28.             msg.setInstanceId(this.mQClientFactory.getClientConfig().getNamespace());
  29.             topicWithNamespace = true;
  30.         }
  32.         int sysFlag = 0;
  33.         boolean msgBodyCompressed = false;
  34.         //压缩msg
  35.         if (this.tryToCompressMessage(msg)) {
  36.             sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
  37.             msgBodyCompressed = true;
  38.         }
  39.         //系统标识
  40.         final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
  41.         if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
  42.             sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
  43.         }
  44.         ../部分代码省略
  45.         //组装消息头
  46.         SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
  47.         requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
  48.         requestHeader.setTopic(msg.getTopic());
  49.         requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
  50.         requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
  51.         requestHeader.setQueueId(mq.getQueueId());
  52.         requestHeader.setSysFlag(sysFlag);
  53.         requestHeader.setBornTimestamp(System.currentTimeMillis());
  54.         requestHeader.setFlag(msg.getFlag());
  55.         requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
  56.         requestHeader.setReconsumeTimes(0);
  57.         requestHeader.setUnitMode(this.isUnitMode());
  58.         //批量标识
  59.         requestHeader.setBatch(msg instanceof MessageBatch);
  60.         //判断Topic是否重试
  61.         if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
  62.             String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
  63.             if (reconsumeTimes != null) {
  64.                 requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
  65.                 MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
  66.             }
  68.             String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(msg);
  69.             if (maxReconsumeTimes != null) {
  70.                 requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
  71.                 MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_MAX_RECONSUME_TIMES);
  72.             }
  73.         }
  75.         SendResult sendResult = null;
  76.         switch (communicationMode) {
  77.             //异步
  78.             case ASYNC:
  79.                 ../部分代码省略
  80.                 //计算耗时
  81.                 long costTimeAsync = System.currentTimeMillis() - beginStartTime;
  82.                 if (timeout < costTimeAsync) {
  83.                     throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
  84.                 }
  85.                 //发送消息
  86.                 sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
  87.                     brokerAddr,
  88.                     mq.getBrokerName(),
  89.                     tmpMessage,
  90.                     requestHeader,
  91.                     timeout - costTimeAsync,
  92.                     communicationMode,
  93.                     sendCallback,
  94.                     topicPublishInfo,
  95.                     this.mQClientFactory,
  96.                     this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(),
  97.                     context,
  98.                     this);
  99.                 break;
  100.             case ONEWAY:
  101.             case SYNC:
  102.                 //计算耗时
  103.                 long costTimeSync = System.currentTimeMillis() - beginStartTime;
  104.                 if (timeout < costTimeSync) {
  105.                     throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
  106.                 }
  107.                 //发送消息
  108.                 sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
  109.                     brokerAddr,
  110.                     mq.getBrokerName(),
  111.                     msg,
  112.                     requestHeader,
  113.                     timeout - costTimeSync,
  114.                     communicationMode,
  115.                     context,
  116.                     this);
  117.                 break;
  118.             default:
  119.                 assert false;
  120.                 break;
  121.         }
  122.         ../部分代码省略

MQClientAPIImpl#sendMessage

  1. ../部分代码省略
  3. switch (communicationMode) {
  4.     case ONEWAY:
  5.         //单向发送
  6.         this.remotingClient.invokeOneway(addr, request, timeoutMillis);
  7.         return null;
  8.     case ASYNC:
  9.         final AtomicInteger times = new AtomicInteger();
  10.         //计算耗时
  11.         long costTimeAsync = System.currentTimeMillis() - beginStartTime;
  12.         if (timeoutMillis < costTimeAsync) {
  13.             throw new RemotingTooMuchRequestException("sendMessage call timeout");
  14.         }
  15.         //异步发送
  16.         this.sendMessageAsync(addr, brokerName, msg, timeoutMillis - costTimeAsync, request, sendCallback, topicPublishInfo, instance,
  17.                               retryTimesWhenSendFailed, times, context, producer);
  18.         return null;
  19.     case SYNC:
  20.         //计算耗时
  21.         long costTimeSync = System.currentTimeMillis() - beginStartTime;
  22.         if (timeoutMillis < costTimeSync) {
  23.             throw new RemotingTooMuchRequestException("sendMessage call timeout");
  24.         }
  25.         //同步发送
  26.         return this.sendMessageSync(addr, brokerName, msg, timeoutMillis - costTimeSync, request);
  27.     default:
  28.         assert false;
  29.         break;
  30. }
  32. return null;

MQClientAPIImpl#sendMessageSync

  1. //远程调用同步请求
  2. RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
  3. assert response != null;
  4. //返回结果
  5. return this.processSendResponse(brokerName, msg, response,addr);

MQClientAPIImpl#invokeSync

  1. //计算开始时间
  2. long beginStartTime = System.currentTimeMillis();
  3. //根据Broker的地址创建通道
  4. final Channel channel = this.getAndCreateChannel(addr);
  5. //判断通道不为空且存活
  6. if (channel != null && channel.isActive()) {
  7.     try {
  8.         //执行before钩子
  9.         doBeforeRpcHooks(addr, request);
  10.         //计算耗时
  11.         long costTime = System.currentTimeMillis() - beginStartTime;
  12.         if (timeoutMillis < costTime) {
  13.             throw new RemotingTimeoutException("invokeSync call timeout");
  14.         }
  15.         //调用同步实现，返回结果
  16.         RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
  17.         //执行after钩子
  18.         doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
  19.         return response;
  20.     } catch (RemotingSendRequestException e) {
  21.         log.warn("invokeSync: send request exception, so close the channel[{}]", addr);
  22.         this.closeChannel(addr, channel);
  23.         throw e;
  24.     } catch (RemotingTimeoutException e) {
  25.         //超时关闭通道
  26.         if (nettyClientConfig.isClientCloseSocketIfTimeout()) {
  27.             this.closeChannel(addr, channel);
  28.             log.warn("invokeSync: close socket because of timeout, {}ms, {}", timeoutMillis, addr);
  29.         }
  30.         log.warn("invokeSync: wait response timeout exception, the channel[{}]", addr);
  31.         throw e;
  32.     }
  33. } else {
  34.     //关闭通道，抛异常
  35.     this.closeChannel(addr, channel);
  36.     throw new RemotingConnectException(addr);
  37. }

MQClientAPIImpl#invokeSyncImpl

  1. //请求ID，producer唯一
  2. final int opaque = request.getOpaque();
  3. try {
  4.     //根据请求ID，创建返回future
  5.     final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis, null, null);
  6.     //放入返回集合
  7.     this.responseTable.put(opaque, responseFuture);
  8.     //获取Socket服务器地址,也就是Broker地址
  9.     final SocketAddress addr = channel.remoteAddress();
  10.     //写入请求消息，并刷新
  11.     channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
  12.         @Override
  13.         public void operationComplete(ChannelFuture f) throws Exception {
  14.             if (f.isSuccess()) {
  15.                 responseFuture.setSendRequestOK(true);
  16.                 return;
  17.             } else {
  18.                 responseFuture.setSendRequestOK(false);
  19.             }
  21.             responseTable.remove(opaque);
  22.             responseFuture.setCause(f.cause());
  23.             responseFuture.putResponse(null);
  24.             log.warn("send a request command to channel <" + addr + "> failed.");
  25.         }
  26.     });
  27.     //这里利用的是Netty的机制，通过CountDownLatch来实现
  28.     RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
  29.     if (null == responseCommand) {
  30.         if (responseFuture.isSendRequestOK()) {
  31.             throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
  32.                                                responseFuture.getCause());
  33.         } else {
  34.             throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
  35.         }
  36.     }
  38.     return responseCommand;
  39. } finally {
  40.     //删除该结果，避免集合越来越庞大
  41.     this.responseTable.remove(opaque);
  42. }

2）：同步消息发送逻辑总结

  1. 消息长度验证
  2. 查找主题路由信息，若没有从NameServer中更新
  3. 设置重试次数，并选择Broker的消息队列。若出现异常情况，切换Broker进行重试
  4. 发送消息（检查钩子，压缩消息，组装头部信息）
    1. 判断发送方式，调用Netty的Socket通道进行发送消息请求
    2. 获取结果指令，返回