RocketMQ源码解析九（消息消费失败重试流程）

我不是女神ヾ 2022-09-12 07:47 28阅读 0赞

> **RocketMQ版本4.6.0，记录自己看源码的过程**

## Consumer ##

直接从ConsumeMessageConcurrentlyService发送重试消息方法开始

public boolean sendMessageBack(final MessageExt msg, final ConsumeConcurrentlyContext context) { 
        // 默认0
        int delayLevel = context.getDelayLevelWhenNextConsume();
    
        // Wrap topic with namespace before sending back message.
        msg.setTopic(this.defaultMQPushConsumer.withNamespace(msg.getTopic()));
        try { 
            // 发送重试消息
            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;
    }

/** * 向broker发送重试消息 */
    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());
            // 发送重试消息
            this.mQClientFactory.getMQClientAPIImpl().consumerSendMessageBack(brokerAddr, msg,
                this.defaultMQPushConsumer.getConsumerGroup(), delayLevel, 5000, getMaxReconsumeTimes());
        } catch (Exception e) { 
            log.error("sendMessageBack Exception, " + this.defaultMQPushConsumer.getConsumerGroup(), e);
    
            Message newMsg = new Message(MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup()), msg.getBody());
    
            String originMsgId = MessageAccessor.getOriginMessageId(msg);
            MessageAccessor.setOriginMessageId(newMsg, UtilAll.isBlank(originMsgId) ? msg.getMsgId() : originMsgId);
    
            newMsg.setFlag(msg.getFlag());
            MessageAccessor.setProperties(newMsg, msg.getProperties());
            MessageAccessor.putProperty(newMsg, MessageConst.PROPERTY_RETRY_TOPIC, msg.getTopic());
            MessageAccessor.setReconsumeTime(newMsg, String.valueOf(msg.getReconsumeTimes() + 1));
            MessageAccessor.setMaxReconsumeTimes(newMsg, String.valueOf(getMaxReconsumeTimes()));
            newMsg.setDelayTimeLevel(3 + msg.getReconsumeTimes());
    
            this.mQClientFactory.getDefaultMQProducer().send(newMsg);
        } finally { 
            msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQPushConsumer.getNamespace()));
        }
    }

/** * 消费者向broker发送重试消息 */
    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);
    
        // 只发送消息的一些基本信息，不发送消息体，等发送到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);
        assert response != null;
        switch (response.getCode()) { 
            case ResponseCode.SUCCESS: { 
                return;
            }
            default:
                break;
        }
    
        throw new MQBrokerException(response.getCode(), response.getRemark());
    }

发送这边没什么可分析的，发送的请求码是CONSUMER\_SEND\_MSG\_BACK。

## Broker ##

在broker服务中是使用SendMessageProcessor该处理器来处理的。

