當(dāng)前位置：首頁 > 编程资源 > 综合教程 >内容正文

综合教程

RocketMQ(4.8.0)——消费方式

發(fā)布時(shí)間：2023/12/31 综合教程 29 生活家

生活随笔收集整理的這篇文章主要介紹了 RocketMQ(4.8.0)——消费方式小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

消費(fèi)方式

　　RocketMQ 包含2種消費(fèi)方式：

Pull
Push

　　Pull方式：用戶主動(dòng)Pull消息，自主管理位點(diǎn)。默認(rèn)的 Push 消費(fèi)者實(shí)現(xiàn)類：D:ocketmq-masterclientsrcmainjavaorgapacheocketmqclientconsumerDefaultMQPullConsumer.java

　　　　優(yōu)點(diǎn)：可以靈活地掌控消費(fèi)進(jìn)度和消費(fèi)速度，適合流計(jì)算、消費(fèi)特別耗時(shí)等特殊的消費(fèi)場(chǎng)景。

　　　　缺點(diǎn)：需要從代碼層面精準(zhǔn)地控制消費(fèi)，對(duì)開發(fā)人員有一定要求。

　　Push方式：用戶主動(dòng)Pull消息，自主管理位點(diǎn)。默認(rèn)的 Push 消費(fèi)者實(shí)現(xiàn)類：D:ocketmq-masterclientsrcmainjavaorgapacheocketmqclientconsumerDefaultMQPushConsumer.java

　　　　優(yōu)點(diǎn)：代碼接入非常簡(jiǎn)單，適合大部分業(yè)務(wù)場(chǎng)景。

　　　　缺點(diǎn)：靈活度差，在了解其消費(fèi)原理后，排查消費(fèi)問題方可簡(jiǎn)單快捷。

　　針對(duì)Pull和Push，下面對(duì)兩種方式進(jìn)行簡(jiǎn)單的比較。

消費(fèi)方式/對(duì)比項(xiàng)	Pull	Push	備注
是否需要主動(dòng)拉取	理解分區(qū)后，需要主動(dòng)拉取各個(gè)分區(qū)消息	自動(dòng)	Pull 消息靈活，Push使用更簡(jiǎn)單
位點(diǎn)管理	用戶自行管理或者主動(dòng)提交給 Broker 管理	Broker 管理	Pull 自主管理位點(diǎn)，消費(fèi)靈活； Push 位點(diǎn)交由 Broker 管理
Topic 路由變更是否影響消費(fèi)	否	否	Pull 模式需要編碼實(shí)現(xiàn)路由感知； Push 模式自動(dòng)換行 Rebalance，以適應(yīng)路由變更。

1.1 Pull消費(fèi)流程

　　D:ocketmq-masterclientsrcmainjavaorgapacheocketmqclientconsumerDefaultMQPullConsumer.java 消費(fèi)過程如下：

Pull消費(fèi)的具體步驟：

　　第一步：fetchSubscribeMessageQueue(String Topic)。拉取全部可以消費(fèi)的 Queue。如果某一個(gè) Broker 下線，這里也可以實(shí)時(shí)感知到。

　　第二步：遍歷全部Queue，拉取每個(gè) Queue 可以消費(fèi)的消息。

　　第三步：如果拉取到消息，則執(zhí)行用戶編寫的消費(fèi)代碼。

　　第四步：保存消費(fèi)進(jìn)度。消費(fèi)進(jìn)度可以執(zhí)行 updateConsumeOffset()方法，將消費(fèi)位點(diǎn)上報(bào)給 Broker，也可以自行保存消費(fèi)位點(diǎn)。比如流計(jì)算平臺(tái)Flink使用Pull方式拉取消息消費(fèi)，通過 Checkpoint 管理消費(fèi)進(jìn)度。

1.2 Push消費(fèi)流程

Push消費(fèi)過程如下：

　　第一步：初始化 Push 消費(fèi)者實(shí)例。業(yè)務(wù)代碼初始化 DefaultMQPushConsumer 實(shí)例，啟動(dòng) Pull 服務(wù) PullMessageService。該服務(wù)是一個(gè)線程服務(wù)，不斷執(zhí)行 run() 方法拉取已經(jīng)訂閱 Topic 的全部隊(duì)列消息，將消息保存在本地的緩存隊(duì)列中。

　　第二步：消費(fèi)消息。由消息服務(wù) ConsumeMessageConcurrentlyService 或者 ConsumeMessageOrderlyService 將本地緩存隊(duì)列中的消息不斷放入消費(fèi)線程池，異步回調(diào)業(yè)務(wù)消費(fèi)代碼，此時(shí)業(yè)務(wù)代碼可以消費(fèi)消息。(核心知識(shí)點(diǎn))

