[RocketMQ]消息中间件—RocketMQ消息消费（三）（消息消费重试）

摘要：若是Consumer端消费消息失败，那么RocketMQ是如何对失败的异常状况进行处理？
前面两篇RocketMQ消息消费（一）/（二）篇，主要从Push/Pull两种消费模式的简要流程、长轮询机制和Consumer端负载均衡这几点内容出发，介绍了RocketMQ消息消费的正常流程和细节内容，本篇内容将主要介绍Consumer端消费失败的异常流程。
这里先回顾往期RocketMQ技术分享的篇幅：
（1）消息中间件—RocketMQ的RPC通讯（一）
（2）消息中间件—RocketMQ的RPC通讯（二）
（3）消息中间件—RocketMQ消息发送
（4）消息中间件—RocketMQ消息消费（一）
（5）消息中间件—RocketMQ消息消费（二）（push模式实现）

1、其余MQ中间件消费端可靠性的保障

在业务开发中，你们必定都遇到过业务工程由于各种异常（多是业务工程自己的异常、JVM内存异常或者系统所在的虚拟机宕机等），而致使MQ中间件发送过来的业务消息消费失败而没法再次消费该消息的状况。目前，不少MQ消息中间件都有相应的机制和方法来保证Consumer端消费消息的可靠性。下面先来看看RabbitMQ和Kafka这两款MQ消息中间件是如何来保证消费者端消息处理的可靠性的呢？

1.1 简谈RabbitMQ的手动消息确认ACK机制

RabbitMQ提供了消息确认机制。消费者在订阅队列时，能够在代码中手动设置autoAck参数为false，这时RabbitMQ会等待消费者显式地回复确认信号（即为显式地调用channel.basicAck(envelope.getDeliveryTag(), false)方法）后才从集群中的内存（或磁盘）节点上移除消息，从而保证了这条消息不会由于消费失败而致使丢失。

1.2 简析Kafka消息消费的手动提交

在Kafka中，也能够采用上面那种的消费后的确认机制，经过在Consumer端设置“enable.auto.commit”属性为false后，待业务工程正常处理完消费后，在代码中手动调用KafkaConsumer实例的commitSync()方法提交（ps：这里指的是同步阻塞commit消费的偏移量，等待Broker端的返回响应，须要注意Broker端在对commit请求作出响应以前，消费端会处于阻塞状态，从而限制消息的处理性能和总体吞吐量），以确保消息可以正常被消费。若是在消费过程当中，消费端忽然Crash，这时候消费偏移量没有commit，等正常恢复后依然还会处理刚刚未commit的消息。

2、RocketMQ消费失败后的消费重试机制

对比了另外两款MQ中间件后，接下来进入正题，主要来讲说RocketMQ在消费失败后的是如何来保证消息消费的可靠性？

2.1 重试队列与死信队列的概念

在介绍RocketMQ的消费重试机制以前，须要先来讲下“重试队列”和“死信队列”两个概念。
（1）重试队列：若是Consumer端由于各类类型异常致使本次消费失败，为防止该消息丢失而须要将其从新回发给Broker端保存，保存这种由于异常没法正常消费而回发给MQ的消息队列称之为重试队列。RocketMQ会为每一个消费组都设置一个Topic名称为“%RETRY%+consumerGroup”的重试队列（这里须要注意的是，这个Topic的重试队列是针对消费组，而不是针对每一个Topic设置的），用于暂时保存由于各类异常而致使Consumer端没法消费的消息。考虑到异常恢复起来须要一些时间，会为重试队列设置多个重试级别，每一个重试级别都有与之对应的从新投递延时，重试次数越多投递延时就越大。RocketMQ对于重试消息的处理是先保存至Topic名称为“SCHEDULE_TOPIC_XXXX”的延迟队列中，后台定时任务按照对应的时间进行Delay后从新保存至“%RETRY%+consumerGroup”的重试队列中（具体细节后面会详细阐述）。
（2）死信队列：因为有些缘由致使Consumer端长时间的没法正常消费从Broker端Pull过来的业务消息，为了确保消息不会被无端的丢弃，那么超过配置的“最大重试消费次数”后就会移入到这个死信队列中。在RocketMQ中，SubscriptionGroupConfig配置常量默认地设置了两个参数，一个是retryQueueNums为1（重试队列数量为1个），另一个是retryMaxTimes为16（最大重试消费的次数为16次）。Broker端经过校验判断，若是超过了最大重试消费次数则会将消息移至这里所说的死信队列。这里，RocketMQ会为每一个消费组都设置一个Topic命名为“%DLQ%+consumerGroup"的死信队列。通常在实际应用中，移入至死信队列的消息，须要人工干预处理；