/** * 处理发送消息请求 */
    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx,
                                            RemotingCommand request) throws RemotingCommandException { 
        SendMessageContext mqtraceContext;
        switch (request.getCode()) { 
            // 消费者发送的重试消息
            case RequestCode.CONSUMER_SEND_MSG_BACK:
                return this.consumerSendMsgBack(ctx, request);
            // 生产者发送的普通消息
            default:
                // 省略。。
        }
    }
    
    /** * 消费者发送的重试消息请求 */
    private RemotingCommand consumerSendMsgBack(final ChannelHandlerContext ctx, final RemotingCommand request)
        throws RemotingCommandException { 
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        final ConsumerSendMsgBackRequestHeader requestHeader =
            (ConsumerSendMsgBackRequestHeader)request.decodeCommandCustomHeader(ConsumerSendMsgBackRequestHeader.class);
    
        String namespace = NamespaceUtil.getNamespaceFromResource(requestHeader.getGroup());
        if (this.hasConsumeMessageHook() && !UtilAll.isBlank(requestHeader.getOriginMsgId())) { 
    
            ConsumeMessageContext context = new ConsumeMessageContext();
            context.setNamespace(namespace);
            context.setConsumerGroup(requestHeader.getGroup());
            context.setTopic(requestHeader.getOriginTopic());
            context.setCommercialRcvStats(BrokerStatsManager.StatsType.SEND_BACK);
            context.setCommercialRcvTimes(1);
            context.setCommercialOwner(request.getExtFields().get(BrokerStatsManager.COMMERCIAL_OWNER));
    
            this.executeConsumeMessageHookAfter(context);
        }
    
        // 订阅信息
        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 response;
        }
    
        if (!PermName.isWriteable(this.brokerController.getBrokerConfig().getBrokerPermission())) { 
            response.setCode(ResponseCode.NO_PERMISSION);
            response.setRemark("the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1() + "] sending message is forbidden");
            return response;
        }
    
        // 重试队列数量小于1，说明该消费组不支持重试，直接返回成功
        if (subscriptionGroupConfig.getRetryQueueNums() <= 0) { 
            response.setCode(ResponseCode.SUCCESS);
            response.setRemark(null);
            return response;
        }
    
        // 重试消息的主题是 "%RETRY%"+groupName
        String newTopic = MixAll.getRetryTopic(requestHeader.getGroup());
        // 从重试队列中随机选择一个
        int queueIdInt = Math.abs(this.random.nextInt() % 99999999) % subscriptionGroupConfig.getRetryQueueNums();
    
        int topicSysFlag = 0;
        if (requestHeader.isUnitMode()) { 
            topicSysFlag = TopicSysFlag.buildSysFlag(false, true);
        }
    
        // 获取或创建重试时创建重试消息的主题，并向所有的nameServer注册
        // 重试主题在consumer启动时就已经创建了
        TopicConfig topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(
            newTopic,
            subscriptionGroupConfig.getRetryQueueNums(),
            PermName.PERM_WRITE | PermName.PERM_READ, topicSysFlag);
        if (null == topicConfig) { 
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("topic[" + newTopic + "] not exist");
            return response;
        }
    
        if (!PermName.isWriteable(topicConfig.getPerm())) { 
            response.setCode(ResponseCode.NO_PERMISSION);
            response.setRemark(String.format("the topic[%s] sending message is forbidden", newTopic));
            return response;
        }
    
        // 由于发送重试消息不包含消息体，所以需要根据消息偏移量从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 response;
        }
    
        final String retryTopic = msgExt.getProperty(MessageConst.PROPERTY_RETRY_TOPIC);
        if (null == retryTopic) {  // 将消息的原本主题存到property中，key为"RETRY_TOPIC"
            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()) { 
            maxReconsumeTimes = requestHeader.getMaxReconsumeTimes();
        }
    
        // 如果消息重试次数已到达了最大重试次数 或者 延迟级别小于0，则再次改变主题为 "%DLQ%"+groupName，并创建主题
        // 该主题的权限为只写，说明消息一旦进入该主题队列，则消费者无法消费到，需要人工干预
        if (msgExt.getReconsumeTimes() >= maxReconsumeTimes
            || delayLevel < 0) { 
            newTopic = MixAll.getDLQTopic(requestHeader.getGroup());
            queueIdInt = Math.abs(this.random.nextInt() % 99999999) % DLQ_NUMS_PER_GROUP;
    
            topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(newTopic,
                DLQ_NUMS_PER_GROUP,
                PermName.PERM_WRITE, 0
            );
            if (null == topicConfig) { 
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("topic[" + newTopic + "] not exist");
                return response;
            }
        } else { 
            // 延时级别设为 3+已重试次数
            if (0 == delayLevel) { 
                delayLevel = 3 + msgExt.getReconsumeTimes();
            }
            System.out.println("延时级别：" + delayLevel);
            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);
        msgInner.setSysFlag(msgExt.getSysFlag());
        msgInner.setBornTimestamp(msgExt.getBornTimestamp());
        msgInner.setBornHost(msgExt.getBornHost());
        msgInner.setStoreHost(this.getStoreHost());
        // 已重试次数+1
        msgInner.setReconsumeTimes(msgExt.getReconsumeTimes() + 1);
    
        String originMsgId = MessageAccessor.getOriginMessageId(msgExt);
        MessageAccessor.setOriginMessageId(msgInner, UtilAll.isBlank(originMsgId) ? msgExt.getMsgId() : originMsgId);
    
        // 向文件中写入该新设置的消息，如果是延时消息，则会再将主题改为 SCHEDULE_TOPIC_XXXX
        PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
        if (putMessageResult != null) { 
            switch (putMessageResult.getPutMessageStatus()) { 
                case PUT_OK:
                    String backTopic = msgExt.getTopic();
                    String correctTopic = msgExt.getProperty(MessageConst.PROPERTY_RETRY_TOPIC);
                    if (correctTopic != null) { 
                        backTopic = correctTopic;
                    }
    
                    this.brokerController.getBrokerStatsManager().incSendBackNums(requestHeader.getGroup(), backTopic);
    
                    response.setCode(ResponseCode.SUCCESS);
                    response.setRemark(null);
    
                    return response;
                default:
                    break;
            }
    
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark(putMessageResult.getPutMessageStatus().name());
            return response;
        }
    
        response.setCode(ResponseCode.SYSTEM_ERROR);
        response.setRemark("putMessageResult is null");
        return response;
    }

Step1：判断订阅配置是否为空。  
Step2：判断重试队列数量是否小于1，如果小于1，说明该消费组不支持重试，直接返回成功。  
Step3：获取重试消息topic和queueId，topic默认为"%RETRY%"+groupName，queueId为0。  
Step4：获取或创建重试消息主题，并向所有的nameServer注册（topic可能在consumer启动时就创建好了）。  
Step5：由于接收到的重试消息不包含消息体，所以需要根据消息偏移量从commitLog中取出完整的消息。  
Step6：将消息的原本主题存到property中，key为"RETRY\_TOPIC"。  
Step7：判断消息重试次数，如果消息重试次数已到达了最大重试次数 或者 延迟级别小于0，则再次改变主题为 “%DLQ%”+groupName，并创建主题，即将进入死信队列，否则延时级别设置为3+已重试次数。  
Step8：根据原先消息创建一个新的消息，该重试消息拥有自己的消息ID，会将原先的主题、消息ID存入消息的属性中，其它属性与原先消息保持相同。  
Step9：已重试次数+1。  
Step10：向文件中写入该新的消息，由于重试消息有延时，所以在写入commitLog时会再将主题改为 SCHEDULE\_TOPIC\_XXXX，队列ID为延迟级别减1。再次将消息主题、队列存人消息的属性中，键分别为: PROPERTY\_ REAL\_ TOPIC、PROPERTY\_ REAL\_ QUEUE\_ \_ID。

写入延时的CommitLog后会依托定时任务从延时队列中取出重新放到重试队列中，由消费者从重试队列中拉取消费。延时的具体流程等后面的文档分析。  
![在这里插入图片描述][8ac4cdb3d55f413014bf14b3d5d27475.png_pic_center]

## 参考资料 ##

《儒猿技术窝——从 0 开始带你成为消息中间件实战高手》

[8ac4cdb3d55f413014bf14b3d5d27475.png_pic_center]: /images/20220828/95073e20fcdd4e76b0eb2fc4e92563a4.png