【RocketMQ】消息的消費_ZenDei技術網路在線

上一講【RocketMQ】消息的拉取消息消費當RocketMQ進行消息消費的時候，是通過ConsumeMessageConcurrentlyService的submitConsumeRequest方法，將消息提交到線程池中進行消費，具體的處理邏輯如下：如果本次消息的個數小於等於批量消費的大小c ...

上一講【RocketMQ】消息的拉取

消息消費

當RocketMQ進行消息消費的時候，是通過ConsumeMessageConcurrentlyService的submitConsumeRequest方法，將消息提交到線程池中進行消費，具體的處理邏輯如下：

如果本次消息的個數小於等於批量消費的大小consumeBatchSize，構建消費請求ConsumeRequest，直接提交到線程池中進行消費即可
如果本次消息的個數大於批量消費的大小consumeBatchSize，說明需要分批進行提交，每次構建consumeBatchSize個消息提交到線程池中進行消費
如果出現拒絕提交的異常，調用submitConsumeRequestLater方法延遲進行提交

RocketMQ消息消費是批量進行的，如果一批消息的個數小於預先設置的批量消費大小，直接構建消費請求將消費任務提交到線程池處理即可，否則需要分批進行提交。

public class ConsumeMessageConcurrentlyService implements ConsumeMessageService {
    @Override
    public void submitConsumeRequest(
        final List<MessageExt> msgs,
        final ProcessQueue processQueue,
        final MessageQueue messageQueue,
        final boolean dispatchToConsume) {
        final int consumeBatchSize = this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
        // 如果消息的個數小於等於批量消費的大小
        if (msgs.size() <= consumeBatchSize) {
            // 構建消費請求
            ConsumeRequest consumeRequest = new ConsumeRequest(msgs, processQueue, messageQueue);
            try {
                // 加入到消費線程池中
                this.consumeExecutor.submit(consumeRequest);
            } catch (RejectedExecutionException e) {
                this.submitConsumeRequestLater(consumeRequest);
            }
        } else {
            // 遍歷消息
            for (int total = 0; total < msgs.size(); ) {
                // 創建消息列表，大小為consumeBatchSize，用於批量提交使用
                List<MessageExt> msgThis = new ArrayList<MessageExt>(consumeBatchSize);
                for (int i = 0; i < consumeBatchSize; i++, total++) {
                    if (total < msgs.size()) {
                        // 加入到消息列表中
                        msgThis.add(msgs.get(total));
                    } else {
                        break;
                    }
                }
                // 創建ConsumeRequest
                ConsumeRequest consumeRequest = new ConsumeRequest(msgThis, processQueue, messageQueue);
                try {
                    // 加入到消費線程池中
                    this.consumeExecutor.submit(consumeRequest);
                } catch (RejectedExecutionException e) {
                    for (; total < msgs.size(); total++) {
                        msgThis.add(msgs.get(total));
                    }
                    // 如果出現拒絕提交異常，延遲進行提交
                    this.submitConsumeRequestLater(consumeRequest);
                }
            }
        }
    }
}

消費任務運行

ConsumeRequest是ConsumeMessageConcurrentlyService的內部類，實現了Runnable介面，在run方法中，對消費任務進行了處理：

判斷消息所屬的處理隊列processQueue是否處於刪除狀態，如果已被刪除，不進行處理
重置消息的重試主題

因為延遲消息的主題在後續處理的時候被設置為SCHEDULE_TOPIC_XXXX，所以這裡需要重置。
如果設置了消息消費鉤子函數，執行executeHookBefore鉤子函數
獲取消息監聽器，調用消息監聽器的consumeMessage進行消息消費，並返回消息的消費結果狀態，狀態有兩種分別為CONSUME_SUCCESS和RECONSUME_LATER

CONSUME_SUCCESS：表示消息消費成功。

RECONSUME_LATER：表示消費失敗，稍後延遲重新進行消費。
獲取消費的時長，判斷是否超時
如果設置了消息消費鉤子函數，執行executeHookAfter鉤子函數
再次判斷消息所屬的處理隊列是否處於刪除狀態，如果不處於刪除狀態，調用processConsumeResult方法處理消費結果

public class ConsumeMessageConcurrentlyService implements ConsumeMessageService {
    class ConsumeRequest implements Runnable {
        private final List<MessageExt> msgs;
        private final ProcessQueue processQueue; // 處理隊列
        private final MessageQueue messageQueue; // 消息隊列
      
