RocketMQ源码分析——Broker接收发送消息

发表于 | 分类于 | 本文总阅读量 次

引：前面我们说过了Producer和Consumer是如何发送消息的，当我们了解完Broker的消息存储之后，我们就可以看看Broker是如何接收发来的消息啦~

前言

在分析完Producer和Consumer之后，我们知道无论是消费者还是生产者来说，Broker都是接受消息的一方，同时我们知道RocketMQ是通过Netty实现通信的，对应接收消息，我们很容易就会想到一定是一个ChannelHandler在处理，而在RocketMQ的设计中对应是NettyRequestProcessor，所以我们就可以分析一下接收消息的几种类型啦，有几种消息类型，就有几种Processor，如下图：
broker-processor.png

我们看到有很多种消息类型，我们这次只分析SendMessage。

SendMessage

对于SendMessage有下面这几种可能：单条消息、批量消息、RETRY消息。Retry消息即consumer消费失败，要求broker重发的消息。失败的原因有两种，一种是业务端代码处理失败；还有一种是消息在consumer的缓存队列中待的时间超时，consumer会将消息从队列中移除，然后退回给Broker重发。

SendMessageProcessor

我们肯定会找到org.apache.rocketmq.broker.processor.SendMessageProcessor#processRequest()这个方法，因为它实现自NettyRequestProcessor接口的：

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
public RemotingCommand processRequest(ChannelHandlerContext ctx,
                                      RemotingCommand request) throws RemotingCommandException {
    SendMessageContext mqtraceContext;
    switch (request.getCode()) {
        case RequestCode.CONSUMER_SEND_MSG_BACK:
            // RETRY消息
            return this.consumerSendMsgBack(ctx, request);
        default:
            SendMessageRequestHeader requestHeader = parseRequestHeader(request);
            if (requestHeader == null) {
                return null;
            }

            mqtraceContext = buildMsgContext(ctx, requestHeader);
            this.executeSendMessageHookBefore(ctx, request, mqtraceContext);

            RemotingCommand response;
            if (requestHeader.isBatch()) {
                // 批量发送
                response = this.sendBatchMessage(ctx, request, mqtraceContext, requestHeader);
            } else {
                // 单条消息
                response = this.sendMessage(ctx, request, mqtraceContext, requestHeader);
            }

            this.executeSendMessageHookAfter(response, mqtraceContext);
            return response;
    }
}

Consumer的RETRY消息

对于Consumer的RETRY消息，我们可以找到org.apache.rocketmq.broker.processor.SendMessageProcessor#consumerSendMsgBack()方法：

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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
private RemotingCommand consumerSendMsgBack(final ChannelHandlerContext ctx, final RemotingCommand request)
       throws RemotingCommandException {
       final RemotingCommand response = RemotingCommand.createResponseCommand(null);
       final ConsumerSendMsgBackRequestHeader requestHeader =
           (ConsumerSendMsgBackRequestHeader)request.decodeCommandCustomHeader(ConsumerSendMsgBackRequestHeader.class);

 			// ...
			// consumer group是否存在
       SubscriptionGroupConfig subscriptionGroupConfig =
           this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(requestHeader.getGroup());
       if (null == subscriptionGroupConfig) {
           response.setCode(ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST);
           response.setRemark("subscription group not exist, " + requestHeader.getGroup() + " "
               + FAQUrl.suggestTodo(FAQUrl.SUBSCRIPTION_GROUP_NOT_EXIST));
           return response;
       }
			// ... 
       // 获得重试队列Topic和queue
       String newTopic = MixAll.getRetryTopic(requestHeader.getGroup());
       int queueIdInt = Math.abs(this.random.nextInt() % 99999999) % subscriptionGroupConfig.getRetryQueueNums();

       int topicSysFlag = 0;
       if (requestHeader.isUnitMode()) {
           topicSysFlag = TopicSysFlag.buildSysFlag(false, true);
       }

       TopicConfig topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(
           newTopic,
           subscriptionGroupConfig.getRetryQueueNums(),
     	// ...
			// 根据offset找到MessageExt
       MessageExt msgExt = this.brokerController.getMessageStore().lookMessageByOffset(requestHeader.getOffset());
       if (null == msgExt) {
           response.setCode(ResponseCode.SYSTEM_ERROR);
           response.setRemark("look message by offset failed, " + requestHeader.getOffset());
           return response;
       }

       final String retryTopic = msgExt.getProperty(MessageConst.PROPERTY_RETRY_TOPIC);
       if (null == retryTopic) {
           MessageAccessor.putProperty(msgExt, MessageConst.PROPERTY_RETRY_TOPIC, msgExt.getTopic());
       }
       msgExt.setWaitStoreMsgOK(false);
			// 查看延迟等级
       //  messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
       int delayLevel = requestHeader.getDelayLevel();
			// 最大重复次数16次
       int maxReconsumeTimes = subscriptionGroupConfig.getRetryMaxTimes();
       if (request.getVersion() >= MQVersion.Version.V3_4_9.ordinal()) {
           maxReconsumeTimes = requestHeader.getMaxReconsumeTimes();
       }
       // 如果大于重试次数，则放入DLQ死信队列
       if (msgExt.getReconsumeTimes() >= maxReconsumeTimes
           || delayLevel < 0) {
         	// 将topic和queueid设置为死信队列的
           newTopic = MixAll.getDLQTopic(requestHeader.getGroup());
           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
           );
           if (null == topicConfig) {
               response.setCode(ResponseCode.SYSTEM_ERROR);
               response.setRemark("topic[" + newTopic + "] not exist");
               return response;
           }
       } else {
     			// 加三个等级的延迟
           if (0 == delayLevel) {
               delayLevel = 3 + msgExt.getReconsumeTimes();
           }

           msgExt.setDelayTimeLevel(delayLevel);
       }
			// 将msgExt转化为MessageExtBrokerInner
       MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
       msgInner.setTopic(newTopic);
       msgInner.setBody(msgExt.getBody());
       msgInner.setFlag(msgExt.getFlag());
       MessageAccessor.setProperties(msgInner, msgExt.getProperties());
       msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));
       msgInner.setTagsCode(MessageExtBrokerInner.tagsString2tagsCode(null, msgExt.getTags()));

       msgInner.setQueueId(queueIdInt);
       msgInner.setSysFlag(msgExt.getSysFlag());
       msgInner.setBornTimestamp(msgExt.getBornTimestamp());
       msgInner.setBornHost(msgExt.getBornHost());
       msgInner.setStoreHost(this.getStoreHost());
       msgInner.setReconsumeTimes(msgExt.getReconsumeTimes() + 1);

       String originMsgId = MessageAccessor.getOriginMessageId(msgExt);
       MessageAccessor.setOriginMessageId(msgInner, UtilAll.isBlank(originMsgId) ? msgExt.getMsgId() : originMsgId);
       // 存储消息
       PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
       if (putMessageResult != null) {
           switch (putMessageResult.getPutMessageStatus()) {
               case PUT_OK:
                   String backTopic = msgExt.getTopic();
                   String correctTopic = msgExt.getProperty(MessageConst.PROPERTY_RETRY_TOPIC);
                   if (correctTopic != null) {
                       backTopic = correctTopic;
                   }

                   this.brokerController.getBrokerStatsManager().incSendBackNums(requestHeader.getGroup(), backTopic);

                   response.setCode(ResponseCode.SUCCESS);
                   response.setRemark(null);

                   return response;
               default:
                   break;
           }

           response.setCode(ResponseCode.SYSTEM_ERROR);
           response.setRemark(putMessageResult.getPutMessageStatus().name());
           return response;
       }

       response.setCode(ResponseCode.SYSTEM_ERROR);
       response.setRemark("putMessageResult is null");
       return response;
   }

