RocketMQ 消息重试与死信队列
RocketMQ 前面系列文章如下:
- Rocke服务器托管网tMQ系列(一) 基本介绍
- RocketMQ 系列(二) 环境搭建
- RocketMQ 系列(三) 集成 SpringBoot
- RocketMQ 系列(四) 消息存储
- RocketMQ 系列(五)高可用与负载均衡
消息队列中的消息消费时并不能保证总是成功的,那失败的消息该怎么进行消息补偿呢?这就用到今天的主角消息重试和死信队列了。
1、Producer 消息重试
有时因为网路等原因生产者也可能发送消息失败,也会进行消息重试,Producer 消息重试比较简单,在 Springboot 中只要在配置文件中配置一下就可以了。
yml 文件配置如下:
rocketmq:
producer:
# 发送同步消息失败时,重试次数,默认是 2
retry-times-when-send-failed: 2
# 发送异步消息失败时,重试次数,默认是 2
retry-times-when-send-async-failed: 2
2、Consumer 消息重试
有两种消费模式:集群消费模式和广播消费模式。消息重试只针对集群消费模式生效;广播消费模式不提供失败重试特性,即消费失败后,失败消息不再重试,继续消费新的消息。我们知道消费类型又分为 push 消费与 pull 消费,消息重试只针对 push 消费,分为顺序消息的重试与无序消息的重试。
2.1、顺序消息的消费重试
对于顺序消息,为了保证消息消费的顺序性,当consumer消费失败后,消息队列会自动不断进行消息重试(每次间隔时间为 1s),
这时会导致consumer消费被阻塞的情况,故必须保证应用能够及时监控并处理消费失败的情况,避免阻塞现象的发生
2.2、无序消息的消费重试
对于无序消息(普通、定时、延时、事务消息),当消费者消费消息失败时,您可以通过设置返回状态达到消息重试的结果。
无序消息的重试只针对集群消费方式生效;广播方式不提供失败重试特性,即消费失败后,失败消息不再重试,继续消费新的消息。
无序消息消费失败后并不是投递回原 Topic,而是投递到一个特殊 Topic,其命名为 %RETRY%ConsumerGroupName,集群模式下并发消费每一个ConsumerGroup 会对应一个特殊 Topic,并会订阅该 Topic。
顺序消息与无序消息的消费参数差别如下:
消费类型 | 重试间隔 | 最大重试次数 |
---|---|---|
顺序消息消费 | 间隔时间可通过自定义设置,SuspendCurrentQueueTimeMillis | 最大重试次数可通过自定义参数MaxReconsumeTimes取值进行配置。该参数取值无最大限制。若未设置参数值,默认最大重试次数为Integer.MAX |
无序消息消费 | 间隔时间根据重试次数阶梯变化,取值范围:1秒~2小时。不支持自定义配置 | 最大重试次数可通过自定义参数MaxReconsumeTimes取值进行配置。默认值为16次,该参数取值无最大限制,建议使用默认值 |
无序消息的重试间隔如下,可以看到与延迟消息第三个等级开始的时间完全一致:
第几次重试 | 与上次重试的间隔时间 | 第几次重试 | 与上次重试的间隔时间 |
---|---|---|---|
1 | 10s | 9 | 7min |
2 | 30s | 10 | 8min |
3 | 1min | 11 | 9min |
4 | 2min | 12 | 10min |
5 | 3min | 13 | 20min |
6 | 4min | 14 | 30min |
7 | 5min | 15 | 1h |
8 | 6min | 16 | 2h |
如果严格按照上述重试时间间隔计算,某条消息在一直消费失败的前提下,将会在接下来的 4 小时 46 分钟之内进行 16 次重试,超过这个时间范围消息将不再重试投递。
在老版本的 RocketMQ 中,一条消息无论重试多少次,这些重试消息的 Message Id 始终都是一样的。
但是在4.7.1版本之后,每次重试 MessageId 都会重建。
2.3、配置重试次数
消息队列 RocketMQ 允许 Consumer 启动的时候设置最大重试次数,重试时间间隔将按照如下策略:
- 最大重试次数小于等于 16 次,则重试时间间隔同上表描述。
- 最大重试次数大于 16 次,超过 16 次的重试时间间隔均为每次 2 小时。
那么问题来了,怎么在 SpringBoot 项目中自定义重试次数呢?
我们都知道消息的消费是通过实现 RocketMQListener
接口来监服务器托管网听的,那么只要在监听器里完成次数的配置就可以了:
@Component
@RocketMQMessageListener(
consumerGroup = "simple-group", //消费者组
topic = "retry-topic", //topic
selectorExpression = "tagA", //tag
maxReconsumeTimes = 2 //最大消息重试次数
)
public class RetryConsumerListener implements RocketMQListener {
@Override
public void onMessage(String message) {
System.out.println("receive retry message:" + message);
//此处抛出一个 RuntimeException 异常,模拟消费失败
throw new RuntimeException("故意抛出异常用于消息重试");
}
}
在消费消息的时候故意抛出异常,进而触发消息重试。由于重试次数设置的是 2,按照上面重试间隔的表格,第一次重试间隔是 10s, 第二次重试间隔是 30s。
查看控制台打印结果如下:
receive retry message:this is a retry message
2023-09-04 23:21:04.736 java.lang.RuntimeException: 故意抛出异常用于消息重试
//第一次重试
receive retry message:this is a retry message
2023-09-04 23:21:14.795 java.lang.RuntimeException: 故意抛出异常用于消息重试
//第二次重试
receive retry message:this is a retry message
2023-09-04 23:21:44.816 java.lang.RuntimeException: 故意抛出异常用于消息重试
结果几乎符合重试间隔(消息处理需要时间),那说明重试次数的配置是成功的。
通过 RocketMQ 控制台 Topic 一栏可以看到有一个重试消费者组 %RETRY%consumer_group,这个消费者组内存放的就是 consumer_group 消费者组消费失败重试的消息:
思考一个问题,当重试次数用完了,消息还是消费失败的时候,那消息应该作何处理?
这个问题就涉及到死信队列的知识点,不妨往下看。
3、死信队列
当一条消息初次消费失败,消息队列 RocketMQ 会自动进行消息重试;达到最大重试次数后,若消费依然失败,则表明消费者在正常情况下无法正确地消费该消息,此时,消息队列 RocketMQ 不会立刻将消息丢弃,而是将其发送到该消费者对应的特殊队列中。
在消息队列 RocketMQ 中,这种正常情况下无法被消费的消息称为死信消息(Dead-Letter Message),存储死信消息的特殊队列称为死信队列(Dead-Letter Queue)。
3.1、死信特性
死信消息具有以下特性:
- 不会再被消费者正常消费。
- 有效期与正常消息相同,均为 3 天,3 天后会被自动删除。因此,请在死信消息产生后的 3 天内及时处理。
死信队列具有以下特性:
- 一个死信队列对应一个 Group ID, 而不是对应单个消费者实例。
- 如果一个 Group ID 未产生死信消息,消息队列 RocketMQ 不会为其创建相应的死信队列。
- 一个死信队列包含了对应 Group ID 产生的所有死信消息,不论该消息属于哪个 Topic。
3.2、查看或发送死信消息
最简单查看死信消息的方式是 RocketMQ 控制台 Topic 界面。
死信队列 Topic 创建规则是是把消费者组的前缀加个 %DLQ%,如下所示:
选择重新发送消息
一条消息进入死信队列,意味着某些因素导致消费者无法正常消费该消息,因此,通常需要对其进行特殊处理。排查可疑因素并解决问题后,
可以看到 Topic 右边有 SEND MESSAGE 按钮,点击可以重新发送该消息,让消费者重新消费一次(前提是你监听了这个 Topic)。
最后一个主要的知识点是消息的幂等性,这里顺便简略带过。
4、消息幂等
消息队列 RocketMQ 消费者在接收到消息以后,有必要根据业务上的唯一 Key 对消息做幂等处理的必要性。
4.1、必要性
消息队列 RocketMQ 的消息有可能会出现重复,这个重复简单可以概括为以下情况:
-
发送时消息重复
当一条消息已被成功发送到服务端并完成持久化,此时出现了网络闪断或者客户端宕机,导致服务端对客户端应答失败。 如果此时生产者意识到消息发送失败并尝试再次发送消息,消费者后续会收到两条内容相同并且 Message ID 也相同的消息。
-
投递时消息重复
消息消费的场景下,消息已投递到消费者并完成业务处理,当客户端给服务端反馈应答的时候网络闪断。 为了保证消息至少被消费一次,消息队列 RocketMQ 的服务端将在网络恢复后再次尝试投递之前已被处理过的消息,消费者后续会收到两条内容相同并且 Message ID 也相同的消息。
-
负载均衡时消息重复(包括但不限于网络抖动、Broker 重启以及订阅方应用重启)
当消息队列 RocketMQ 的 Broker 或客户端重启、扩容或缩容时,会触发 Rebalance,此时消费者可能会收到重复消息。
4.2、处理方式
因为 Message ID 有可能出现冲突(重复)的情况,所以真正安全的幂等处理,不建议以 Message ID 作为处理依据。 最好的方式是以业务唯一标识作为幂等处理的关键依据,而业务的唯一标识可以通过消息 Key 进行设置:
发送消息时设置唯一的业务 key:
@RequestMapping("/containKeySend")
public void containKeySend() {
//指定Topic与Tag,格式: `topicName:tags`
SendResult sendResult = rocketMQTemplate.syncSend("key-topic:tagTest",
MessageBuilder
.withPayload("this is message")
// key:可通过key查询消息轨迹,如消息被谁消费,定位消息丢失问题。由于是哈希索引,须保证key尽可能唯一
.setHeader(MessageConst.PROPERTY_KEYS, "123456").build());
System.out.println("发送携带key消息:" + sendResult.toString());
}
消费者接收到 key 进行业务处理:
@Component
@RocketMQMessageListener(
consumerGroup = "key-group", //消费者组
topic = "key-topic", //topic
selectorExpression = "tagTest" //tag
)
public class KeyConsumerListener implements RocketMQListener {
@Override
public void onMessage(Message message) {
System.out.println("接收业务key:" + message.getKeys());
}
}
本篇讲了 Producer 及 Consumer 是怎么实现消息重试的,当然侧重点在于 Consumer,当重试次数超过配置之后,消息转成死信消息进入死信队列,描述了死信队列的查询及消息重发流程,最后是讲了消息的幂等性,通过设置唯一的 key 保证每一条消息的唯一性。
这个系列的文章到这里初步搞定,后续的文章有缘更新。
参考资料:
- https://www.cnblogs.com/crazymakercircle/p/15426300.html#autoid-h2-2-20-0
- https://juejin.cn/post/6989542586050412580#heading-96
- https://rocketmq.apache.org/zh/docs/
- https://blog.51cto.com/u_14861909/5505825
- https://github.com/apache/rocketmq-spring
服务器托管,北京服务器托管,服务器租用 http://www.fwqtg.net
机房租用,北京机房租用,IDC机房托管, http://www.fwqtg.net