RocketMQ 消息重试与死信队列
RocketMQ 前面系列文章如下：
- RocketMQ系列(一) 基本介绍
- 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、无序消息的消费重试
对于无序消息（普通、定时、延时、事务消息），当消费者消费消息失败时，您可以通过设置返回状态达到消息重试的结果。
无序消息的重试只针对集群消费方式生效；广播方式不提供失败重试特性，即消费失败后，失败消息不再重试，继续消费新的消息。
顺序消息与无序消息的消费参数差别如下：
无序消息的重试间隔如下，可以看到与延迟消息第三个等级开始的时间完全一致：
如果严格按照上述重试时间间隔计算，某条消息在一直消费失败的前提下，将会在接下来的 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<String> {
@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<Message> {
@Override
public void onMessage(Message message) {
System.out.println("接收业务key:" + message.getKeys());
}
}
本篇讲了 Producer 及 Consumer 是怎么实现消息重试的，当然侧重点在于 Consumer，当重试次数超过配置之后，消息转成死信消息进入死信队列，描述了死信队列的查询及消息重发流程，最后是讲了消息的幂等性，通过设置唯一的 key 保证每一条消息的唯一性。
这个系列的文章到这里初步搞定，后续的文章有缘更新。
参考资料: