Python 官方文档:入门教程 => 点击学习
目录一、概述二、生产端的消息重试三、消费端的消息重试四、消息重试次数五、消息重试配置六、消息重试原理一、概述 由于网络抖动、服务宕机等一些不确定的因素,RocketMQ在发送消息的时
由于网络抖动、服务宕机等一些不确定的因素,RocketMQ在发送消息的时候很有可能出现消息发送或者消费失败的问题。
Consumer消费消息失败通常可以认为有以下几种情况:
如果没有消息重试机制,就可能产生消息丢失的问题,这样就会对系统产生较大的影响。RocketMQ内部封装了消息重试的处理流程,无需开发人员手动处理,并且支持了生产端、消费端两端的重试机制。
生产端的消息重试是指:Producer往Broker上发消息没有发送成功,比如网络原因导致生产者发送消息到MQ失败,即发送端没有收到Broker的ACK,导致最终Consumer无法消费消息,此时RocketMQ会自动进行重试。
生产者端的消息重试配置比较简单,只需要在定义生产者的时候,调用producer.setRetryTimesWhenSendFailed(xxx)方法设置消息发送失败的最大重试次数。如下:
// 同步发送消息,如果5秒内没有发送成功,则重试3次
DefaultMQProducer producer = new DefaultMQProducer("DefaultProducer");
producer.setRetryTimesWhenSendFailed(3);
producer.send(msg, 5000L);
同样的,由于网络原因,Broker发送消息给消费者后,没有受到消费端的ACK响应,所以Broker又会尝试将消息重新发送给Consumer,在实际开发过程中,我们更应该考虑的是消费端的重试。消费端的消息重试可以分为顺序消息的重试以及无序消息的重试。
(1)、顺序消息的重试
对于顺序消息,当消费者消费消息失败后,消息队列 RocketMQ 会自动不断进行消息重试(每次间隔时间为 1 秒),这时应用会出现消息消费被阻塞的情况。因此,在使用顺序消息时,务必保证应用能够及时监控并处理消费失败的情况,避免阻塞现象的发生。
(2)、无序消息的重试
对于无序消息(普通、延时、事务消息),当消费者消费消息失败时,可以通过设置返回状态达到消息重试的结果。
需要注意的是:无序消息的重试只会针对集群消费方式(MessageModel.CLUSTERING)生效;广播方式不提供失败重试特性,即消费失败后,失败的消息不再重试,继续消费新的消息。
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 次后仍然失败,消息将不再投递。
注意: 一条消息无论重试多少次,这些重试消息的 Message ID 不会改变。所以就需要我们消费者端做好消费幂等操作。
集群消费方式下,消息消费失败后期望消息重试,需要在消息监听器接口的实现中明确进行配置(下述三种方式任选一种):
public class MessageListenerImpl implements MessageListener {
@Override
public Action consume(Message message, ConsumeContext context) {
//处理消息
//.....
//方式1:返回 Action.ReconsumeLater,消息将重试
return Action.ReconsumeLater;
//方式2:返回 null,消息将重试
return null;
//方式3:直接抛出异常, 消息将重试
throw new RuntimeException("消费消息发生异常");
}
}
集群消费方式下,如果希望消息失败后,不进行消息重试,那么我们可以捕获消费逻辑中可能抛出的异常,然后返回Action.CommitMessage,那么这条消息将不会再重试。如下:
public class MessageListenerImpl implements MessageListener {
@Override
public Action consume(Message message, ConsumeContext context) {
try {
// 消费消息....
} catch (Throwable e) {
// 捕获消费逻辑中的所有异常,并返回 Action.CommitMessage;
return Action.CommitMessage;
}
// 消息处理正常,直接返回 Action.CommitMessage;
return Action.CommitMessage;
}
}
当然,RocketMQ也允许Consumer 启动的时候设置最大重试次数,重试时间间隔将按照如下策略:
Properties properties = new Properties();
// 配置对应 Group ID的最大消息重试次数为 20 次
properties.put(PropertyKeyConst.MaxReconsumeTimes, "20");
Consumer consumer =ONSFactory.createConsumer(properties);
注意:
RocketMQ会为每个消费者组都设置一个Topic名称为“%RETRY%+consumerGroup”的重试队列(这里需要注意的是,这个Topic的重试队列是针对消费组,而不是针对每个Topic设置的),用于暂时保存因为各种异常而导致Consumer端无法消费的消息。
考虑到异常恢复需要一些时间,RocketMQ会为重试队列设置多个重试级别,每个重试级别都有与之对应的重新投递延时,重试次数越多投递延时就越大。RocketMQ对于重试消息的处理是先保存至Topic名称为“SCHEDULE_TOPIC_XXXX”的延迟队列中,后台定时任务按照对应的时间进行Delay后重新保存至“%RETRY%+consumerGroup”的重试队列中。
到此这篇关于RocketMQ消息重试机制原理分析讲解的文章就介绍到这了,更多相关RocketMQ消息重试内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网!
--结束END--
本文标题: RocketMQ消息重试机制原理分析讲解
本文链接: https://www.lsjlt.com/news/195809.html(转载时请注明来源链接)
有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
下载Word文档到电脑,方便收藏和打印~
2024-03-01
2024-03-01
2024-03-01
2024-02-29
2024-02-29
2024-02-29
2024-02-29
2024-02-29
2024-02-29
2024-02-29
回答
回答
回答
回答
回答
回答
回答
回答
回答
回答
0