　　第三步：保存消費(fèi)進(jìn)度。業(yè)務(wù)代碼消費(fèi)后，將消費(fèi)結(jié)果返回給消費(fèi)服務(wù)，再由消費(fèi)服務(wù)將消費(fèi)進(jìn)度保存在本地，由消費(fèi)進(jìn)度管理服務(wù)定時(shí)和不定時(shí)地持久化到本地（LocalFileOffsetStore 支持）或者遠(yuǎn)程 Broker（RemoteBrokerOffsetStore支持）種。對(duì)于消費(fèi)失敗的消息，RocketMQ 客戶端處理后發(fā)回給 Broker，并告知消費(fèi)失敗。(核心知識(shí)點(diǎn))

Push消費(fèi)者如何拉取消息消費(fèi)：

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

　　第二步：消費(fèi)者拉取消息并消費(fèi)。

　　　　2.1 基本校驗(yàn)。校驗(yàn) ProcessQueue 是否dropped;校驗(yàn)消費(fèi)者服務(wù)狀態(tài)是否正常;校驗(yàn)消費(fèi)者是否被掛起。

　　　　2.2 拉取條數(shù)、字節(jié)數(shù)限制檢查。如果本地緩存消息數(shù)量大于配置的最大拉取條數(shù)（默認(rèn)為 1000，可以調(diào)整），則延遲一段時(shí)間再拉取；如果本地緩存消息字節(jié)數(shù)大于配置的最大緩存字節(jié)數(shù)，則延遲一段時(shí)間再拉取。這兩種校驗(yàn)方式都相當(dāng)于本地流控。

　　　　2.3 并發(fā)消費(fèi)和順序消費(fèi)校驗(yàn)。

　　　　　　在并發(fā)消費(fèi)時(shí)，processQueue.getMaxSpan()方法是用于計(jì)算本地緩存隊(duì)列中第一個(gè)消息和最后一個(gè)消息的 offset 差值。

　　　　　　本地緩存隊(duì)列的 Span 如果大于配置的最大差值（默認(rèn)為2000，可以調(diào)整），則認(rèn)為本地消費(fèi)過慢，需要執(zhí)行本地流控。

　　　　　　順序消費(fèi)時(shí)，如果當(dāng)前拉取的隊(duì)列在 Broker 端沒有被鎖定，說明已經(jīng)有拉取正在執(zhí)行，當(dāng)前拉取請(qǐng)求晚點(diǎn)執(zhí)行；如果不是第一次拉取，需要先計(jì)算最新的拉取位點(diǎn)并修正本地最新的待拉取位點(diǎn)信息，再執(zhí)行拉取。代碼路徑：D:ocketmq-masterclientsrcmainjavaorgapacheocketmqclientimplconsumerDefaultMQPushConsumerImpl.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) 訂閱關(guān)系校驗(yàn)。如果待拉取的 Topic 在本地緩存中訂閱關(guān)系為空，則本地拉取不執(zhí)行，待下一個(gè)正常心跳或者 Rebalance 后訂閱關(guān)系恢復(fù)正常，方可正常拉取。

　　(2) 封裝拉取請(qǐng)求和拉取后的回調(diào)對(duì)象 PullCallback。這里主要將消息拉取請(qǐng)求和拉取結(jié)果處理封裝成 PullCallback，并通過調(diào)用 PullAPIWrapper.pullKernelImpl() 方法拉取請(qǐng)求發(fā)出。

　　拉取結(jié)果存在多種可能性。這里以拉取消息的情況舉例說下：

 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

　　如果拉取到消息，那么將消息保存到對(duì)應(yīng)的本地緩存隊(duì)列 ProcessQueue 中，然后將這些消息提交給 ConsumeMessageService 服務(wù)。

　　ConsumeMessageService 是一個(gè)通用消費(fèi)服務(wù)接口，它包含兩個(gè)實(shí)現(xiàn)類：orgapacheocketmqclientimplconsumerConsumeMessageConcurrentlyService 和srcmainjavaorgapacheocketmqclientimplconsumerConsumeMessageOrderlyService，這兩個(gè)類分別用于并發(fā)消費(fèi)和順序消費(fèi)。

 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()方法分別在啟動(dòng)和關(guān)閉服務(wù)時(shí)使用。

　　updateCorePoolSize()：更新消費(fèi)線程池的核心線程數(shù)。
　　incCorePoolSize()：增加一個(gè)消費(fèi)線程池的核心線程數(shù)。

　　decCorePoolSize()：減少一個(gè)消費(fèi)線程池的核心線程數(shù)。

　　getCorePoolSize()：獲取消費(fèi)線程池的核心線程數(shù)。

