RocketMQ 消息发送&存储

发表于 | 分类于

5. 消息发送并存储

发送分为同步和异步的方式，本章主要解析同步方式，异步方式与同步方式很相似。
发送的过程，Producer发送消息，Broker接收消息并存储。

发送

发送过程，最主要逻辑在DefaultMQProducerImpl.sendDefaultImpl中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
private SendResult sendDefaultImpl(Message msg, final CommunicationMode communicationMode, final SendCallback sendCallback, final long timeout) throws ... {
   final long invokeID = random.nextLong();
   long beginTimestampFirst = System.currentTimeMillis();
   long beginTimestampPrev = beginTimestampFirst;
   @SuppressWarnings("UnusedAssignment")
   long endTimestamp = beginTimestampFirst;
   // 获取topic路由信息
   TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
   if (topicPublishInfo != null && topicPublishInfo.ok()) {
       MessageQueue mq = null;
       SendResult sendResult = null;
       int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
       int times = 0;
       String[] brokersSent = new String[timesTotal];
       // 最多重试timeTotal次
       for (; times < timesTotal; times++) {
           String lastBrokerName = null == mq ? null : mq.getBrokerName();
           // 按一定策略 选择一个队列进行发送消息（目前是轮询法）
           @SuppressWarnings("SpellCheckingInspection")
           MessageQueue tmpmq = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
           if (tmpmq != null) {
               mq = tmpmq;
               brokersSent[times] = mq.getBrokerName();
               try {
                   beginTimestampPrev = System.currentTimeMillis();
                   // 调用发送消息核心方法
                   sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout);
                   endTimestamp = System.currentTimeMillis();
                   // 更新Broker可用性信息
                   this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
                   switch (communicationMode) {
                       case ASYNC:
                           return null;
                       case ONEWAY:
                           return null;
                       case SYNC:
                           //重试
                           if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                               if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
                                   continue;
                               }
                           }
                           return sendResult;
                       default:
                           break;
                   }
               } catch //...
           } else {
               break;
           }
       }
       if (sendResult != null) {
           return sendResult;
       }
       //...
   }
   //...
}

可以看出整个过程：首先获取路由信息，然后按一定策略选择一个队列进行发送消息，当失败时会重试，最多重试timesTotal次。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
private SendResult sendKernelImpl(final Message msg, final MessageQueue mq, final CommunicationMode communicationMode, final SendCallback sendCallback, final TopicPublishInfo topicPublishInfo, final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
   // 跟据队列对应broker地址
   String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
   if (null == brokerAddr) {
       tryToFindTopicPublishInfo(mq.getTopic());
       brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
   }
   SendMessageContext context = null;
   if (brokerAddr != null) {
       // 是否使用VIP通道，否认使用，之前在介绍broker的时候介绍过broker会起两个server，端口号-2 的server为fast server
       brokerAddr = MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
       try {
           // 生成msg唯一ID 时间差 + counter
           MessageClientIDSetter.setUniqID(msg);
           int sysFlag = 0;
           // 消息压缩
           if (this.tryToCompressMessage(msg)) {
               sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
           }
           // 事务
           final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
           if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
               sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
           }
           // ...发送前校验逻辑
           // 构建发送消息请求
           SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
           requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
           requestHeader.setTopic(msg.getTopic());
           requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
           requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
           requestHeader.setQueueId(mq.getQueueId());
           requestHeader.setSysFlag(sysFlag);
           requestHeader.setBornTimestamp(System.currentTimeMillis());
           requestHeader.setFlag(msg.getFlag());
           requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
           requestHeader.setReconsumeTimes(0);
           requestHeader.setUnitMode(this.isUnitMode());
           // 消息重发
           if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
               String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
               if (reconsumeTimes != null) {
                   requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
                   MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
               }
               String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(msg);
               if (maxReconsumeTimes != null) {
                   requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
                   MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_MAX_RECONSUME_TIMES);
               }
           }
           // 发送消息
           SendResult sendResult = null;
           switch (communicationMode) {
               case SYNC:
                    // 调用MQclient进行发送消息
                   sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
                       brokerAddr,
                       mq.getBrokerName(),
                       msg,
                       requestHeader,
                       timeout,
                       communicationMode,
                       context,
                       this);
                   break;
               case //...
           }
           // 返回发送结果
           return sendResult;
       } catch //...
   }
}

接收

Broker的fast Server接收到消息后，交给SendMessageProcessor进行处理。以下为其主要核心逻辑：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
private RemotingCommand sendMessage(final ChannelHandlerContext ctx, final RemotingCommand request, final SendMessageContext sendMessageContext, final SendMessageRequestHeader requestHeader) throws RemotingCommandException {
   // 初始化响应
   final RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
   final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader) response.readCustomHeader();
   response.setOpaque(request.getOpaque());
   response.addExtField(MessageConst.PROPERTY_MSG_REGION, this.brokerController.getBrokerConfig().getRegionId());
   response.addExtField(MessageConst.PROPERTY_TRACE_SWITCH, String.valueOf(this.brokerController.getBrokerConfig().isTraceOn()));
   final byte[] body = request.getBody();
   //.... 校验
   // 如果队列小于0，从可用队列随机选择
   int queueIdInt = requestHeader.getQueueId();
   TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
   if (queueIdInt < 0) {
       queueIdInt = Math.abs(this.random.nextInt() % 99999999) % topicConfig.getWriteQueueNums();
   }
   int sysFlag = requestHeader.getSysFlag();
   if (TopicFilterType.MULTI_TAG == topicConfig.getTopicFilterType()) {
       sysFlag |= MessageSysFlag.MULTI_TAGS_FLAG;
   }
   String newTopic = requestHeader.getTopic();
   //重试消息
   if (null != newTopic && newTopic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
       String groupName = newTopic.substring(MixAll.RETRY_GROUP_TOPIC_PREFIX.length());
       SubscriptionGroupConfig subscriptionGroupConfig =
           this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(groupName);
       if (null == subscriptionGroupConfig) {
           response.setCode(ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST);
           response.setRemark("subscription group not exist, " + groupName + " " + FAQUrl.suggestTodo(FAQUrl.SUBSCRIPTION_GROUP_NOT_EXIST));
           return response;
       }
       int maxReconsumeTimes = subscriptionGroupConfig.getRetryMaxTimes();
       if (request.getVersion() >= MQVersion.Version.V3_4_9.ordinal()) {
           maxReconsumeTimes = requestHeader.getMaxReconsumeTimes();
       }
       //超过消费最大次数则放入死信队列中
       int reconsumeTimes = requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes();
       if (reconsumeTimes >= maxReconsumeTimes) {
           newTopic = MixAll.getDLQTopic(groupName);
           queueIdInt = Math.abs(this.random.nextInt() % 99999999) % DLQ_NUMS_PER_GROUP;
           topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(newTopic, DLQ_NUMS_PER_GROUP, PermName.PERM_WRITE, 0);
       }
   }
   // 创建MessageExtBrokerInner
   MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
   msgInner.setTopic(newTopic);
   msgInner.setBody(body);
   msgInner.setFlag(requestHeader.getFlag());
   MessageAccessor.setProperties(msgInner, MessageDecoder.string2messageProperties(requestHeader.getProperties()));
   msgInner.setPropertiesString(requestHeader.getProperties());
   msgInner.setTagsCode(MessageExtBrokerInner.tagsString2tagsCode(topicConfig.getTopicFilterType(), msgInner.getTags()));
   msgInner.setQueueId(queueIdInt);
   msgInner.setSysFlag(sysFlag);
   msgInner.setBornTimestamp(requestHeader.getBornTimestamp());
   msgInner.setBornHost(ctx.channel().remoteAddress());
   msgInner.setStoreHost(this.getStoreHost());
   msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());

   // 校验是否不允许发送事务消息
   if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {
       String traFlag = msgInner.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
       if (traFlag != null) {
           response.setCode(ResponseCode.NO_PERMISSION);
           response.setRemark(
               "the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1() + "] sending transaction message is forbidden");
           return response;
       }
   }
   // 调用MessageStore存储消息
   PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
   //... 统计与响应
   return response;
}

当消息为重试消息且超过设定的最大消费次数时，将会被放入死信队列中。
经过一些校验后，调用MessageStore.putMessage（详见第3章 store）存储消息，最后进行broker接收消息的统计和响应。

您的支持是我创作源源不断的动力