        @Override
        public void run() {
            // 如果處理隊列已被刪除
            if (this.processQueue.isDropped()) {
                log.info("the message queue not be able to consume, because it's dropped. group={} {}", ConsumeMessageConcurrentlyService.this.consumerGroup, this.messageQueue);
                return;
            }
            // 獲取消息監聽器
            MessageListenerConcurrently listener = ConsumeMessageConcurrentlyService.this.messageListener;
            ConsumeConcurrentlyContext context = new ConsumeConcurrentlyContext(messageQueue);
            ConsumeConcurrentlyStatus status = null;
            // 重置消息重試主題名稱 
            defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
            ConsumeMessageContext consumeMessageContext = null;
            // 如果設置了鉤子函數
            if (ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.hasHook()) {
                // ...
// 執行鉤子函數            
              ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext);
            }

            long beginTimestamp = System.currentTimeMillis();
            boolean hasException = false;
            ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
            try {
                if (msgs != null && !msgs.isEmpty()) {
                    for (MessageExt msg : msgs) {
                        // 設置消費開始時間戳
                        MessageAccessor.setConsumeStartTimeStamp(msg, String.valueOf(System.currentTimeMillis()));
                    }
                }
                // 通過消息監聽器的consumeMessage進行消息消費，並返回消費結果狀態
                status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
            } catch (Throwable e) {
                log.warn(String.format("consumeMessage exception: %s Group: %s Msgs: %s MQ: %s",
                    RemotingHelper.exceptionSimpleDesc(e),
                    ConsumeMessageConcurrentlyService.this.consumerGroup,
                    msgs,
                    messageQueue), e);
                hasException = true;
            }
            // 計算消費時長
            long consumeRT = System.currentTimeMillis() - beginTimestamp;
            if (null == status) {
                if (hasException) {
                    // 出現異常
                    returnType = ConsumeReturnType.EXCEPTION;
                } else {
                    // 返回NULL
                    returnType = ConsumeReturnType.RETURNNULL;
                }
            } else if (consumeRT >= defaultMQPushConsumer.getConsumeTimeout() * 60 * 1000) { // 判斷超時
                returnType = ConsumeReturnType.TIME_OUT; // 返回類型置為超時
            } else if (ConsumeConcurrentlyStatus.RECONSUME_LATER == status) { // 如果延遲消費
                returnType = ConsumeReturnType.FAILED; // 返回類置為失敗
            } else if (ConsumeConcurrentlyStatus.CONSUME_SUCCESS == status) { // 如果成功狀態
                returnType = ConsumeReturnType.SUCCESS; // 返回類型為成功
            }
            // ...
            // 如果消費狀態為空
            if (null == status) {
                log.warn("consumeMessage return null, Group: {} Msgs: {} MQ: {}",
                    ConsumeMessageConcurrentlyService.this.consumerGroup,
                    msgs,
                    messageQueue);
                // 狀態置為延遲消費
                status = ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            // 如果設置了鉤子函數
            if (ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.hasHook()) {
                consumeMessageContext.setStatus(status.toString());
                consumeMessageContext.setSuccess(ConsumeConcurrentlyStatus.CONSUME_SUCCESS == status);
                // 執行executeHookAfter方法
                ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.executeHookAfter(consumeMessageContext);
            }
            ConsumeMessageConcurrentlyService.this.getConsumerStatsManager()
                .incConsumeRT(ConsumeMessageConcurrentlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);
            if (!processQueue.isDropped()) {
                // 處理消費結果
                ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this);
            } else {
                log.warn("processQueue is dropped without process consume result. messageQueue={}, msgs={}", messageQueue, msgs);
            }
        }
    }
}

// 重置消息重試主題
public class DefaultMQPushConsumerImpl implements MQConsumerInner {
   public void resetRetryAndNamespace(final List<MessageExt> msgs, String consumerGroup) {
        // 獲取消費組的重試主題：%RETRY% + 消費組名稱
        final String groupTopic = MixAll.getRetryTopic(consumerGroup);
        for (MessageExt msg : msgs) {
            // 獲取消息的重試主題名稱
            String retryTopic = msg.getProperty(MessageConst.PROPERTY_RETRY_TOPIC);
            // 如果重試主題不為空並且與消費組的重試主題一致
            if (retryTopic != null && groupTopic.equals(msg.getTopic())) {
                // 設置重試主題
                msg.setTopic(retryTopic);
            }
            if (StringUtils.isNotEmpty(this.defaultMQPushConsumer.getNamespace())) {
                msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQPushConsumer.getNamespace()));
            }
        }
    }
  
}

// 消費結果狀態
public enum ConsumeConcurrentlyStatus {
    /**
     * 消費成功
     */
    CONSUME_SUCCESS,
    /**
     * 消費失敗，延遲進行消費
     */
    RECONSUME_LATER;
}

處理消費結果

一、設置ackIndex