Producer的单条消息

对于Producer的单条消息，我们可以找到org.apache.rocketmq.broker.processor.SendMessageProcessor#sendMessage()方法：

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
74
75
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()));

    log.debug("receive SendMessage request command, {}", request);
    // 判断当前时间broker是否提供服务，不提供则返回code为SYSTEM_ERROR的response
    final long startTimstamp = this.brokerController.getBrokerConfig().getStartAcceptSendRequestTimeStamp();
    if (this.brokerController.getMessageStore().now() < startTimstamp) {
        response.setCode(ResponseCode.SYSTEM_ERROR);
        response.setRemark(String.format("broker unable to service, until %s", UtilAll.timeMillisToHumanString2(startTimstamp)));
        return response;
    }

    response.setCode(-1);
    // 检查topic和queue，如果不存在且broker设置中允许自动创建，则自动创建
    super.msgCheck(ctx, requestHeader, response);
    if (response.getCode() != -1) {
        return response;
    }

    final byte[] body = request.getBody();

    int queueIdInt = requestHeader.getQueueId();
    // 获取topic的配置
    TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
    // 如果消息中的queueId小于0，则随机选取一个queue
    if (queueIdInt < 0) {
        queueIdInt = Math.abs(this.random.nextInt() % 99999999) % topicConfig.getWriteQueueNums();
    }
    // 重新封装request中的message成MessageExtBrokerInner
    MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
    msgInner.setTopic(requestHeader.getTopic());
    msgInner.setQueueId(queueIdInt);
    // 对于RETRY消息的一些处理
    if (!handleRetryAndDLQ(requestHeader, response, request, msgInner, topicConfig)) {
        return response;
    }
    // 转化消息为MessageExtBrokerInner
    msgInner.setBody(body);
    msgInner.setFlag(requestHeader.getFlag());
    MessageAccessor.setProperties(msgInner, MessageDecoder.string2messageProperties(requestHeader.getProperties()));
    msgInner.setPropertiesString(requestHeader.getProperties());
    msgInner.setBornTimestamp(requestHeader.getBornTimestamp());
    msgInner.setBornHost(ctx.channel().remoteAddress());
    msgInner.setStoreHost(this.getStoreHost());
    msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());
    PutMessageResult putMessageResult = null;
    Map<String, String> oriProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());
    String traFlag = oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
    if (traFlag != null && Boolean.parseBoolean(traFlag)) {
        // 是否拒绝事务消息
        if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {
            response.setCode(ResponseCode.NO_PERMISSION);
            response.setRemark(
                "the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
                    + "] sending transaction message is forbidden");
            return response;
        }
        putMessageResult = this.brokerController.getTransactionalMessageService().prepareMessage(msgInner);
    } else {
        // 调用MessageStore接口存储消息，这里就到我们之前分析的消息存储哪里了
        putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
    }
    // 根据存储结果设置repsonse状态，更新broker统计信息，成功则回复producer，更新context上下文
    return handlePutMessageResult(putMessageResult, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt);

}

问题

其实我有个问题就是对于Producer的单条信息，为了要判断重试次数？？ 问题先放在这，等以后知道了，再来回答！！

参考

  1. RocketMQ源码解析(八)-Broker#消息发送

  2. RocketMQ消费失败消息深入分析（consumer,broker的具体处理逻辑）