RocketMQ原理学习--失败消息实现原理

柔光的暖阳◎ 2022-04-04 05:50 305阅读 0赞

上一篇博客[《 RocketMQ原理学习--延时消息实现原理》][RocketMQ_--]我们介绍了延时消息的实现原理，这篇博客我们看看RocketMQ在消费者消息消费失败的情况下是如何处理的。

正常情况下消费者在消息消费失败时会马上重新消费消息，正常情况下再次消费应该会将消息消费完成的，但是当一次消费消费不成功时我们可以设置让Broker延时一段时间再重新投递消息，是不是看到这就发现和延时消息类似了。

### 1、示例 ###

当我们的业务没有处理完成时，可以返回ConsumeConcurrentlyStatus.RECONSUME\_LATER，让Broker延时再投递消息。

public class PushConsumer {
    
        public static void main(String[] args) throws InterruptedException, MQClientException {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rmq-group");
            consumer.setNamesrvAddr("localhost:9876");
            //consumer.setInstanceName("rmq-instance2");
            consumer.subscribe("TopicA-test", "TagA");
    
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                public ConsumeConcurrentlyStatus consumeMessage(
                        List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        System.out.println(new String(msg.getBody()));
                    }
                    //延时再发送
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            });
            consumer.start();
            System.out.println("Consumer Started.");
        }
    }

### 2、Broker延时消息处理 ###

Broker的SendMessageProcessor的consumerSendMsgBack方法会处理消费者ack失败的消息，发送回的消息会设置重试Topic，重试Topic的名称默认为："%RETRY%" + ConsumerGroup，然后会设置消息的延时机制，默认级别为3表示10秒，这样再延时投送消息。

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);
    
            if (this.hasConsumeMessageHook() && !UtilAll.isBlank(requestHeader.getOriginMsgId())) {
    
                ConsumeMessageContext context = new ConsumeMessageContext();
                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;
            }
    
            if (subscriptionGroupConfig.getRetryQueueNums() <= 0) {
                response.setCode(ResponseCode.SUCCESS);
                response.setRemark(null);
                return response;
            }
    
            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);
            }
    
            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;
            }
    
            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;
            }
            //设置为新的topic名称:"%RETRY%+consumerGroup
            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()) {
                maxReconsumeTimes = requestHeader.getMaxReconsumeTimes();
            }
    
            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 {
                //设置延时级别，这样消息在写入commitlog中会被作为延时消息持久化
                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);
            msgInner.setSysFlag(msgExt.getSysFlag());
            msgInner.setBornTimestamp(msgExt.getBornTimestamp());
            msgInner.setBornHost(msgExt.getBornHost());
            msgInner.setStoreHost(this.getStoreHost());
            msgInner.setReconsumeTimes(msgExt.getReconsumeTimes() + 1);
    
            String originMsgId = MessageAccessor.getOriginMessageId(msgExt);
            MessageAccessor.setOriginMessageId(msgInner, UtilAll.isBlank(originMsgId) ? msgExt.getMsgId() : originMsgId);
            //持久化消息，会将消息作为延时消息处理
            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;
        }

总结：

对于失败重试消息RocketMQ将其处理为延时消息进行操作，进行延时投递就可以了，设计还是比较牛的。

引用：

[RocketMQ 消息消费失败的处理策略][RocketMQ]

[RocketMQ_--]: https://blog.csdn.net/qq924862077/article/details/84987179
[RocketMQ]: http://blog.soliloquize.org/2018/09/02/RocketMQ-%E6%B6%88%E6%81%AF%E6%B6%88%E8%B4%B9%E5%A4%B1%E8%B4%A5%E7%9A%84%E5%A4%84%E7%90%86%E7%AD%96%E7%95%A5/