ackIndex的值用來判斷失敗消息的個數，在processConsumeResult方法中根據消費結果狀態進行判斷，對ackIndex的值進行設置，前面可知消費結果狀態有以下兩種：

CONSUME_SUCCESS：消息消費成功，此時ackIndex設置為消息大小 - 1，表示消息都消費成功。
RECONSUME_LATER：消息消費失敗，返回延遲消費狀態，此時ackIndex置為-1，表示消息都消費失敗。

二、處理消費失敗的消息

廣播模式

廣播模式下，如果消息消費失敗，只將失敗的消息列印出來不做其他處理。

集群模式

開啟for迴圈，初始值為i = ackIndex + 1，結束條件為i < consumeRequest.getMsgs().size()，上面可知ackIndex有兩種情況：

消費成功：ackIndex值為消息大小-1，此時ackIndex + 1的值等於消息的個數大小，不滿足for迴圈的執行條件，相當於消息都消費成功，不需要進行失敗的消息處理。
延遲消費：ackIndex值為-1，此時ackIndex+1為0，滿足for迴圈的執行條件，從第一條消息開始遍歷到最後一條消息，調用sendMessageBack方法向Broker發送CONSUMER_SEND_MSG_BACK消息，如果發送成功Broker會根據延遲等級，放入不同的延遲隊列中，到達延遲時間後，消費者將會重新進行拉取，如果發送失敗，加入到失敗消息列表中，稍後重新提交消費任務進行處理。

三、移除消息，更新拉取偏移量

以上步驟處理完畢後，首先調用removeMessage從處理隊列中移除消息並返回拉取消息的偏移量，然後調用updateOffset更新拉取偏移量。

public class ConsumeMessageConcurrentlyService implements ConsumeMessageService {
    public void processConsumeResult(
        final ConsumeConcurrentlyStatus status,
        final ConsumeConcurrentlyContext context,
        final ConsumeRequest consumeRequest
    ) {
        // 獲取ackIndex
        int ackIndex = context.getAckIndex();
        if (consumeRequest.getMsgs().isEmpty())
            return;

        switch (status) {
            case CONSUME_SUCCESS: // 如果消費成功
                // 如果ackIndex大於等於消息的大小
                if (ackIndex >= consumeRequest.getMsgs().size()) {
                    // 設置為消息大小-1
                    ackIndex = consumeRequest.getMsgs().size() - 1;
                }
                // 計算消費成功的的個數
                int ok = ackIndex + 1;
                // 計算消費失敗的個數
                int failed = consumeRequest.getMsgs().size() - ok;
                this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), ok);
                this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), failed);
                break;
            case RECONSUME_LATER: // 如果延遲消費
                // ackIndex置為-1
                ackIndex = -1;
                this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(),
                    consumeRequest.getMsgs().size());
                break;
            default:
                break;
        }
        // 判斷消費模式
        switch (this.defaultMQPushConsumer.getMessageModel()) {
            case BROADCASTING: // 廣播模式
                for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
                    MessageExt msg = consumeRequest.getMsgs().get(i);
                    log.warn("BROADCASTING, the message consume failed, drop it, {}", msg.toString());
                }
                break;
            case CLUSTERING: // 集群模式
                List<MessageExt> msgBackFailed = new ArrayList<MessageExt>(consumeRequest.getMsgs().size());
                // 遍歷消費失敗的消息
                for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
                    // 獲取消息
                    MessageExt msg = consumeRequest.getMsgs().get(i);
                    // 向Broker發送延遲消息
                    boolean result = this.sendMessageBack(msg, context);
                    // 如果發送失敗
                    if (!result) {
                        // 消費次數+1
                        msg.setReconsumeTimes(msg.getReconsumeTimes() + 1);
                        // 加入失敗消息列表中
                        msgBackFailed.add(msg);
                    }
                }
                // 如果不為空
                if (!msgBackFailed.isEmpty()) {
                    consumeRequest.getMsgs().removeAll(msgBackFailed);
                    // 稍後重新進行消費
                    this.submitConsumeRequestLater(msgBackFailed, consumeRequest.getProcessQueue(), consumeRequest.getMessageQueue());
                }
                break;
            default:
                break;
        }
        // 從處理隊列中移除消息
        long offset = consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs());
        if (offset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {
            // 更新拉取偏移量
            this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), offset, true);
        }
    }
}

發送CONSUMER_SEND_MSG_BACK消息

延遲級別

RocketMQ的延遲級別對應的延遲時間常量定義在MessageStoreConfig的messageDelayLevel變數中：

public class MessageStoreConfig {
    private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
}

延遲級別與延遲時間對應關係：

延遲級別0 ---> 對應延遲時間1s，也就是延遲1秒後消費者重新從Broker拉取進行消費

