RocketMQ(4.8.0)——消费方式消费方式

　　RocketMQ 包含2种消费方式：

- Pull
- Push

　　Pull方式：用户主动Pull消息，自主管理位点。默认的 Push 消费者实现类：D: ocketmq-masterclientsrcmainjavaorgapache ocketmqclientconsumerDefaultMQPullConsumer.java

　　　　优点：可以灵活地掌控消费进度和消费速度，适合流计算、消费特别耗时等特殊的消费场景。

　　　　缺点：需要从代码层面精准地控制消费，对开发人员有一定要求。

　　Push方式：用户主动Pull消息，自主管理位点。默认的 Push 消费者实现类：D: ocketmq-masterclientsrcmainjavaorgapache ocketmqclientconsumerDefaultMQPushConsumer.java

　　　　优点：代码接入非常简单，适合大部分业务场景。

　　　　缺点：灵活度差，在了解其消费原理后，排查消费问题方可简单快捷。

　　针对Pull和Push，下面对两种方式进行简单的比较。

消费方式/对比项	Pull	Push	备注
是否需要主动拉取	理解分区后，需要主动拉取各个分区消息	自动	Pull 消息灵活，Push使用更简单
位点管理	用户自行管理或者主动提交给 Broker 管理	Broker 管理	Pull 自主管理位点，消费灵活； Push 位点交由 Broker 管理
Topic 路由变更是否影响消费	否	否	Pull 模式需要编码实现路由感知； Push 模式自动换行 Rebalance，以适应路由变更。

1.1 Pull消费流程

　　D: ocketmq-masterclientsrcmainjavaorgapache ocketmqclientconsumerDefaultMQPullConsumer.java 消费过程如下：

RocketMQ(4.8.0)——消费方式
消费方式

Pull消费的具体步骤：

　　第一步：fetchSubscribeMessageQueue(String Topic)。拉取全部可以消费的 Queue。如果某一个 Broker 下线，这里也可以实时感知到。

　　第二步：遍历全部Queue，拉取每个 Queue 可以消费的消息。

　　第三步：如果拉取到消息，则执行用户编写的消费代码。

　　第四步：保存消费进度。消费进度可以执行 updateConsumeOffset()方法，将消费位点上报给 Broker，也可以自行保存消费位点。比如流计算平台Flink使用Pull方式拉取消息消费，通过 Checkpoint 管理消费进度。

1.2 Push消费流程

RocketMQ(4.8.0)——消费方式
消费方式

Push消费过程如下：

　　第一步：初始化 Push 消费者实例。业务代码初始化 DefaultMQPushConsumer 实例，启动 Pull 服务 PullMessageService。该服务是一个线程服务，不断执行 run() 方法拉取已经订阅 Topic 的全部队列消息，将消息保存在本地的缓存队列中。

　　第二步：消费消息。由消息服务 ConsumeMessageConcurrentlyService 或者 ConsumeMessageOrderlyService 将本地缓存队列中的消息不断放入消费线程池，异步回调业务消费代码，此时业务代码可以消费消息。(核心知识点)

　　第三步：保存消费进度。业务代码消费后，将消费结果返回给消费服务，再由消费服务将消费进度保存在本地，由消费进度管理服务定时和不定时地持久化到本地（LocalFileOffsetStore 支持）或者远程 Broker（RemoteBrokerOffsetStore支持）种。对于消费失败的消息，RocketMQ 客户端处理后发回给 Broker，并告知消费失败。(核心知识点)

Push消费者如何拉取消息消费：

RocketMQ(4.8.0)——消费方式
消费方式

　　第一步：PullMessageService 不断拉取消息。

　　第二步：消费者拉取消息并消费。

　　　　2.1 基本校验。校验 ProcessQueue 是否dropped;校验消费者服务状态是否正常;校验消费者是否被挂起。

　　　　2.2 拉取条数、字节数限制检查。如果本地缓存消息数量大于配置的最大拉取条数（默认为 1000，可以调整），则延迟一段时间再拉取；如果本地缓存消息字节数大于配置的最大缓存字节数，则延迟一段时间再拉取。这两种校验方式都相当于本地流控。

　　　　2.3 并发消费和顺序消费校验。