　　consumeMessageDirectly()：如果一個(gè)消息已經(jīng)被消費(fèi)過了，但是還想再消費(fèi)一次，就需要實(shí)現(xiàn)這個(gè)方法。
　　submitConsumeRequest()：將消息封裝成線程池任務(wù)，提交給消費(fèi)服務(wù)，消費(fèi)服務(wù)再將消息傳遞給業(yè)務(wù)消費(fèi)進(jìn)行處理。

　　(1) ConsumeMessageService 消息消費(fèi)分發(fā)。ConsumeMessageService 服務(wù)通過 DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest 接口接收消息消費(fèi)任務(wù)后，將消息按照固定條數(shù)封裝成多個(gè) ConsumeRequest 任務(wù)對(duì)象，并發(fā)送到消費(fèi)線程，等待分發(fā)給業(yè)務(wù)消費(fèi)；ConsumeMessageOrderlyService 先將 Pull 的全部消息放在另外一個(gè)本地隊(duì)列中，然后提交一個(gè) ConsumeRequest 到消費(fèi)線程池。

　　(2) 消費(fèi)消息。消費(fèi)的主要邏輯在 ConsumeMessageService 接口的兩個(gè)實(shí)現(xiàn)類中。下面以并發(fā)消息實(shí)現(xiàn)類 orgapacheocketmqclientimplconsumerConsumeMessageConcurrentlyService，代碼如下：

 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             //消費(fèi)前
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                 //預(yù)處理重試隊(duì)列消息
31                 if (msgs != null && !msgs.isEmpty()) {
32                     for (MessageExt msg : msgs) {
33                         MessageAccessor.setConsumeStartTimeStamp(msg, String.valueOf(System.currentTimeMillis()));
34                     }
35                 }
36                 //消費(fèi)回調(diào)
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             //消費(fèi)執(zhí)行后
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                 //處理消費(fèi)結(jié)果
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

　　消費(fèi)消息主要分為消費(fèi)前預(yù)處理、消費(fèi)回調(diào)、消費(fèi)結(jié)果統(tǒng)計(jì)、消費(fèi)結(jié)果處理 4 個(gè)步驟。

　　第一步：消費(fèi)執(zhí)行前進(jìn)行預(yù)處理。執(zhí)行消費(fèi)前的 hook 和重試消息預(yù)處理。消費(fèi)前的 hook 可以理解為消費(fèi)前的消息預(yù)處理（比如消息格式校驗(yàn)）。如果拉取的消息來自重試隊(duì)列，則將 Topic 名重置為原來的 Topic 名，而不用重試 Topic 名。

　　第二步：消費(fèi)回調(diào)。首先設(shè)置消息開始消費(fèi)時(shí)間為當(dāng)前時(shí)間，再將消息列表轉(zhuǎn)為不可修改的 List，然后通過listener.consumeMessage(Collections.unmodifiableList(msgs), context) 方法將消息傳遞給用戶編寫的業(yè)務(wù)消費(fèi)代碼進(jìn)行處理。

　　第三步：消費(fèi)結(jié)果統(tǒng)計(jì)和執(zhí)行消費(fèi)后的 hook。客戶端原生支持基本消費(fèi)指標(biāo)統(tǒng)計(jì)，比如消費(fèi)耗時(shí)；消費(fèi)后的 hook 和消費(fèi)前的 hook 要一一對(duì)應(yīng)，用戶可以用消費(fèi)后的 hook 統(tǒng)計(jì)與自身業(yè)務(wù)相關(guān)的指標(biāo)。

　　第四步：消費(fèi)結(jié)果處理。包含消費(fèi)指標(biāo)統(tǒng)計(jì)、消費(fèi)重試處理和消費(fèi)位點(diǎn)處理。消費(fèi)指標(biāo)主要是對(duì)消費(fèi)成功和失敗的 TPS 的統(tǒng)計(jì)；消費(fèi)重試處理主要將消費(fèi)重試次數(shù)加 1；消費(fèi)位點(diǎn)處理主要根據(jù)消費(fèi)結(jié)構(gòu)更新消費(fèi)位點(diǎn)記錄。

　　至此，Push 消費(fèi)流程完畢。

　　RocketMQ 是一個(gè)消息隊(duì)列，F(xiàn)IFO(Fist In First Out，先進(jìn)先出)規(guī)則如何在消費(fèi)失敗時(shí)保證消息的順序執(zhí)行呢？