延遲級別1 ---> 延遲時間5s

延遲級別2 ---> 延遲時間10s

...

以此類推，最大的延遲時間為2h

在sendMessageBack方法中，首先從上下文中獲取了延遲級別（ConsumeConcurrentlyContext中可以看到，延遲級別預設為0），並對主題加上Namespace，然後調用defaultMQPushConsumerImpl的sendMessageBack發送消息：

public class ConsumeMessageConcurrentlyService implements ConsumeMessageService {
   public boolean sendMessageBack(final MessageExt msg, final ConsumeConcurrentlyContext context) {
        // 獲取延遲級別
        int delayLevel = context.getDelayLevelWhenNextConsume();
        // 對主題添加上Namespace
        msg.setTopic(this.defaultMQPushConsumer.withNamespace(msg.getTopic()));
        try {
            // 向Broker發送消息
            this.defaultMQPushConsumerImpl.sendMessageBack(msg, delayLevel, context.getMessageQueue().getBrokerName());
            return true;
        } catch (Exception e) {
            log.error("sendMessageBack exception, group: " + this.consumerGroup + " msg: " + msg.toString(), e);
        }
        return false;
    }
}

// 併發消費上下文
public class ConsumeConcurrentlyContext {
    /**
     * -1，不進行重試，加入DLQ隊列
     * 0, Broker控制重試頻率
     * >0, 客戶端控制
     */
    private int delayLevelWhenNextConsume = 0; // 預設為0
}

DefaultMQPushConsumerImp的sendMessageBack方法中又調用了MQClientAPIImpl的consumerSendMessageBack方法進行發送：

public class DefaultMQPushConsumerImpl implements MQConsumerInner {
    public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName)
        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
        try {
            // 獲取Broker地址
            String brokerAddr = (null != brokerName) ? this.mQClientFactory.findBrokerAddressInPublish(brokerName)
                : RemotingHelper.parseSocketAddressAddr(msg.getStoreHost());
            // 調用consumerSendMessageBack方法發送消息
            this.mQClientFactory.getMQClientAPIImpl().consumerSendMessageBack(brokerAddr, msg,
                this.defaultMQPushConsumer.getConsumerGroup(), delayLevel, 5000, getMaxReconsumeTimes());
        } catch (Exception e) {
            // ...
        } finally {
            msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQPushConsumer.getNamespace()));
        }
    }
}

在MQClientAPIImpl的consumerSendMessageBack方法中，可以看到設置的請求類型是CONSUMER_SEND_MSG_BACK，然後設置了消息的相關信息，向Broker發送請求：

public class MQClientAPIImpl {
    public void consumerSendMessageBack(
        final String addr,
        final MessageExt msg,
        final String consumerGroup,
        final int delayLevel,
        final long timeoutMillis,
        final int maxConsumeRetryTimes
    ) throws RemotingException, MQBrokerException, InterruptedException {
        // 創建請求頭
        ConsumerSendMsgBackRequestHeader requestHeader = new ConsumerSendMsgBackRequestHeader();
        // 設置請求類型為CONSUMER_SEND_MSG_BACK
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONSUMER_SEND_MSG_BACK, requestHeader);
        // 設置消費組
        requestHeader.setGroup(consumerGroup);
        requestHeader.setOriginTopic(msg.getTopic());
        // 設置消息物理偏移量
        requestHeader.setOffset(msg.getCommitLogOffset());
        // 設置延遲級別
        requestHeader.setDelayLevel(delayLevel);
        // 設置消息ID
        requestHeader.setOriginMsgId(msg.getMsgId());
        // 設置最大消費次數
        requestHeader.setMaxReconsumeTimes(maxConsumeRetryTimes);
        // 向Broker發送請求
        RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr),
            request, timeoutMillis);
        assert response != null;
        switch (response.getCode()) {
            case ResponseCode.SUCCESS: {
                return;
            }
            default:
                break;
        }
        throw new MQBrokerException(response.getCode(), response.getRemark(), addr);
    }
}

Broker對請求的處理

Broker對CONSUMER_SEND_MSG_BACK類型的請求在SendMessageProcessor中，處理邏輯如下：