　　　　　　在并发消费时，processQueue.getMaxSpan()方法是用于计算本地缓存队列中第一个消息和最后一个消息的 offset 差值。

RocketMQ(4.8.0)——消费方式
消费方式

　　　　　　本地缓存队列的 Span 如果大于配置的最大差值（默认为2000，可以调整），则认为本地消费过慢，需要执行本地流控。

　　　　　　顺序消费时，如果当前拉取的队列在 Broker 端没有被锁定，说明已经有拉取正在执行，当前拉取请求晚点执行；如果不是第一次拉取，需要先计算最新的拉取位点并修正本地最新的待拉取位点信息，再执行拉取。代码路径：D: ocketmq-masterclientsrcmainjavaorgapache ocketmqclientimplconsumerDefaultMQPushConsumerImpl.java

 1         if (!this.consumeOrderly) {
 2             if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {
 3                 this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
 4                 if ((queueMaxSpanFlowControlTimes++ % 1000) == 0) {
 5                     log.warn(
 6                         "the queue's messages, span too long, so do flow control, minOffset={}, maxOffset={}, maxSpan={}, pullRequest={}, flowControlTimes={}",
 7                         processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), processQueue.getMaxSpan(),
 8                         pullRequest, queueMaxSpanFlowControlTimes);
 9                 }
10                 return;
11             }
12         } else {
13             if (processQueue.isLocked()) {
14                 if (!pullRequest.isLockedFirst()) {
15                     final long offset = this.rebalanceImpl.computePullFromWhere(pullRequest.getMessageQueue());
16                     boolean brokerBusy = offset < pullRequest.getNextOffset();
17                     log.info("the first time to pull message, so fix offset from broker. pullRequest: {} NewOffset: {} brokerBusy: {}",
18                         pullRequest, offset, brokerBusy);
19                     if (brokerBusy) {
20                         log.info("[NOTIFYME]the first time to pull message, but pull request offset larger than broker consume offset. pullRequest: {} NewOffset: {}",
21                             pullRequest, offset);
22                     }
23 
24                     pullRequest.setLockedFirst(true);
25                     pullRequest.setNextOffset(offset);
26                 }
27             } else {
28                 this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
29                 log.info("pull message later because not locked in broker, {}", pullRequest);
30                 return;
31             }
32         }

View Code

　　(1) 订阅关系校验。如果待拉取的 Topic 在本地缓存中订阅关系为空，则本地拉取不执行，待下一个正常心跳或者 Rebalance 后订阅关系恢复正常，方可正常拉取。

　　(2) 封装拉取请求和拉取后的回调对象 PullCallback。这里主要将消息拉取请求和拉取结果处理封装成 PullCallback，并通过调用 PullAPIWrapper.pullKernelImpl() 方法拉取请求发出。

　　拉取结果存在多种可能性。这里以拉取消息的情况举例说下：

 1                                 boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
 2                                 DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
 3                                     pullResult.getMsgFoundList(),
 4                                     processQueue,
 5                                     pullRequest.getMessageQueue(),
 6                                     dispatchToConsume);
 7 
 8                                 if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) {
 9                                     DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,
10                                         DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());
11                                 } else {
12                                     DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
13                                 }
14                             }
15 
16                             if (pullResult.getNextBeginOffset() < prevRequestOffset
17                                 || firstMsgOffset < prevRequestOffset) {
18                                 log.warn(
19                                     "[BUG] pull message result maybe data wrong, nextBeginOffset: {} firstMsgOffset: {} prevRequestOffset: {}",
20                                     pullResult.getNextBeginOffset(),
21                                     firstMsgOffset,
22                                     prevRequestOffset);
23                             }
24 
25                             break;

View Code

　　如果拉取到消息，那么将消息保存到对应的本地缓存队列 ProcessQueue 中，然后将这些消息提交给 ConsumeMessageService 服务。