2.1 Consumer端回发消息至Broker端

在业务工程中的Consumer端（Push消费模式下），若是消息可以正常消费须要在注册的消息监听回调方法中返回CONSUME_SUCCESS的消费状态，不然由于各种异常消费失败则返回RECONSUME_LATER的消费状态。消费状态的枚举类型以下所示：

public enum ConsumeConcurrentlyStatus {
    //业务方消费成功
    CONSUME_SUCCESS,
    //业务方消费失败，以后进行从新尝试消费
    RECONSUME_LATER;
}

若是业务工程对消息消费失败了，那么则会抛出异常而且返回这里的RECONSUME_LATER状态。这里，在消费消息的服务线程—consumeMessageService中，将封装好的消息消费任务ConsumeRequest提交至线程池—consumeExecutor异步执行。从消息消费任务ConsumeRequest的run()方法中会执行业务工程中注册的消息监听回调方法，并在processConsumeResult方法中根据业务工程返回的状态（CONSUME_SUCCESS或者RECONSUME_LATER）进行判断和作对应的处理（下面讲的都是在消费通讯模式为集群模型下的，广播模型下的比较简单就再也不分析了）。
（1）业务方正常消费（CONSUME_SUCCESS）：正常状况下，设置ackIndex的值为consumeRequest.getMsgs().size() - 1，所以后面的遍历consumeRequest.getMsgs()消息集合条件不成立，不会调用回发消费失败消息至Broker端的方法—sendMessageBack(msg, context)。最后，更新消费的偏移量；
（2）业务方消费失败（RECONSUME_LATER）：异常状况下，设置ackIndex的值为-1，这时就会进入到遍历consumeRequest.getMsgs()消息集合的for循环中，执行回发消息的方法—sendMessageBack(msg, context)。这里，首先会根据brokerName获得Broker端的地址信息，而后经过网络通讯的Remoting模块发送RPC请求到指定的Broker上，若是上述过程失败，则建立一条新的消息从新发送给Broker，此时新消息的Topic为“%RETRY%+ConsumeGroupName”—重试队列的主题。其中，在MQClientAPIImpl实例的consumerSendMessageBack()方法中封装了ConsumerSendMsgBackRequestHeader的请求体，随后完成回发消费失败消息的RPC通讯请求（业务请求码为：CONSUMER_SEND_MSG_BACK）。假若上面的回发消息流程失败，则会延迟5S后从新在Consumer端进行从新消费。与正常消费的状况同样，在最后更新消费的偏移量；

2.3 Broker端对于回发消息处理的主要流程

Broker端收到这条Consumer端回发过来的消息后，经过业务请求码（CONSUMER_SEND_MSG_BACK）匹配业务处理器—SendMessageProcessor来处理。在完成一系列的前置校验（这里主要是“消费分组是否存在”、“检查Broker是否有写入权限”、“检查重试队列数是否大于0”等）后，尝试获取重试队列的TopicConfig对象（若是是第一次没法获取到，则调用createTopicInSendMessageBackMethod()方法进行建立）。根据回发过来的消息偏移量尝试从commitlog日志文件中查询消息内容，若不存在则返回异常错误。
而后，设置重试队列的Topic—“%RETRY%+consumerGroup”至MessageExt的扩展属性“RETRY_TOPIC”中，并对根据延迟级别delayLevel和最大重试消费次数maxReconsumeTimes进行判断，若是超过最大重试消费次数（默认16次），则会建立死信队列的TopicConfig对象（用于后面将回发过来的消息移入死信队列）。在构建完成须要落盘的MessageExtBrokerInner对象后，调用“commitLog.putMessage(msg)”方法作消息持久化。这里，须要注意的是，在putMessage(msg)的方法里会使用“SCHEDULE_TOPIC_XXXX”和对应的延迟级别队列Id分别替换MessageExtBrokerInner对象的Topic和QueueId属性值，并将原来设置的重试队列主题（“%RETRY%+consumerGroup”）的Topic和QueueId属性值作一个备份分别存入扩展属性properties的“REAL_TOPIC”和“REAL_QID”属性中。看到这里也就大体明白了，回发给Broker端的消费失败的消息并不是直接保存至重试队列中，而是会先存至Topic为“SCHEDULE_TOPIC_XXXX”的定时延迟队列中。

