文档章节

[RocketMQ]消息中间件—RocketMQ消息消费(三)(消息消费重试)

morpheusWB
 morpheusWB
发布于 2018/09/29 20:00
字数 3201
阅读 123
收藏 4

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

一、其他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的消息。

二、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

三、总结

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



作者:癫狂侠
链接:https://www.jianshu.com/p/5843cdcd02aa
來源:简书
简书著作权归作者所有,任何形式的转载都请联系作者获得授权并注明出处。

本文转载自:https://www.jianshu.com/p/5843cdcd02aa

morpheusWB
粉丝 27
博文 84
码字总数 14703
作品 0
西安
程序员
私信 提问
[RocketMQ]消息中间件—RocketMQ消息消费(一)

文章摘要:在发送消息给RocketMQ后,消费者需要消费。消息的消费比发送要复杂一些,那么RocketMQ是如何来做的呢? 在RocketMQ系列文章的前面几篇幅中已经对其“RPC通信部分”和“普通消息发送...

morpheusWB
2018/09/29
0
0
消息中间件—RocketMQ消息消费(一)

文章摘要:在发送消息给RocketMQ后,消费者需要消费。消息的消费比发送要复杂一些,那么RocketMQ是如何来做的呢? 在RocketMQ系列文章的前面几篇幅中已经对其“RPC通信部分”和“普通消息发送...

癫狂侠
2018/08/12
0
0
RocketMQ与Kafka对比

RocketMQ与Kafka对比(18项差异) 淘宝内部的交易系统使用了淘宝自主研发的Notify消息中间件,使用Mysql作为消息存储媒介,可完全水平扩容,为了进一步降低成本,我们认为存储部分可以进一步...

莫问viva
2015/05/08
0
0
使用 RocketMQ 实现延时消息

一. 延时消息 延时消息是指消息被发送以后,并不想让消费者立即拿到消息,而是等待指定时间后,消费者才拿到这个消息进行消费。 使用延时消息的典型场景,例如: 在电商系统中,用户下完订单...

fengzhizi715
07/01
0
0
Apache RocketMQ QuickStart

RocketMQ作为一款分布式的消息中间件(阿里的说法是不遵循任何规范的,所以不能完全用JMS的那一套东西来看它),经历了Metaq1.x、Metaq2.x的发展和淘宝双十一的洗礼,在功能和性能上远超Act...

程序员诗人
2017/09/29
0
0

没有更多内容

加载失败,请刷新页面

加载更多

RocketMQ的事务投递

RocketMQ的事务投递 这是阿里的分布式事务图: 1、A服务先发送个Half Message给Brock端,消息中携带 B服务 即将要+100元的信息。 2、当A服务知道Half Message发送成功后,那么开始第3步执行本...

春哥大魔王的博客
17分钟前
1
0
Qt编写自定义控件31-面板仪表盘控件

一、前言 在Qt自定义控件中,仪表盘控件是数量最多的,写仪表盘都写到快要吐血,可能是因为各种工业控制领域用的比较多吧,而且仪表盘又是比较生动直观的,这次看到百度的echart中有这个控件...

飞扬青云
23分钟前
1
0
DisplayPort 迎来重大更新,数据带宽性能提高3倍

VESA宣布了他们对DisplayPort接口三年来的第一次重大更新。 与DP 1.4a相比,DisplayPort 2.0提供了三倍于DP 1.4a的数据带宽性能,支持超过8K的分辨率,更高的刷新率和更高分辨率的HDR,以及其...

linuxCool
30分钟前
1
0
《Linux就该这么学》2019年7月20日第八天上课笔记

du命令 du -sh /newFS/ 察看文件/文件夹数据占用量 SWAP 交换分区的设置 磁盘容量配额 RHEL 5/6 usrquota RHEL 7 quota 软硬连接 ln 软 指针指向inode 硬 建立新的inode RAID 0 1 5 1+0...

2lodoss
32分钟前
1
0
适合钱包应用开发的ERC20代币数据集

Erc20Tokens数据集包含超过1000种主流的以太坊ERC20代币的描述数据清单和图标,可用于钱包等区块链应用的开发,支持使用Java、Python、Php、NodeJs、C#等各种开发语言查询主流ERC20代币的相关...

汇智网教程
56分钟前
1
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部