根據消費組獲取訂閱信息配置，如果獲取為空，記錄錯誤信息，直接返回
獲取消費組的重試主題，然後從重試隊列中隨機選取一個隊列，並創建TopicConfig主題配置信息
根據消息的物理偏移量從commitlog中獲取消息
判斷消息的消費次數是否大於等於最大消費次數 或者 延遲等級小於0：
- 如果條件滿足，表示需要把消息放入到死信隊列DLQ中，此時設置DLQ隊列ID
- 如果不滿足，判斷延遲級別是否為0，如果為0，使用3 + 消息的消費次數作為新的延遲級別
新建消息MessageExtBrokerInner，設置消息的相關信息，此時相當於生成了一個全新的消息（會設置之前消息的ID），會重新添加到CommitLog中，消息主題的設置有兩種情況：
- 達到了加入DLQ隊列的條件，此時主題為DLQ主題（%DLQ% + 消費組名稱），消息之後會添加到選取的DLQ隊列中
- 未達到DLQ隊列的條件，此時主題為重試主題（%RETRY% + 消費組名稱），之後重新進行消費
調用asyncPutMessage添加消息，詳細過程可參考之前的文章【消息的存儲】

public class SendMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor {
    // 處理請求
    public CompletableFuture<RemotingCommand> asyncProcessRequest(ChannelHandlerContext ctx,
                                                                  RemotingCommand request) throws RemotingCommandException {
        final SendMessageContext mqtraceContext;
        switch (request.getCode()) {
            case RequestCode.CONSUMER_SEND_MSG_BACK:
                // 處理請求
                return this.asyncConsumerSendMsgBack(ctx, request);
            default:
                // ...
        }
    }
  
    private CompletableFuture<RemotingCommand> asyncConsumerSendMsgBack(ChannelHandlerContext ctx,
                                                                        RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        final ConsumerSendMsgBackRequestHeader requestHeader =
                (ConsumerSendMsgBackRequestHeader)request.decodeCommandCustomHeader(ConsumerSendMsgBackRequestHeader.class);
        // ...
        // 根據消費組獲取訂閱信息配置
        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 CompletableFuture.completedFuture(response);
        }
        // ...
    
        // 獲取消費組的重試主題
        String newTopic = MixAll.getRetryTopic(requestHeader.getGroup());
        // 從重試隊列中隨機選取一個隊列
        int queueIdInt = ThreadLocalRandom.current().nextInt(99999999) % subscriptionGroupConfig.getRetryQueueNums();
        int topicSysFlag = 0;
        if (requestHeader.isUnitMode()) {
            topicSysFlag = TopicSysFlag.buildSysFlag(false, true);
        }
        // 創建TopicConfig主題配置信息
        TopicConfig topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(
            newTopic,
            subscriptionGroupConfig.getRetryQueueNums(),
            PermName.PERM_WRITE | PermName.PERM_READ, topicSysFlag);
        //...
    
        // 根據消息物理偏移量從commitLog文件中獲取消息
        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 CompletableFuture.completedFuture(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);
        // 延遲等級獲取
        int delayLevel = requestHeader.getDelayLevel();
        // 獲取最大消費重試次數
        int maxReconsumeTimes = subscriptionGroupConfig.getRetryMaxTimes();
        if (request.getVersion() >= MQVersion.Version.V3_4_9.ordinal()) {
            Integer times = requestHeader.getMaxReconsumeTimes();
            if (times != null) {
                maxReconsumeTimes = times;
            }
        }
        // 判斷消息的消費次數是否大於等於最大消費次數 或者 延遲等級小於0
        if (msgExt.getReconsumeTimes() >= maxReconsumeTimes
            || delayLevel < 0) {
            // 獲取DLQ主題
            newTopic = MixAll.getDLQTopic(requestHeader.getGroup());
            // 選取一個隊列
            queueIdInt = ThreadLocalRandom.current().nextInt(99999999) % DLQ_NUMS_PER_GROUP;
            // 創建DLQ的topicConfig
            topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(newTopic,
                    DLQ_NUMS_PER_GROUP,
                    PermName.PERM_WRITE | PermName.PERM_READ, 0);
            // ...
        } else {
             // 如果延遲級別為0
            if (0 == delayLevel) {
                // 更新延遲級別
                delayLevel = 3 + msgExt.getReconsumeTimes();
            }
            // 設置延遲級別
            msgExt.setDelayTimeLevel(delayLevel);
        }
        // 新建消息
        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); // 設置隊列ID
        msgInner.setSysFlag(msgExt.getSysFlag());
        msgInner.setBornTimestamp(msgExt.getBornTimestamp());
        msgInner.setBornHost(msgExt.getBornHost());
        msgInner.setStoreHost(msgExt.getStoreHost()); 
        msgInner.setReconsumeTimes(msgExt.getReconsumeTimes() + 1);// 設置消費次數
        // 原始的消息ID
        String originMsgId = MessageAccessor.getOriginMessageId(msgExt);
        // 設置消息ID
        MessageAccessor.setOriginMessageId(msgInner, UtilAll.isBlank(originMsgId) ? msgExt.getMsgId() : originMsgId);
        msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));
        // 添加重試消息
        CompletableFuture<PutMessageResult> putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner);
        return putMessageResult.thenApply((r) -> {
            if (r != null) {
                switch (r.getPutMessageStatus()) {
                    case PUT_OK:
                        // ...
                        return response;
                    default:
                        break;
                }
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark(r.getPutMessageStatus().name());
                return response;
            }
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("putMessageResult is null");
            return response;
        });
    }
}