　　ConsumeMessageService 是一个通用消费服务接口，它包含两个实现类：orgapache ocketmqclientimplconsumerConsumeMessageConcurrentlyService 和 srcmainjavaorgapache ocketmqclientimplconsumerConsumeMessageOrderlyService，这两个类分别用于并发消费和顺序消费。

 1 /*
 2  * Licensed to the Apache Software Foundation (ASF) under one or more
 3  * contributor license agreements.  See the NOTICE file distributed with
 4  * this work for additional information regarding copyright ownership.
 5  * The ASF licenses this file to You under the Apache License, Version 2.0
 6  * (the "License"); you may not use this file except in compliance with
 7  * the License.  You may obtain a copy of the License at
 8  *
 9  *     http://www.apache.org/licenses/LICENSE-2.0
10  *
11  * Unless required by applicable law or agreed to in writing, software
12  * distributed under the License is distributed on an "AS IS" BASIS,
13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  * See the License for the specific language governing permissions and
15  * limitations under the License.
16  */
17 package org.apache.rocketmq.client.impl.consumer;
18 
19 import java.util.List;
20 import org.apache.rocketmq.common.message.MessageExt;
21 import org.apache.rocketmq.common.message.MessageQueue;
22 import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
23 
24 public interface ConsumeMessageService {
25     void start();
26 
27     void shutdown(long awaitTerminateMillis);
28 
29     void updateCorePoolSize(int corePoolSize);
30 
31     void incCorePoolSize();
32 
33     void decCorePoolSize();
34 
35     int getCorePoolSize();
36 
37     ConsumeMessageDirectlyResult consumeMessageDirectly(final MessageExt msg, final String brokerName);
38 
39     void submitConsumeRequest(
40         final List<MessageExt> msgs,
41         final ProcessQueue processQueue,
42         final MessageQueue messageQueue,
43         final boolean dispathToConsume);
44 }

ConsumeMessageService{}

　　start()方法 和 shutdown()方法 分别在启动和关闭服务时使用。

　　updateCorePoolSize()：更新消费线程池的核心线程数。
　　incCorePoolSize()：增加一个消费线程池的核心线程数。

　　decCorePoolSize()：减少一个消费线程池的核心线程数。

　　getCorePoolSize()：获取消费线程池的核心线程数。

　　consumeMessageDirectly()：如果一个消息已经被消费过了，但是还想再消费一次，就需要实现这个方法。
　　submitConsumeRequest()：将消息封装成线程池任务，提交给消费服务，消费服务再将消息传递给业务消费进行处理。

　　(1) ConsumeMessageService 消息消费分发。ConsumeMessageService 服务通过 DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest 接口接收消息消费任务后，将消息按照固定条数封装成多个 ConsumeRequest 任务对象，并发送到消费线程，等待分发给业务消费；ConsumeMessageOrderlyService 先将 Pull 的全部消息放在另外一个本地队列中，然后提交一个 ConsumeRequest 到消费线程池。