　　從消費(fèi)任務(wù)實(shí)現(xiàn)類 ConsumeRequest 和本地緩存隊(duì)列 ProcessQueue 的涉及來看主要差異。并發(fā)消息（無序消費(fèi)）的消費(fèi)請(qǐng)求對(duì)象實(shí)現(xiàn)類代碼路徑：D:ocketmq-masterclientsrcmainjavaorgapacheocketmqclientimplconsumerConsumeMessageConcurrentlyService.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             //消費(fèi)前
 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                 //預(yù)處理重試隊(duì)列消息
 50                 if (msgs != null && !msgs.isEmpty()) {
 51                     for (MessageExt msg : msgs) {
 52                         MessageAccessor.setConsumeStartTimeStamp(msg, String.valueOf(System.currentTimeMillis()));
 53                     }
 54                 }
 55                 //消費(fèi)回調(diào)
 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             //消費(fèi)執(zhí)行后
 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                 //處理消費(fèi)結(jié)果
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()

　　順序消費(fèi)的消費(fèi)請(qǐng)求對(duì)象實(shí)現(xiàn)類為D:ocketmq-masterclientsrcmainjavaorgapacheocketmqclientimplconsumerConsumeMessageConcurrentlyService.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 中并沒有保存需要消費(fèi)的消息，在順序消費(fèi)時(shí)通過調(diào)用 ProcessQueue.takeMessage() 方法獲取需要消費(fèi)的消息，而且消費(fèi)也是同步進(jìn)行的。

　　msgTreeMap：是一個(gè) TreeMap<Long，MessageExt>類型，key是消息物理位點(diǎn)值，value 是消息對(duì)象，該容器是 ProcessQueue 用來緩存本地順序消息的，保存的數(shù)據(jù)是按照key（就是物理位點(diǎn)值）順序排列的。

　　consumingMsgOrderlyTreeMap：是一個(gè)TreeMap<Long，MessageExt>類型，key是消息物理位點(diǎn)值，value 是消息對(duì)象，保存當(dāng)前正在處理的順序消息集合，是 msgTreeMap 的一個(gè)子集。保存的數(shù)據(jù)是按照 key（就是物理機(jī)位點(diǎn)值）順序排列的。

　　batchSize：一次從本地緩存中獲取多少條消息回調(diào)給用戶消費(fèi)。順序消費(fèi)是如何通過 PorcessQueue.takeMessage() 獲取消息給業(yè)務(wù)代碼消費(fèi)的呢？

 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 數(shù)量的消息放入 consumingMsgOrderlyTreeMap 中，并返回給用戶消費(fèi)。由于當(dāng)前的 MessageQueue 是被 synchronized 鎖住的，并且獲取的消費(fèi)消息也是按照消費(fèi)位點(diǎn)順序排列的，所以消費(fèi)時(shí)用戶能按照物理位點(diǎn)順序消費(fèi)消息。

　　如果消費(fèi)失敗，又是怎么保證順序的呢？消費(fèi)失敗后的處理方法 ConsumeMessageOrderlyService.processConsumeResult() 的實(shí)現(xiàn)代碼。

　　RocketMQ 支持自動(dòng)提交 offset 和手動(dòng)提交 offset 兩種方式。以下以自動(dòng)提交 offset 為例，手動(dòng)提交 offset 的邏輯與其完全一致。

　　msgs：當(dāng)前處理的一批消息。

　　status：消費(fèi)結(jié)果的狀態(tài)。

　　消費(fèi)成功后，程序會(huì)執(zhí)行 commit() 方法提交當(dāng)前位點(diǎn)，統(tǒng)計(jì)消費(fèi)成功的 TPS。

　　消費(fèi)失敗后，程序會(huì)統(tǒng)計(jì)消費(fèi)失敗的 TPS，通過執(zhí)行 makeMessageToConsumeAgain() 方法刪除消費(fèi)失敗的消息，通過定時(shí)任務(wù)將消費(fèi)失敗的消息在延遲一段時(shí)間后，重新提交到消費(fèi)線程。

　　makeMessageToConsumeAgain()方法將消息 consumingMsgOrderlyTreeMap 中刪除，再重新放入本地緩存隊(duì)列 msgTreeMap 中，等待下次被重新消費(fèi)。

 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() 方法會(huì)執(zhí)行一個(gè)定時(shí)任務(wù)，延遲一定實(shí)踐后重新將消息請(qǐng)求發(fā)送到消費(fèi)線程池中，以供下一輪的消費(fèi)。

 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()

　　做完這兩個(gè)操作后，我們?cè)囅胍幌?，消費(fèi)線程在下一次消費(fèi)時(shí)會(huì)發(fā)生什么事情？如果是從 msgTreeMap 中獲取一批消息，那么返回的消息又是哪些呢？消息物理位點(diǎn)最小的，也就是之前未成功消息的消息。如果順序消息消費(fèi)失敗，會(huì)再次投遞消費(fèi)者消費(fèi)，直到消費(fèi)成功，以此來保證順序性。

總結(jié)

以上是生活随笔為你收集整理的RocketMQ(4.8.0)——消费方式的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇： Sicily 1346. 金明的预算方案
下一篇：关于 iPhone 的 10 条小知识