延遲消息處理

由【消息的存儲】文章可知，消息添加會進入到asyncPutMessage方法中，首先獲取了事務類型，如果未使用事務或者是提交事務的情況下，對延遲時間級別進行判斷，如果延遲時間級別大於0，說明消息需要延遲消費，此時做如下處理：

判斷消息的延遲級別是否超過了最大延遲級別，如果超過了就使用最大延遲級別

獲取RMQ_SYS_SCHEDULE_TOPIC，它是在TopicValidator中定義的常量，值為SCHEDULE_TOPIC_XXXX:

public class TopicValidator {
    // ...
    public static final String RMQ_SYS_SCHEDULE_TOPIC = "SCHEDULE_TOPIC_XXXX";
}

根據延遲級別選取對應的隊列，一般會把相同延遲級別的消息放在同一個隊列中
備份之前的TOPIC和隊列ID
更改消息隊列的主題為RMQ_SYS_SCHEDULE_TOPIC，所以延遲消息的主題最終被設置為RMQ_SYS_SCHEDULE_TOPIC，放在對應的延遲隊列中進行處理

public class CommitLog {
    public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
        // ...
        // 獲取事務類型
        final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
        // 如果未使用事務或者提交事務
        if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
                || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
            // 判斷延遲級別
            if (msg.getDelayTimeLevel() > 0) {
                // 如果超過了最大延遲級別
                if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                    msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
                }
                // 獲取RMQ_SYS_SCHEDULE_TOPIC
                topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
                // 根據延遲級別選取對應的隊列
                int queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());

                // 備份之前的TOPIC和隊列ID
                MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
                MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
                msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
                // 設置SCHEDULE_TOPIC
                msg.setTopic(topic);
                msg.setQueueId(queueId);
            }
        }
        // ...
    }
}

拉取進度持久化

RocketMQ消費模式分為廣播模式和集群模式，廣播模式下消費進度保存在每個消費者端，集群模式下消費進度保存在Broker端。

廣播模式

更新進度

LocalFileOffsetStore中使用了一個ConcurrentMap類型的變數offsetTable存儲消息隊列對應的拉取偏移量，KEY為消息隊列，value為該消息隊列對應的拉取偏移量。

在更新拉取進度的時候，從offsetTable中獲取當前消息隊列的拉取偏移量，如果為空，則新建並保存到offsetTable中，否則獲取之前已經保存的偏移量，對值進行更新，需要註意這裡只是更新了offsetTable中的數據，並沒有持久化到磁碟，持久化的操作在persistAll方法中：

public class LocalFileOffsetStore implements OffsetStore {
    // offsetTable：KEY為消息隊列，value為該消息隊列的拉取偏移量
    private ConcurrentMap<MessageQueue, AtomicLong> offsetTable =
        new ConcurrentHashMap<MessageQueue, AtomicLong>();
  
    @Override
    public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) {
        if (mq != null) {
            // 獲取之前的拉取進度
            AtomicLong offsetOld = this.offsetTable.get(mq);
            if (null == offsetOld) {
                // 如果之前不存在，進行創建
                offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset));
            }
            // 如果不為空
            if (null != offsetOld) {
                if (increaseOnly) {
                    MixAll.compareAndIncreaseOnly(offsetOld, offset);
                } else {
                    // 更新拉取偏移量
                    offsetOld.set(offset);
                }
            }
        }
    }
}

載入進度

由於廣播模式下消費進度保存在消費者端，所以需要從本地磁碟載入之前保存的消費進度文件。

LOCAL_OFFSET_STORE_DIR：消費進度文件所在的根路徑

public final static String LOCAL_OFFSET_STORE_DIR = System.getProperty(
        "rocketmq.client.localOffsetStoreDir", System.getProperty("user.home") + File.separator + ".rocketmq_offsets");

在LocalFileOffsetStore的構造函數中可以看到，對拉取偏移量的保存文件路徑進行了設置，為LOCAL_OFFSET_STORE_DIR + 客戶端ID + 消費組名稱 + offsets.json，從名字上看，消費進度的數據格式是以JSON的形式進行保存的：

this.storePath = LOCAL_OFFSET_STORE_DIR + File.separator + this.mQClientFactory.getClientId() + File.separator +
            this.groupName + File.separator + "offsets.json";