　　(2) 消费消息。消费的主要逻辑在 ConsumeMessageService 接口的两个实现类中。下面以并发消息实现类 orgapache ocketmqclientimplconsumerConsumeMessageConcurrentlyService，代码如下：

 1         @Override
 2         public void run() {
 3             if (this.processQueue.isDropped()) {
 4                 log.info("the message queue not be able to consume, because it's dropped. group={} {}", ConsumeMessageConcurrentlyService.this.consumerGroup, this.messageQueue);
 5                 return;
 6             }
 7 
 8             MessageListenerConcurrently listener = ConsumeMessageConcurrentlyService.this.messageListener;
 9             ConsumeConcurrentlyContext context = new ConsumeConcurrentlyContext(messageQueue);
10             ConsumeConcurrentlyStatus status = null;
11             defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
12 
13             ConsumeMessageContext consumeMessageContext = null;
14             //消费前
15             if (ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.hasHook()) {
16                 consumeMessageContext = new ConsumeMessageContext();
17                 consumeMessageContext.setNamespace(defaultMQPushConsumer.getNamespace());
18                 consumeMessageContext.setConsumerGroup(defaultMQPushConsumer.getConsumerGroup());
19                 consumeMessageContext.setProps(new HashMap<String, String>());
20                 consumeMessageContext.setMq(messageQueue);
21                 consumeMessageContext.setMsgList(msgs);
22                 consumeMessageContext.setSuccess(false);
23                 ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext);
24             }
25 
26             long beginTimestamp = System.currentTimeMillis();
27             boolean hasException = false;
28             ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
29             try {
30                 //预处理重试队列消息
31                 if (msgs != null && !msgs.isEmpty()) {
32                     for (MessageExt msg : msgs) {
33                         MessageAccessor.setConsumeStartTimeStamp(msg, String.valueOf(System.currentTimeMillis()));
34                     }
35                 }
36                 //消费回调
37                 status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
38             } catch (Throwable e) {
39                 log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}",
40                     RemotingHelper.exceptionSimpleDesc(e),
41                     ConsumeMessageConcurrentlyService.this.consumerGroup,
42                     msgs,
43                     messageQueue);
44                 hasException = true;
45             }
46             long consumeRT = System.currentTimeMillis() - beginTimestamp;
47             if (null == status) {
48                 if (hasException) {
49                     returnType = ConsumeReturnType.EXCEPTION;
50                 } else {
51                     returnType = ConsumeReturnType.RETURNNULL;
52                 }
53             } else if (consumeRT >= defaultMQPushConsumer.getConsumeTimeout() * 60 * 1000) {
54                 returnType = ConsumeReturnType.TIME_OUT;
55             } else if (ConsumeConcurrentlyStatus.RECONSUME_LATER == status) {
56                 returnType = ConsumeReturnType.FAILED;
57             } else if (ConsumeConcurrentlyStatus.CONSUME_SUCCESS == status) {
58                 returnType = ConsumeReturnType.SUCCESS;
59             }
60             //消费执行后
61             if (ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.hasHook()) {
62                 consumeMessageContext.getProps().put(MixAll.CONSUME_CONTEXT_TYPE, returnType.name());
63             }
64 
65             if (null == status) {
66                 log.warn("consumeMessage return null, Group: {} Msgs: {} MQ: {}",
67                     ConsumeMessageConcurrentlyService.this.consumerGroup,
68                     msgs,
69                     messageQueue);
70                 status = ConsumeConcurrentlyStatus.RECONSUME_LATER;
71             }
72 
73             if (ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.hasHook()) {
74                 consumeMessageContext.setStatus(status.toString());
75                 consumeMessageContext.setSuccess(ConsumeConcurrentlyStatus.CONSUME_SUCCESS == status);
76                 ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.executeHookAfter(consumeMessageContext);
77             }
78 
79             ConsumeMessageConcurrentlyService.this.getConsumerStatsManager()
80                 .incConsumeRT(ConsumeMessageConcurrentlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);
81 
82             if (!processQueue.isDropped()) {
83                 //处理消费结果
84                 ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this);
85             } else {
86                 log.warn("processQueue is dropped without process consume result. messageQueue={}, msgs={}", messageQueue, msgs);
87             }
88         }
89 
90         public MessageQueue getMessageQueue() {
91             return messageQueue;
92         }
93 
94     }
95 }

View Code

　　消费消息主要分为 消费前预处理、消费回调、消费结果统计、消费结果 处理 4 个步骤。

　　第一步：消费执行前进行预处理。执行消费前的 hook 和重试消息预处理。消费前的 hook 可以理解为消费前的消息预处理（比如消息格式校验）。如果拉取的消息来自重试队列，则将 Topic 名重置为原来的 Topic 名，而不用重试 Topic 名。

　　第二步：消费回调。首先设置消息开始消费时间为当前时间，再将消息列表转为不可修改的 List，然后通过 listener.consumeMessage(Collections.unmodifiableList(msgs), context) 方法将消息传递给用户编写的业务消费代码进行处理。

　　第三步：消费结果统计和执行消费后的 hook。客户端原生支持基本消费指标统计，比如消费耗时；消费后的 hook 和消费前的 hook 要一一对应，用户可以用消费后的 hook 统计与自身业务相关的指标。

　　第四步：消费结果处理。包含消费指标统计、消费重试处理和消费位点处理。消费指标主要是对消费成功和失败的 TPS 的统计；消费重试处理主要将消费重试次数加 1；消费位点处理主要根据消费结构更新消费位点记录。

　　至此，Push 消费流程完毕。

　　RocketMQ 是一个消息队列，FIFO(Fist In First Out，先进先出)规则如何在消费失败时保证消息的顺序执行呢？

