51. 源代碼解讀-RocketMQ消息重新消費(fèi)

一. 前言

RocketMQ支持消息消費(fèi)失敗后重新消費(fèi)，具體代碼如下：

/*
         *  Register callback to execute on arrival of messages fetched from brokers.
         */
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) {
                System.out.printf(Thread.currentThread().getName()  + "%n");
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });

也就是需要返回ConsumeConcurrentlyStatus.RECONSUME_LATER，那這個(gè)怎么實(shí)現(xiàn)的呢？

二. 代碼流程

以push 非同步消息為例，消息消費(fèi)過(guò)程可以參考https://blog.51cto.com/483181/2056301

我們從客戶端成功獲取到一條消息開(kāi)始，也就是DefaultMQPushConsumerImpl.pullMessage

2.1. 獲取消息回調(diào)

public void pullMessage(final PullRequest pullRequest) {
        ...

        PullCallback pullCallback = new PullCallback() {
            @Override
            public void onSuccess(PullResult pullResult) {
                if (pullResult != null) {
                    pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
                        subscriptionData);

                    switch (pullResult.getPullStatus()) {
                        case FOUND:
                             ...
                                DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
                                    pullResult.getMsgFoundList(),
                                    processQueue,
                                    pullRequest.getMessageQueue(),
                                    dispathToConsume);

                            ....

                            break;

                    }
                }
            }

            @Override
            public void onException(Throwable e) {
                ...
            }
        };

        ...

        try {
            this.pullAPIWrapper.pullKernelImpl(
                pullRequest.getMessageQueue(),
                subExpression,
                subscriptionData.getExpressionType(),
                subscriptionData.getSubVersion(),
                pullRequest.getNextOffset(),
                this.defaultMQPushConsumer.getPullBatchSize(),
                sysFlag,
                commitOffsetValue,
                BROKER_SUSPEND_MAX_TIME_MILLIS,
                CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND,
                CommunicationMode.ASYNC,
                pullCallback
            );
        } catch (Exception e) {

        }
    }

pullKernelImpl獲取消息后，如果是異步請(qǐng)求，那么將會(huì)回調(diào)pullCallback，我們假設(shè)成功拿到消息，也就是FOUND分支。
那么就會(huì)調(diào)用submit 消費(fèi)請(qǐng)求：consumeMessageService.submitConsumeRequest

2.2 submit消息請(qǐng)求

submitConsumeRequest有兩個(gè)實(shí)現(xiàn)類，一個(gè)是pull, 一個(gè)是push。

我們以push為例。

2.3 push submit消息請(qǐng)求

ConsumeMessageConcurrentlyService.java

public void submitConsumeRequest(
        final List<MessageExt> msgs,
        final ProcessQueue processQueue,
        final MessageQueue messageQueue,
        final boolean dispatchToConsume) {
        final int consumeBatchSize = this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
                          ...

                ConsumeRequest consumeRequest = new ConsumeRequest(msgThis, processQueue, messageQueue);
                try {
                    this.consumeExecutor.submit(consumeRequest);
                } catch (RejectedExecutionException e) {

                }
            }
        }
    }

初始化一個(gè)ConsumeRequest Runnable對(duì)象，然后提交到線程池consumeExecutor里面，那么我們繼續(xù)看ConsumeRequest。

2.4 ConsumeRequest

class ConsumeRequest implements Runnable {
        private final List<MessageExt> msgs;
        private final ProcessQueue processQueue;
        private final MessageQueue messageQueue;

        @Override
        public void run() {
            ...

            try {
                ...
                status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
            } catch (Throwable e) {
                ....
            }

                        ....

            if (!processQueue.isDropped()) {
                ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this);
            } else {
                log.warn("processQueue is dropped without process consume result. messageQueue={}, msgs={}", messageQueue, msgs);
            }
        }

    }

首先看try-catch里面，這里面就是回調(diào)客戶端來(lái)消費(fèi)消息。

listener.consumeMessage(Collections.unmodifiableList(msgs), context);

就像我們的消息消費(fèi)重寫(xiě)如下：

/*
         *  Register callback to execute on arrival of messages fetched from brokers.
         */
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) {
                System.out.printf(Thread.currentThread().getName()  + "%n");
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });

然后根據(jù)返回的status調(diào)用processConsumeResult來(lái)處理返回結(jié)果

2.5 processConsumeResult

public void processConsumeResult(
        final ConsumeConcurrentlyStatus status,
        final ConsumeConcurrentlyContext context,
        final ConsumeRequest consumeRequest
    ) {
       ...

        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:
                log.info("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);
                    boolean result = this.sendMessageBack(msg, context);
                    if (!result) {
                        msg.setReconsumeTimes(msg.getReconsumeTimes() + 1);
                        msgBackFailed.add(msg);
                    }
                }

                log.info("msgBackFailed.isEmpty() [{}]", msgBackFailed.isEmpty());
                if (!msgBackFailed.isEmpty()) {
                    log.info("msgBackFailed [{}]", msgBackFailed);
                    consumeRequest.getMsgs().removeAll(msgBackFailed);

                    this.submitConsumeRequestLater(msgBackFailed, consumeRequest.getProcessQueue(), consumeRequest.getMessageQueue());
                }
                break;
            default:
                break;
        }

        ...
    }

2.5.1

首先會(huì)判斷消費(fèi)模式是集群還是廣播模式，如果廣播模式，就日志記錄下不處理了。
如果是集群模式，那么把調(diào)用sendMessageBack發(fā)送消息到broker。等待下一次broker重新消費(fèi)消息。
如果發(fā)送失敗，那么立即就會(huì)消費(fèi)消息。

2.6 發(fā)送消息到Broker

public boolean sendMessageBack(final MessageExt msg, final ConsumeConcurrentlyContext context) {
        int delayLevel = context.getDelayLevelWhenNextConsume();

        try {
            this.defaultMQPushConsumerImpl.sendMessageBack(msg, delayLevel, context.getMessageQueue().getBrokerName());
            return true;
        } catch (Exception e) {
            ...
        }

        return false;
    }

public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName)
        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
        try {
            String brokerAddr = (null != brokerName) ? this.mQClientFactory.findBrokerAddressInPublish(brokerName)
                : RemotingHelper.parseSocketAddressAddr(msg.getStoreHost());
            this.mQClientFactory.getMQClientAPIImpl().consumerSendMessageBack(brokerAddr, msg,
                this.defaultMQPushConsumer.getConsumerGroup(), delayLevel, 5000, getMaxReconsumeTimes());
        } catch (Exception e) {
            ....
        }
    }

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();
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONSUMER_SEND_MSG_BACK, requestHeader);

        requestHeader.setGroup(consumerGroup);
        requestHeader.setOriginTopic(msg.getTopic());
        requestHeader.setOffset(msg.getCommitLogOffset());
        requestHeader.setDelayLevel(delayLevel);
        requestHeader.setOriginMsgId(msg.getMsgId());
        requestHeader.setMaxReconsumeTimes(maxConsumeRetryTimes);

        log.info("addr [{}] request [{}] timeoutMillis [{}]", addr, request, timeoutMillis);
        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());
    }

簡(jiǎn)單的調(diào)用api通過(guò)netty發(fā)到broker而已，請(qǐng)求碼是RequestCode.CONSUMER_SEND_MSG_BACK

public static final int CONSUMER_SEND_MSG_BACK = 36;

Broker會(huì)把消息存儲(chǔ)到文件中，當(dāng)然也會(huì)讓它的reconsume次數(shù)+1

具體可以參考SendMessageProcessor.proce***equest方法，這個(gè)后續(xù)再講