在load方法中，首先從本地讀取 offsets.json文件，並序列化為OffsetSerializeWrapper對象，然後將保存的消費進度加入到offsetTable中：

 public class LocalFileOffsetStore implements OffsetStore {
   
    // 文件路徑
    public final static String LOCAL_OFFSET_STORE_DIR = System.getProperty(
        "rocketmq.client.localOffsetStoreDir",
        System.getProperty("user.home") + File.separator + ".rocketmq_offsets");
    private final String storePath;
    // ...
   
    public LocalFileOffsetStore(MQClientInstance mQClientFactory, String groupName) {
        this.mQClientFactory = mQClientFactory;
        this.groupName = groupName;
        // 設置拉取進度文件的路徑
        this.storePath = LOCAL_OFFSET_STORE_DIR + File.separator +
            this.mQClientFactory.getClientId() + File.separator +
            this.groupName + File.separator +
            "offsets.json";
    }
    @Override
    public void load() throws MQClientException {
        // 從本地讀取拉取偏移量
        OffsetSerializeWrapper offsetSerializeWrapper = this.readLocalOffset();
        if (offsetSerializeWrapper != null && offsetSerializeWrapper.getOffsetTable() != null) {
            // 加入到offsetTable中
            offsetTable.putAll(offsetSerializeWrapper.getOffsetTable());

            for (Entry<MessageQueue, AtomicLong> mqEntry : offsetSerializeWrapper.getOffsetTable().entrySet()) {
                AtomicLong offset = mqEntry.getValue();
                log.info("load consumer's offset, {} {} {}",
                        this.groupName,
                        mqEntry.getKey(),
                        offset.get());
            }
        }
    }
  
    // 從本地載入文件
    private OffsetSerializeWrapper readLocalOffset() throws MQClientException {
        String content = null;
        try {
            // 讀取文件
            content = MixAll.file2String(this.storePath);
        } catch (IOException e) {
            log.warn("Load local offset store file exception", e);
        }
        if (null == content || content.length() == 0) {
            return this.readLocalOffsetBak();
        } else {
            OffsetSerializeWrapper offsetSerializeWrapper = null;
            try {
                // 序列化
                offsetSerializeWrapper =
                    OffsetSerializeWrapper.fromJson(content, OffsetSerializeWrapper.class);
            } catch (Exception e) {
                log.warn("readLocalOffset Exception, and try to correct", e);
                return this.readLocalOffsetBak();
            }

            return offsetSerializeWrapper;
        }
    }
}

OffsetSerializeWrapper

OffsetSerializeWrapper中同樣使用了ConcurrentMap，從磁碟的offsets.json文件中讀取數據後，將JSON轉為OffsetSerializeWrapper對象，就可以通過OffsetSerializeWrapper的offsetTable獲取到之前保存的每個消息隊列的消費進度，然後加入到LocalFileOffsetStore的offsetTable中：

public class OffsetSerializeWrapper extends RemotingSerializable {
    private ConcurrentMap<MessageQueue, AtomicLong> offsetTable =
        new ConcurrentHashMap<MessageQueue, AtomicLong>();

    public ConcurrentMap<MessageQueue, AtomicLong> getOffsetTable() {
        return offsetTable;
    }

    public void setOffsetTable(ConcurrentMap<MessageQueue, AtomicLong> offsetTable) {
        this.offsetTable = offsetTable;
    }
}

持久化進度

updateOffset更新只是將記憶體中的數據進行了更改，並未保存到磁碟中，持久化的操作是在persistAll方法中實現的：

創建OffsetSerializeWrapper對象
遍歷LocalFileOffsetStore的offsetTable，將數據加入到OffsetSerializeWrapper的OffsetTable中
將OffsetSerializeWrapper轉為JSON
調用string2File方法將JSON數據保存到磁碟文件

 public class LocalFileOffsetStore implements OffsetStore {
    @Override
    public void persistAll(Set<MessageQueue> mqs) {
        if (null == mqs || mqs.isEmpty())
            return;OffsetSerializeWrapper
        // 創建
        OffsetSerializeWrapper offsetSerializeWrapper = new OffsetSerializeWrapper();
        // 遍歷offsetTable
        for (Map.Entry<MessageQueue, AtomicLong> entry : this.offsetTable.entrySet()) {
            if (mqs.contains(entry.getKey())) {
                // 獲取拉取偏移量
                AtomicLong offset = entry.getValue();
                // 加入到OffsetSerializeWrapper的OffsetTable中
                offsetSerializeWrapper.getOffsetTable().put(entry.getKey(), offset);
            }
        }
        // 將對象轉為JSON
        String jsonString = offsetSerializeWrapper.toJson(true);
        if (jsonString != null) {
            try {
                // 將JSON數據保存到磁碟文件
                MixAll.string2File(jsonString, this.storePath);
            } catch (IOException e) {
                log.error("persistAll consumer offset Exception, " + this.storePath, e);
            }
        }
    }
}