　　从消费任务实现类 ConsumeRequest 和本地缓存队列 ProcessQueue 的涉及来看主要差异。并发消息（无序消费）的消费请求对象实现类代码路径：D: ocketmq-masterclientsrcmainjavaorgapache ocketmqclientimplconsumerConsumeMessageConcurrentlyService.java，代码如下：

  1    class ConsumeRequest implements Runnable {
  2         private final List<MessageExt> msgs;
  3         private final ProcessQueue processQueue;
  4         private final MessageQueue messageQueue;
  5 
  6         public ConsumeRequest(List<MessageExt> msgs, ProcessQueue processQueue, MessageQueue messageQueue) {
  7             this.msgs = msgs;
  8             this.processQueue = processQueue;
  9             this.messageQueue = messageQueue;
 10         }
 11 
 12         public List<MessageExt> getMsgs() {
 13             return msgs;
 14         }
 15 
 16         public ProcessQueue getProcessQueue() {
 17             return processQueue;
 18         }
 19 
 20         @Override
 21         public void run() {
 22             if (this.processQueue.isDropped()) {
 23                 log.info("the message queue not be able to consume, because it's dropped. group={} {}", ConsumeMessageConcurrentlyService.this.consumerGroup, this.messageQueue);
 24                 return;
 25             }
 26 
 27             MessageListenerConcurrently listener = ConsumeMessageConcurrentlyService.this.messageListener;
 28             ConsumeConcurrentlyContext context = new ConsumeConcurrentlyContext(messageQueue);
 29             ConsumeConcurrentlyStatus status = null;
 30             defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
 31 
 32             ConsumeMessageContext consumeMessageContext = null;
 33             //消费前
 34             if (ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.hasHook()) {
 35                 consumeMessageContext = new ConsumeMessageContext();
 36                 consumeMessageContext.setNamespace(defaultMQPushConsumer.getNamespace());
 37                 consumeMessageContext.setConsumerGroup(defaultMQPushConsumer.getConsumerGroup());
 38                 consumeMessageContext.setProps(new HashMap<String, String>());
 39                 consumeMessageContext.setMq(messageQueue);
 40                 consumeMessageContext.setMsgList(msgs);
 41                 consumeMessageContext.setSuccess(false);
 42                 ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext);
 43             }
 44 
 45             long beginTimestamp = System.currentTimeMillis();
 46             boolean hasException = false;
 47             ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
 48             try {
 49                 //预处理重试队列消息
 50                 if (msgs != null && !msgs.isEmpty()) {
 51                     for (MessageExt msg : msgs) {
 52                         MessageAccessor.setConsumeStartTimeStamp(msg, String.valueOf(System.currentTimeMillis()));
 53                     }
 54                 }
 55                 //消费回调
 56                 status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
 57             } catch (Throwable e) {
 58                 log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}",
 59                     RemotingHelper.exceptionSimpleDesc(e),
 60                     ConsumeMessageConcurrentlyService.this.consumerGroup,
 61                     msgs,
 62                     messageQueue);
 63                 hasException = true;
 64             }
 65             long consumeRT = System.currentTimeMillis() - beginTimestamp;
 66             if (null == status) {
 67                 if (hasException) {
 68                     returnType = ConsumeReturnType.EXCEPTION;
 69                 } else {
 70                     returnType = ConsumeReturnType.RETURNNULL;
 71                 }
 72             } else if (consumeRT >= defaultMQPushConsumer.getConsumeTimeout() * 60 * 1000) {
 73                 returnType = ConsumeReturnType.TIME_OUT;
 74             } else if (ConsumeConcurrentlyStatus.RECONSUME_LATER == status) {
 75                 returnType = ConsumeReturnType.FAILED;
 76             } else if (ConsumeConcurrentlyStatus.CONSUME_SUCCESS == status) {
 77                 returnType = ConsumeReturnType.SUCCESS;
 78             }
 79             //消费执行后
 80             if (ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.hasHook()) {
 81                 consumeMessageContext.getProps().put(MixAll.CONSUME_CONTEXT_TYPE, returnType.name());
 82             }
 83 
 84             if (null == status) {
 85                 log.warn("consumeMessage return null, Group: {} Msgs: {} MQ: {}",
 86                     ConsumeMessageConcurrentlyService.this.consumerGroup,
 87                     msgs,
 88                     messageQueue);
 89                 status = ConsumeConcurrentlyStatus.RECONSUME_LATER;
 90             }
 91 
 92             if (ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.hasHook()) {
 93                 consumeMessageContext.setStatus(status.toString());
 94                 consumeMessageContext.setSuccess(ConsumeConcurrentlyStatus.CONSUME_SUCCESS == status);
 95                 ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.executeHookAfter(consumeMessageContext);
 96             }
 97 
 98             ConsumeMessageConcurrentlyService.this.getConsumerStatsManager()
 99                 .incConsumeRT(ConsumeMessageConcurrentlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);