疑问：上面说了RocketMQ的重试队列的Topic是“%RETRY%+consumerGroup”，为啥这里要保存至Topic是“SCHEDULE_TOPIC_XXXX”的这个延迟队列中呢？

在源码中搜索下关键字—“SCHEDULE_TOPIC_XXXX”，会发现Broker端还存在着一个后台服务线程—ScheduleMessageService（经过消息存储服务—DefaultMessageStore启动），经过查看源码能够知道其中有一个DeliverDelayedMessageTimerTask定时任务线程会根据Topic（“SCHEDULE_TOPIC_XXXX”）与QueueId，先查到逻辑消费队列ConsumeQueue，而后根据偏移量，找到ConsumeQueue中的内存映射对象，从commitlog日志中找到消息对象MessageExt，并作一个消息体的转换（messageTimeup()方法，由定时延迟队列消息转化为重试队列的消息），再次作持久化落盘，这时候才会真正的保存至重试队列中。看到这里就能够解释上面的疑问了，定时延迟队列只是为了用于暂存的，而后延迟一段时间再将消息移入至重试队列中。RocketMQ设定不一样的延时级别delayLevel，而且与定时延迟队列相对应，具体源码以下：

//省略
    private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
    /**
     * 定时延时消息主题的队列与延迟等级对应关系
     * @param delayLevel
     * @return
     */
    public static int delayLevel2QueueId(final int delayLevel) {
        return delayLevel - 1;
    }

2.4 Consumer端消费重试机制

每一个Consumer实例在启动的时候就默认订阅了该消费组的重试队列主题，DefaultMQPushConsumerImpl的copySubscription()方法中的相关代码以下：

private void copySubscription() throws MQClientException {
            //省略其余代码...
            switch (this.defaultMQPushConsumer.getMessageModel()) {
                case BROADCASTING:
                    break;
                case CLUSTERING://若是消息消费模式为集群模式，还须要为该消费组对应一个重试主题
                    final String retryTopic = MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup());
                    SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(),
                        retryTopic, SubscriptionData.SUB_ALL);
                    this.rebalanceImpl.getSubscriptionInner().put(retryTopic, subscriptionData);
                    break;
                default:
                    break;
            }
            //省略其余代码...
      }

所以，这里也就清楚了，Consumer端会一直订阅该重试队列主题的消息，向Broker端发送以下的拉取消息的PullRequest请求，以尝试从新再次消费重试队列中积压的消息。

PullRequest [consumerGroup=CID_JODIE_1, messageQueue=MessageQueue [topic=%RETRY%CID_JODIE_1, brokerName=HQSKCJJIDRRD6KC, queueId=0], nextOffset=51]

最后，给出一张RocketMQ消息重试机制的框图（ps：这里只是描述了消息消费失败后重试拉取的部分重要过程）：

 

RocketMQ消息重试机制.jpg

3、总结

RocketMQ的消息消费（三）（消息消费重试）篇幅就先分析到这里了。关于RocketMQ消息消费的内容比较多也比较复杂，须要读者结合源码并屡次debug（能够经过分别在Consumer端和Broker端的部分重要方法中打印重要对象中的各个属性值的方式，来仔细研究下其中的过程），才能够对其有一个较为深入的理解。限于笔者的才疏学浅，对本文内容可能还有理解不到位的地方，若有阐述不合理之处还望留言一块儿探讨。

做者：癫狂侠 连接：https://www.jianshu.com/p/5843cdcd02aa 來源：简书 简书著做权归做者全部，任何形式的转载都请联系做者得到受权并注明出处。