当前位置:   article > 正文

16.RocketMQ之消费重试以及原理

rocketmq重试

highlight: arduino-light

1.4 消费重试

image.png

对于顺序消息,当消费者消费消息失败后,消费者会在本地自动不断进行消息重试,每次间隔时间为 1 秒,重试最大值是 Integer.MAX_VALUE。

对于无序消息(普通、定时、延时、事务消息)当消费者消费消息失败时可以通过设置返回状态达到重试的目的。

广播模式下消息队列RocketMQ保证每条消息至少被每台客户端消费一次,但是并不会对消费失败的消息进行失败重投,因此业务方需要关注消费失败的情况。

1.4.1顺序消息自动重试:需要避免阻塞

当消费者消费消息失败后,消费者会在本地自动不断进行消息重试,每次间隔时间为 1 秒,重试最大值是 Integer.MAX_VALUE。这时,应用会出现消息消费被阻塞的情况。

因此,在使用顺序消息时,务必保证应用能够及时监控并处理消费失败的情况,避免阻塞现象的发生。

1.4.2无序消息重试:广播模式不会失败重投

无序消息的重试只针对集群消费方式生效。

广播方式不提供失败重试特性,即消费失败后,失败消息不再重试,继续消费新的消息。

广播模式下消息队列 RocketMQ保证每条消息至少被每台客户端消费一次,但是并不会对消费失败的消息进行失败重投,因此业务方需要关注消费失败的情况,可以自己做一些重试的操作比如消费失败做持久化,使用定时器重新重试数据库中持久化的数据。

不会失败重投的原因是一旦消费失败做了消费重投,所有的消费者都需要再消费一次!

1.4.3无序消息重试:集群模式需要返回消费失败状态

对于无序消息(普通、定时、延时、事务消息)当消费者消费消息失败时可以通过设置返回状态达到重试的目的。

无序消息的重试只针对集群消费方式生效。

广播方式不提供失败重试特性,即消费失败后,失败消息不再重试,继续消费新的消息。

默认消息消费重试次数:16

默认最大重试次数为16,所以一个消息最多可以消费17次。

为什么是16?

默认设置延迟级别范围为3到18。因为18个等级,从3开始重试,18-3+1 =15。

消息队列 RocketMQ 默认允许每条消息最多重试 16 次,每次重试的间隔时间如下:

| 第几次重试 | 与上次重试的间隔时间 | 第几次重试 | 与上次重试的间隔时间 | | ----- | ---------- | ----- | ---------- | | 1 | 10 秒 | 9 | 7 分钟 | | 2 | 30 秒 | 10 | 8 分钟 | | 3 | 1 分钟 | 11 | 9 分钟 | | 4 | 2 分钟 | 12 | 10 分钟 | | 5 | 3 分钟 | 13 | 20 分钟 | | 6 | 4 分钟 | 14 | 30 分钟 | | 7 | 5 分钟 | 15 | 1 小时 | | 8 | 6 分钟 | 16 | 2 小时 |

如果消息重试 16 次后仍然失败,消息将不再投递。如果严格按照上述重试时间间隔计算,某条消息在一直消费失败的前提下,将会在接下来的 4 小时 46 分钟之内进行 16 次重试,超过这个时间范围消息将不再重试投递。

注意: 一条消息无论重试多少次,这些重试消息的 Message ID 不会改变。

自定义消息最大重试次数

消息队列 RocketMQ 允许 Consumer 启动的时候设置最大重试次数,重试时间间隔将按照如下策略:

  • 最大重试次数小于等于16 次,则重试时间间隔同上表描述。
  • 最大重试次数大于 16 次,超过 16 次的重试时间间隔均为每次 2 小时。

java Properties properties = new Properties(); //配置对应 Group ID 的最大消息重试次数为 20 次 properties.put(PropertyKeyConst.MaxReconsumeTimes,"20"); Consumer consumer =ONSFactory.createConsumer(properties);

注意:

  • 消息最大重试次数的设置对相同 Group ID 下的所有 Consumer 实例有效。
  • 如果只对相同 Group ID 下两个 Consumer 实例中的其中一个设置了 MaxReconsumeTimes,那么该配置对两个 Consumer 实例均生效。
  • 配置采用覆盖的方式生效,即最后启动的 Consumer 实例会覆盖之前的启动实例的配置
获取消息重试次数

消费者收到消息后,可按照如下方式获取消息的重试次数:

java public class MessageListenerImpl implements MessageListener {    @Override    public Action consume(Message message, ConsumeContext context) {        //获取消息的重试次数        System.out.println(message.getReconsumeTimes());        return Action.CommitMessage;   } }

消费失败后,重试配置方式

集群消费方式下,消息消费失败后期望消息重试,需要在消息监听器接口的实现中明确进行配置(三种方式任选一种):

  • 返回 Action.ReconsumeLater (推荐)
  • 返回 Null
  • 抛出异常

java public class MessageListenerImpl implements MessageListener {    @Override    public Action consume(Message message, ConsumeContext context) {        //处理消息        doConsumeMessage(message);        //方式1:返回 Action.ReconsumeLater,消息将重试        return Action.ReconsumeLater;        //方式2:返回 null,消息将重试        return null;        //方式3:直接抛出异常, 消息将重试        throw new RuntimeException("Consumer Message exceotion");   } }

消费失败后,不重试配置方式

集群消费方式下,消息失败后期望消息不重试,需要捕获消费逻辑中可能抛出的异常,最终返回 Action.CommitMessage,此后这条消息将不会再重试。

java public class MessageListenerImpl implements MessageListener {    @Override    public Action consume(Message message, ConsumeContext context) {        try {            doConsumeMessage(message);       } catch (Throwable e) {            //捕获消费逻辑中的所有异常,并返回 Action.CommitMessage;            return Action.CommitMessage;       }        //消息处理正常,直接返回 Action.CommitMessage;        return Action.CommitMessage;   } }

当然 我们也可以根据

重试原理

发送到重试消费Topic 是%RETRY% + 消费组名 注意是消费组名

我们思考一下为什么是消费者组名而不是topic的名字?

A消费者组消费成功 B消费者组消费失败

如果发回原topic就有问题了,A又会消费一次。

消息消费失败会设置DelayLevel并发回重试主题,重试主题 是%RETRY% + 消费组名,发回失败会立刻重试1次。

发回重试主题时,DelayTimeLeve从3开始

我们知道RocketMQ的默认的配置是messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h,分别代表延迟level1-level18,为什么不是从1开始呢?

newMsg.setDelayTimeLevel(3 + msg.getReconsumeTimes());//默认的重试次数是0

所以默认重试时,实际是从3开始的。也就是第一次重试的时间间隔是10秒,这也解释了为什么是16次!3-18可不就是16次!

SCHEDULETOPICXXXX的队列名称是从2开始到17,对应的delayLevel为3到18,3对应10s,18对应2h,在类MessageStoreConfig中这样定义延时时间:

String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"。

延迟消息对应的Topic是SCHEDULETOPICXXXX,注意就是SCHEDULETOPICXXXX,XXXX不是某某某的意思。

image.png

SCHEDULETOPICXXXX 介绍 SCHEDULETOPICXXXX 是 RocketMQ 一个系统类型的 Topic,用于标识延时消息。

这个 Topic 有 18 个队列,分别唯一对应着 RocketMQ 的 18 个延时等级,对应关系为:queueId = delayTimeLevel – 1。

ScheduleMessageService 介绍 这是 Broker 中的一个延时服务,专门消费 Topic 为 SCHEDULETOPICXXXX 的延时消息,并将其投递到目标 Topic 中。

ScheduleMessageService 在启动时,会创建一个定时器 Timer,并根据延迟级别的个数,启动对应数量的 TimerTask,每个TimerTask负责一个延迟级别的消费与投递。

SCHEDULETOPICXXXX的队列名称是从2开始到17,对应的delayLevel为3到18,3对应10s,18对应2h,在类MessageStoreConfig中这样定义延时时间:

String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"。

SCHEDULETOPICXXXX这个topic只对内部使用,对于consumer只能消费到自己所在的消费者组的重试topic的数据。

比如A是OrderConsumer组的一个消费者,OrderConsumer消费的主题Topic是Order。

那么消费者A在启动的时候,会订阅Order主题的同时,还会订阅%RETRY%_OrderConsumer。

即订阅Order主题的同时,还会订阅%RETRY%+消费者组的主题。

consumer消费失败的消息发回broker后总是先写到SCHEDULETOPICXXXX里面,然后schedule service在延迟时间到了以后会读取SCHEDULETOPICXXXX里面的数据然后重新发回到重试主题,consumer订阅了重试主题,所以会重新消费失败的数据,这样就完成了一个循环。

rocketmq 先将不同延时等级的消息存入内部对应延时队列中,然后不断的从延时队列中拉取消息判断是否到期,然后进行投递到对应的topic中。

从这个过程也能看到,一个消费失败的消息体每次发回broker需要在commitLog里面存储两份。

topic为SCHEDULETOPICXXXX的一份这个主要是为schedule service控制延时用的。

topic为%RETRY%groupName的一份。

参考链接:https://blog.csdn.net/gelald/article/details/126949527

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/IT小白/article/detail/603984
推荐阅读
相关标签
  

闽ICP备14008679号