100 
101             if (!processQueue.isDropped()) {
102                 //处理消费结果
103                 ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this);
104             } else {
105                 log.warn("processQueue is dropped without process consume result. messageQueue={}, msgs={}", messageQueue, msgs);
106             }
107         }
108 
109         public MessageQueue getMessageQueue() {
110             return messageQueue;
111         }
112 
113     }

ConsumeRequest()

　　顺序消费的消费请求对象实现类为 D: ocketmq-masterclientsrcmainjavaorgapache ocketmqclientimplconsumerConsumeMessageConcurrentlyService.ConsumeRequest，代码如下：

  1 class ConsumeRequest implements Runnable {
  2         private final ProcessQueue processQueue;
  3         private final MessageQueue messageQueue;
  4 
  5         public ConsumeRequest(ProcessQueue processQueue, MessageQueue messageQueue) {
  6             this.processQueue = processQueue;
  7             this.messageQueue = messageQueue;
  8         }
  9 
 10         public ProcessQueue getProcessQueue() {
 11             return processQueue;
 12         }
 13 
 14         public MessageQueue getMessageQueue() {
 15             return messageQueue;
 16         }
 17 
 18         @Override
 19         public void run() {
 20             if (this.processQueue.isDropped()) {
 21                 log.warn("run, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
 22                 return;
 23             }
 24 
 25             final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
 26             synchronized (objLock) {
 27                 if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
 28                     || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
 29                     final long beginTime = System.currentTimeMillis();
 30                     for (boolean continueConsume = true; continueConsume; ) {
 31                         if (this.processQueue.isDropped()) {
 32                             log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
 33                             break;
 34                         }
 35 
 36                         if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
 37                             && !this.processQueue.isLocked()) {
 38                             log.warn("the message queue not locked, so consume later, {}", this.messageQueue);
 39                             ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
 40                             break;
 41                         }
 42 
 43                         if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
 44                             && this.processQueue.isLockExpired()) {
 45                             log.warn("the message queue lock expired, so consume later, {}", this.messageQueue);
 46                             ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
 47                             break;
 48                         }
 49 
 50                         long interval = System.currentTimeMillis() - beginTime;
 51                         if (interval > MAX_TIME_CONSUME_CONTINUOUSLY) {
 52                             ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, messageQueue, 10);
 53                             break;
 54                         }
 55 
 56                         final int consumeBatchSize =
 57                             ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
 58 
 59                         List<MessageExt> msgs = this.processQueue.takeMessages(consumeBatchSize);
 60                         defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
 61                         if (!msgs.isEmpty()) {
 62                             final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue);
 63 
 64                             ConsumeOrderlyStatus status = null;
 65 
 66                             ConsumeMessageContext consumeMessageContext = null;
 67                             if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
 68                                 consumeMessageContext = new ConsumeMessageContext();
 69                                 consumeMessageContext
 70                                     .setConsumerGroup(ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumerGroup());
 71                                 consumeMessageContext.setNamespace(defaultMQPushConsumer.getNamespace());
 72                                 consumeMessageContext.setMq(messageQueue);
 73                                 consumeMessageContext.setMsgList(msgs);
 74                                 consumeMessageContext.setSuccess(false);
 75                                 // init the consume context type
 76                                 consumeMessageContext.setProps(new HashMap<String, String>());
 77                                 ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext);
 78                             }
 79 
 80                             long beginTimestamp = System.currentTimeMillis();
 81                             ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
 82                             boolean hasException = false;
 83                             try {
 84                                 this.processQueue.getLockConsume().lock();
 85                                 if (this.processQueue.isDropped()) {
 86                                     log.warn("consumeMessage, the message queue not be able to consume, because it's dropped. {}",
 87                                         this.messageQueue);
 88                                     break;
 89                                 }
 90 
 91                                 status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
 92                             } catch (Throwable e) {
 93                                 log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}",
 94                                     RemotingHelper.exceptionSimpleDesc(e),
 95                                     ConsumeMessageOrderlyService.this.consumerGroup,
 96                                     msgs,
 97                                     messageQueue);
 98                                 hasException = true;
 99                             } finally {
100                                 this.processQueue.getLockConsume().unlock();
101                             }
102 
103                             if (null == status
104                                 || ConsumeOrderlyStatus.ROLLBACK == status
105                                 || ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) {
106                                 log.warn("consumeMessage Orderly return not OK, Group: {} Msgs: {} MQ: {}",
107                                     ConsumeMessageOrderlyService.this.consumerGroup,
108                                     msgs,
109                                     messageQueue);
110                             }
111 
112                             long consumeRT = System.currentTimeMillis() - beginTimestamp;
113                             if (null == status) {
114                                 if (hasException) {
115                                     returnType = ConsumeReturnType.EXCEPTION;
116                                 } else {
117                                     returnType = ConsumeReturnType.RETURNNULL;
118                                 }
119                             } else if (consumeRT >= defaultMQPushConsumer.getConsumeTimeout() * 60 * 1000) {
120                                 returnType = ConsumeReturnType.TIME_OUT;
121                             } else if (ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) {
122                                 returnType = ConsumeReturnType.FAILED;
123                             } else if (ConsumeOrderlyStatus.SUCCESS == status) {
124                                 returnType = ConsumeReturnType.SUCCESS;
125                             }
126 
127                             if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
128                                 consumeMessageContext.getProps().put(MixAll.CONSUME_CONTEXT_TYPE, returnType.name());
129                             }
130 
131                             if (null == status) {
132                                 status = ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
133                             }
134 
135                             if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
136                                 consumeMessageContext.setStatus(status.toString());
137                                 consumeMessageContext
138                                     .setSuccess(ConsumeOrderlyStatus.SUCCESS == status || ConsumeOrderlyStatus.COMMIT == status);
139                                 ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookAfter(consumeMessageContext);
140                             }
141 
142                             ConsumeMessageOrderlyService.this.getConsumerStatsManager()
143                                 .incConsumeRT(ConsumeMessageOrderlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);
144 
145                             continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);
146                         } else {
147                             continueConsume = false;
148                         }
149                     }
150                 } else {
151                     if (this.processQueue.isDropped()) {
152                         log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
153                         return;
154                     }
155 
156                     ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
157                 }
158             }
159         }
160 
161     }