集群模式

集群模式下消費進度保存在Broker端。

更新進度

集群模式下的更新進度與廣播模式下的更新類型，都是只更新了offsetTable中的數據：

public class RemoteBrokerOffsetStore implements OffsetStore {
    
    private ConcurrentMap<MessageQueue, AtomicLong> offsetTable =
        new ConcurrentHashMap<MessageQueue, AtomicLong>();
    @Override
    public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) {
        if (mq != null) {
            // 獲取消息隊列的進度
            AtomicLong offsetOld = this.offsetTable.get(mq);
            if (null == offsetOld) {
                // 將消費進度保存在offsetTable中
                offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset));
            }
            if (null != offsetOld) {
                if (increaseOnly) {
                    MixAll.compareAndIncreaseOnly(offsetOld, offset);
                } else {
                    // 更新拉取偏移量
                    offsetOld.set(offset);
                }
            }
        }
    }
}

載入

集群模式下載入消費進度需要從Broker獲取，在消費者發送消息拉取請求的時候，Broker會計算消費偏移量，所以RemoteBrokerOffsetStore的load方法為空，什麼也沒有乾：

public class RemoteBrokerOffsetStore implements OffsetStore {
    @Override
    public void load() {
    }
}

持久化

由於集群模式下消費進度保存在Broker端，所以persistAll方法中調用了updateConsumeOffsetToBroker向Broker發送請求進行消費進度保存：

public class RemoteBrokerOffsetStore implements OffsetStore {
    @Override
    public void persistAll(Set<MessageQueue> mqs) {
        if (null == mqs || mqs.isEmpty())
            return;

        final HashSet<MessageQueue> unusedMQ = new HashSet<MessageQueue>();

        for (Map.Entry<MessageQueue, AtomicLong> entry : this.offsetTable.entrySet()) {
            MessageQueue mq = entry.getKey();
            AtomicLong offset = entry.getValue();
            if (offset != null) {
                if (mqs.contains(mq)) {
                    try {
                        // 向Broker發送請求更新拉取偏移量
                        this.updateConsumeOffsetToBroker(mq, offset.get());
                        log.info("[persistAll] Group: {} ClientId: {} updateConsumeOffsetToBroker {} {}",
                            this.groupName,
                            this.mQClientFactory.getClientId(),
                            mq,
                            offset.get());
                    } catch (Exception e) {
                        log.error("updateConsumeOffsetToBroker exception, " + mq.toString(), e);
                    }
                } else {
                    unusedMQ.add(mq);
                }
            }
        }
        // ...
    }
}

持久化的觸發

MQClientInstance在啟動定時任務的方法startScheduledTask中註冊了定時任務，定時調用persistAllConsumerOffset對拉取進度進行持久化，persistAllConsumerOffset中又調用了MQConsumerInner的persistConsumerOffset方法：

public class MQClientInstance {
    private void startScheduledTask() {
        // ...
        // 註冊定時任務，定時持久化拉取進度
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    // 持久化
                    MQClientInstance.this.persistAllConsumerOffset();
                } catch (Exception e) {
                    log.error("ScheduledTask persistAllConsumerOffset exception", e);
                }
            }
        }, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
        // ...
    }
    
    private void persistAllConsumerOffset() {
        Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<String, MQConsumerInner> entry = it.next();
            MQConsumerInner impl = entry.getValue();
            // 調用persistConsumerOffset進行持久化
            impl.persistConsumerOffset();
        }
    }
}

DefaultMQPushConsumerImpl是MQConsumerInner的一個子類，以它為例可以看到在persistConsumerOffset方法中調用了offsetStore的persistAll方法進行持久化：

public class DefaultMQPushConsumerImpl implements MQConsumerInner {
    @Override
    public void persistConsumerOffset() {
        try {
            this.makeSureStateOK();
            Set<MessageQueue> mqs = new HashSet<MessageQueue>();
            Set<MessageQueue> allocateMq = this.rebalanceImpl.getProcessQueueTable().keySet();
            mqs.addAll(allocateMq);
            // 拉取進度持久化
            this.offsetStore.persistAll(mqs);
        } catch (Exception e) {
            log.error("group: " + this.defaultMQPushConsumer.getConsumerGroup() + " persistConsumerOffset exception", e);
        }
    }
}

總結

參考
丁威、周繼鋒《RocketMQ技術內幕》

RocketMQ版本：4.9.3