RocketMQ源码解析：消息消费失败后的重试逻辑

RocketMQ源码解析：消息消费失败后的重试逻辑

消息消费失败重新投递的流程

我们接着消息消费的逻辑分析，当消费完成后会调用ConsumeMessageConcurrentlyService#processConsumeResult的方法处理消费的结果，如果消费失败，则会将消息再次发送到这条消息原来存储的broker

为什么要将消息发送到原来存储的broker呢？
因为重试的消息只会消息的基本信息，没有具体的消息体，需要重新从commitLog中获取原来的消息获得消息体

// MQClientAPIImpl#consumerSendMessageBack
ConsumerSendMsgBackRequestHeader requestHeader = new ConsumerSendMsgBackRequestHeader();
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONSUMER_SEND_MSG_BACK, requestHeader);

// 只发送消息的基本信息，等发送到broker重新从commitLog文件取消息
requestHeader.setGroup(consumerGroup);
requestHeader.setOriginTopic(msg.getTopic());
requestHeader.setOffset(msg.getCommitLogOffset());
requestHeader.setDelayLevel(delayLevel);
requestHeader.setOriginMsgId(msg.getMsgId());
requestHeader.setMaxReconsumeTimes(maxConsumeRetryTimes);

// 同步发送重试消息
RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr),
    request, timeoutMillis);

当消息到达broker端的时候，会根据偏移量从commitLog中找到原来的消息，然后构建新的消息，消息体和原来的消息体一样，消息的topic变更为%RETRY% + consumerGroup，并且将消息的delayLevel+1，如果是第一次重试，则直接将delayLevel设置为3，此时这条消息的存储和消费逻辑就会按照定时消息来处理，定时消息的处理逻辑在后一节分析

源码解析

Consumer端处理消费结果

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:
            // 消费成功
            if (ackIndex >= consumeRequest.getMsgs().size()) {
                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;
            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 msgBackFailed = new ArrayList(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);
                }
            }

            // 重发消息失败，5s后再重新消费
            if (!msgBackFailed.isEmpty()) {
                consumeRequest.getMsgs().removeAll(msgBackFailed);

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

    // 不管消息消费成功与否，都会更新消费进度
    // 将消费过的消息从 ProcessQueue 中删除
    long offset = consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs());
    if (offset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {
        // 更新本地进度，后台定时任务会定时将消费进度同步到broker中
        this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), offset, true);
    }
}

从这段代码中可以看到当客户端的消费状态为RECONSUME_LATER时，会发送重试消息，追之前的消息消费逻辑可以发现，当消费者的消费状态为如下三种时会进行重试

1. 返回ConsumeConcurrentlyStatus.RECONSUME_LATER
2. 返回null
3. 主动或被动抛出异常

Consumer端消费重试消息

// DefaultMQPushConsumerImpl#copySubscription
switch (this.defaultMQPushConsumer.getMessageModel()) {
    case BROADCASTING:
        break;
    case CLUSTERING:
        // 订阅重试主题
        final String retryTopic = MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup());
        SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(retryTopic, SubscriptionData.SUB_ALL);
        this.rebalanceImpl.getSubscriptionInner().put(retryTopic, subscriptionData);
        break;
    default:
        break;
}

Broker端存储消息

Broker端存储消息的流程在SendMessageProcessor#asyncConsumerSendMsgBack，挑重要的源码分析一下

// 超过最大消费次数16次，或者 delayLevel < 0
// 则将消息投递到死信队列
if (msgExt.getReconsumeTimes() >= maxReconsumeTimes
    || delayLevel < 0) {
    newTopic = MixAll.getDLQTopic(requestHeader.getGroup());
    queueIdInt = ThreadLocalRandom.current().nextInt(99999999) % DLQ_NUMS_PER_GROUP;

    topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(newTopic,
            DLQ_NUMS_PER_GROUP,
            PermName.PERM_WRITE | PermName.PERM_READ, 0);

    if (null == topicConfig) {
        response.setCode(ResponseCode.SYSTEM_ERROR);
        response.setRemark("topic[" + newTopic + "] not exist");
        return CompletableFuture.completedFuture(response);
    }
    msgExt.setDelayTimeLevel(0);
} else {
    if (0 == delayLevel) {
        delayLevel = 3 + msgExt.getReconsumeTimes();
    }
    msgExt.setDelayTimeLevel(delayLevel);
}

当消息超过最大消费次数16次，或者 delayLevel < 0时，会将消息投递到死信队列中，死信队列的topic名为%DLQ% + consumerGroup

参考博客

[1]https://blog.csdn.net/hsz2568952354/article/details/120422714
好文
[2]https://gitbook.cn/books/5d340810c43fe20aeadc88db/index.html
[3]https://www.baiyp.ren/RocketMQ%E6%B6%88%E8%B4%B9%E8%80%85%E4%BF%9D%E9%9A%9C.html