View Code

　　由上面代码可知，顺序消息的 ConsumeRequest 中并没有保存需要消费的消息，在顺序消费时通过调用 ProcessQueue.takeMessage() 方法获取需要消费的消息，而且消费也是同步进行的。

　　msgTreeMap：是一个 TreeMap<Long，MessageExt>类型，key是消息物理位点值，value 是消息对象，该容器是 ProcessQueue 用来缓存本地顺序消息的，保存的数据是按照key（就是物理位点值）顺序排列的。

　　consumingMsgOrderlyTreeMap：是一个TreeMap<Long，MessageExt>类型，key是消息物理位点值，value 是消息对象，保存当前正在处理的顺序消息集合，是 msgTreeMap 的一个子集。保存的数据是按照 key（就是物理机位点值）顺序排列的。

　　batchSize：一次从本地缓存中获取多少条消息回调给用户消费。顺序消费是如何通过 PorcessQueue.takeMessage() 获取消息给业务代码消费的呢？

 1     public List<MessageExt> takeMessages(final int batchSize) {
 2         List<MessageExt> result = new ArrayList<MessageExt>(batchSize);
 3         final long now = System.currentTimeMillis();
 4         try {
 5             this.lockTreeMap.writeLock().lockInterruptibly();
 6             this.lastConsumeTimestamp = now;
 7             try {
 8                 if (!this.msgTreeMap.isEmpty()) {
 9                     for (int i = 0; i < batchSize; i++) {
10                         Map.Entry<Long, MessageExt> entry = this.msgTreeMap.pollFirstEntry();
11                         if (entry != null) {
12                             result.add(entry.getValue());
13                             consumingMsgOrderlyTreeMap.put(entry.getKey(), entry.getValue());
14                         } else {
15                             break;
16                         }
17                     }
18                 }
19 
20                 if (result.isEmpty()) {
21                     consuming = false;
22                 }
23             } finally {
24                 this.lockTreeMap.writeLock().unlock();
25             }
26         } catch (InterruptedException e) {
27             log.error("take Messages exception", e);
28         }
29 
30         return result;
31     }

