51. 源代码解读-RocketMQ消息重新消费

系统管理员 2021-09-19 06:18 439阅读 0赞

## 一. 前言 ##

RocketMQ支持消息消费失败后重新消费，具体代码如下：

/* * 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，那这个怎么实现的呢？

## 二. 代码流程 ##

以push 非同步消息为例，消息消费过程可以参考http://blog.51cto.com/483181/2056301

我们从客户端成功获取到一条消息开始，也就是DefaultMQPushConsumerImpl.pullMessage

### 2.1. 获取消息回调 ###

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获取消息后，如果是异步请求，那么将会回调pullCallback，我们假设成功拿到消息，也就是FOUND分支。  
那么就会调用submit 消费请求：consumeMessageService.submitConsumeRequest

### 2.2 submit消息请求 ###

submitConsumeRequest有两个实现类，一个是pull, 一个是push。

![51. 源代码解读-RocketMQ消息重新消费][51. _-RocketMQ]

我们以push为例。

### 2.3 push submit消息请求 ###

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) {
    
                    }
                }
            }
        }

初始化一个ConsumeRequest Runnable对象，然后提交到线程池consumeExecutor里面，那么我们继续看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里面，这里面就是回调客户端来消费消息。

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

就像我们的消息消费重写如下：

然后根据返回的status调用processConsumeResult来处理返回结果

### 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 ####

首先会判断消费模式是集群还是广播模式，如果广播模式，就日志记录下不处理了。  
如果是集群模式，那么把调用sendMessageBack发送消息到broker。等待下一次broker重新消费消息。  
如果发送失败，那么立即就会消费消息。

### 2.6 发送消息到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());
        }

简单的调用api通过netty发到broker而已，请求码是RequestCode.CONSUMER\_SEND\_MSG\_BACK

public static final int CONSUMER_SEND_MSG_BACK = 36;

Broker会把消息存储到文件中，当然也会让它的reconsume次数+1

具体可以参考SendMessageProcessor.processRequest方法，这个后续再讲

![51. 源代码解读-RocketMQ消息重新消费][51. _-RocketMQ 1]

本文转自rongwei84n 51CTO博客，原文链接：http://blog.51cto.com/483181/2058135，如需转载请自行联系原作者

[51. _-RocketMQ]: /images/20210726/060bf271287345d08b4aa76936c3c17a.png
[51. _-RocketMQ 1]: /images/20210726/e54a9f69acb841d286cab001fb136fa5.png