takeMessages(final int batchSize)

　　这段代码 msgTreeMap 中获取 batchSzie 数量的消息放入 consumingMsgOrderlyTreeMap 中，并返回给用户消费。由于当前的 MessageQueue 是被 synchronized 锁住的，并且获取的消费消息也是按照消费位点顺序排列的，所以消费时用户能按照物理位点顺序消费消息。

　　如果消费失败，又是怎么保证顺序的呢？消费失败后的处理方法 ConsumeMessageOrderlyService.processConsumeResult() 的实现代码。

　　RocketMQ 支持自动提交 offset 和手动提交 offset 两种方式。以下以自动提交 offset 为例，手动提交 offset 的逻辑与其完全一致。

　　msgs：当前处理的一批消息。

　　status：消费结果的状态。

　　消费成功后，程序会执行 commit() 方法提交当前位点，统计消费成功的 TPS。

　　消费失败后，程序会统计消费失败的 TPS，通过执行 makeMessageToConsumeAgain() 方法删除消费失败的消息，通过定时任务将消费失败的消息在延迟一段时间后，重新提交到消费线程。

　　makeMessageToConsumeAgain()方法将消息 consumingMsgOrderlyTreeMap 中删除，再重新放入本地缓存队列 msgTreeMap 中，等待下次被重新消费。

 1     public void makeMessageToConsumeAgain(List<MessageExt> msgs) {
 2         try {
 3             this.lockTreeMap.writeLock().lockInterruptibly();
 4             try {
 5                 for (MessageExt msg : msgs) {
 6                     this.consumingMsgOrderlyTreeMap.remove(msg.getQueueOffset());
 7                     this.msgTreeMap.put(msg.getQueueOffset(), msg);
 8                 }
 9             } finally {
10                 this.lockTreeMap.writeLock().unlock();
11             }
12         } catch (InterruptedException e) {
13             log.error("makeMessageToCosumeAgain exception", e);
14         }
15     }

makeMessageToConsumeAgain()

　　submitConsumeRequestLater() 方法会执行一个定时任务，延迟一定实践后重新将消息请求发送到消费线程池中，以供下一轮的消费。

 1     private void submitConsumeRequestLater(
 2         final ProcessQueue processQueue,
 3         final MessageQueue messageQueue,
 4         final long suspendTimeMillis
 5     ) {
 6         long timeMillis = suspendTimeMillis;
 7         if (timeMillis == -1) {
 8             timeMillis = this.defaultMQPushConsumer.getSuspendCurrentQueueTimeMillis();
 9         }
10 
11         if (timeMillis < 10) {
12             timeMillis = 10;
13         } else if (timeMillis > 30000) {
14             timeMillis = 30000;
15         }
16 
17         this.scheduledExecutorService.schedule(new Runnable() {
18 
19             @Override
20             public void run() {
21                 ConsumeMessageOrderlyService.this.submitConsumeRequest(null, processQueue, messageQueue, true);
22             }
23         }, timeMillis, TimeUnit.MILLISECONDS);
24     }

submitConsumeRequestLater()

　　做完这两个操作后，我们试想一下，消费线程在下一次消费时会发生什么事情？如果是从 msgTreeMap 中获取一批消息，那么返回的消息又是哪些呢？消息物理位点最小的，也就是之前未成功消息的消息。如果顺序消息消费失败，会再次投递消费者消费，直到消费成功，以此来保证顺序性。

RocketMQ(4.8.0)——消费方式 消费方式

1.1 Pull消费流程

1.2 Push消费流程

相关推荐

RocketMQ(4.8.0)——消费方式消费方式