RabbitMQ保证消息可靠性、防止重复消费

冷不防 2023-09-28 08:11 65阅读 0赞

### 场景 ###

比如未付款订单，超过一定时间后，系统自动取消订单并释放占有库存。

#### 常用解决方案： ####

rabbitmq的消息TTL和死信Exchange结合。

### 消息的TTL(Time To Live) ###

1、消息的TTL就是**消息的存活时间**。

2、RabbitMQ可以**对队列和消息分别设置TTL**

*  对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信.
 *  如果队列设置了，消息也设置了，那么会取小的。所以一个消息如果被路由到不同的队列中，这个消息死亡的时间有可能不一样（不同的队列设置）。这里单讲单个消息的TTL，因为它才是实现延迟任务的关键。可以通过设置消息的expiration字段或者x- message-ttl属性来设置时间，两者是一样的效果。

### 死信路由 ###

*  一个消息在满足如下条件下，会进死信路由，记住这里是路由而不是队列，一个路由可以对应很多队列。（什么是死信）
 *  Dead Letter Exchange其实就是一种普通的exchange，和创建其他exchange没有两样。只是在某一个设置Dead Letter Exchange的队列中有消息过期了，会自动触发消息的转发，发送到Dead Letter Exchange中去。
 *  **我们既可以控制消息在一段时间后变成死信，又可以控制变成死信的消息被路由到某一个指定的交换机，结合二者，其实就可以实现一个延时队列**。  
    ![在这里插入图片描述][e0821f5251fc467aace8067bdc8547c4.png]

### 如何保证消息可靠性 - 消息丢失 ###

#### 1、持久化 ####

RabbitMQ中发消息的时候会有个durable参数可以设置，设置为true，就会持久化

![在这里插入图片描述][b5c12c1ae4c24600a4972b1c4634aa29.png]  
这样的话MQ服务器即使宕机，重启后磁盘文件中有消息的存储，这样就不会丢失了吧。是的这样就一定概率的保障了消息不丢失。

但还会有个场景，就是消息刚刚保存到MQ内存中，但还没有来得及更新到磁盘文件中，突然宕机了。这个场景在持续的大量消息投递的过程中，会很常见。

那怎么办？我们如何作才能保障一定会持久化到磁盘上面呢？

#### confirm机制 ####

RabbitMQ利用confirm机制来通知我们是否持久化成功？

![在这里插入图片描述][bbd1bbdb5da547358bf280344de5b785.png]  
confirm机制的原理：  
（1）消息生产者把消息发送给MQ,如果接收成功，MQ会返回一个ack消息给生产者；  
（2）如果消息接收不成功，MQ会返回一个nack消息给生产者。

这样是不是就可以保障100%消息不丢失了呢？

**如果我们生产者每发一条消息，都要MQ持久化到磁盘中，然后再发起ack或nack的回调。这样的话是不是我们MQ的吞吐量很不高，因为每次都要把消息持久化到磁盘中。写入磁盘这个动作是很慢的。这个在高并发场景下是不能够接受的，吞吐量太低了。**

所以MQ持久化磁盘真实的实现，是通过异步调用处理的，他是有一定的机制，如：**等到有几千条消息的时候，会一次性的刷盘到磁盘上面。而不是每来一条消息，就刷盘一次**。

所以comfirm机制其实是一个**异步监听的机制**，是为了保证系统的高吞吐量，这样就导致了还是不能够100%保障消息不丢失，因为即使加上了confirm机制，消息在MQ内存中还没有刷盘到磁盘就宕机了，还是没法处理。

#### 数据库事务机制 ####

生产者在投递消息之前，可以在本地数据库建一张消息表，先把消息持久化到Redis或DB中，这样就可以利用本地数据库的事务机制。事务提交成功后，将消息表中的消息转移到消息队列中。

confirm机制监听消息是否发送成功？如ack成功消息，删除DB中此消息。

如果nack不成功的消息，这个可以根据自身的业务选择是否重发此消息。也可以删除此消息，由自己的业务决定。

### 如何避免消息重复消费 ###

幂等性也就是相同条件下对一个业务的操作，不管操作多少次，结果都是一样。

#### 重复消费的问题？ ####

导致重复消费的原因可能出现在生产者，也可能出现在MQ或消费者。

这里说的重复消费问题是指同一个数据被执行了两次，不单单指 MQ 中一条消息被消费了两次，也可能是 MQ 中存在两条一模一样的消费。

*  生产者：生产者可能会重复推送一条数据到 MQ 中，为什么会出现这种情况呢？也许是一个 Controller 接口被重复调用了 2 次，没有做接口幂等性导致的；也可能是推送消息到 MQ 时响应比较慢，**生产者的重试机制导致再次推送了一次消息**。
 *  MQ：在消费者消费完一条数据响应 ack 信号消费成功时，MQ 突然挂了，导致 MQ 以为消费者还未消费该条数据，MQ 恢复后再次推送了该条消息，导致了重复消费。
 *  消费者：消费者已经消费完了一条消息，正准备但是还未给 MQ 发送 ack 信号时，此时消费者挂了，服务重启后 MQ 以为消费者还没有消费该消息，再次推送了该条消息。

#### 如何保证幂等性？ ####

消费者怎么解决重复消费问题呢？这里提供两种方法：

*  **状态判断法**：消费者消费数据后把消费数据记录在 redis 中，下次消费时先到 redis 中查看是否存在该消息，存在则表示消息已经消费过，直接丢弃消息。
 *  **业务判断法**：通常数据消费后都需要插入到数据库中，使用数据库的唯一性约束防止重复消费。每次消费直接尝试插入数据，如果提示唯一性字段重复，则直接丢失消息。一般都是通过这个业务判断的方法就可以简单高效地避免消息的重复处理了。

[e0821f5251fc467aace8067bdc8547c4.png]: https://img-blog.csdnimg.cn/e0821f5251fc467aace8067bdc8547c4.png
[b5c12c1ae4c24600a4972b1c4634aa29.png]: https://img-blog.csdnimg.cn/b5c12c1ae4c24600a4972b1c4634aa29.png
[bbd1bbdb5da547358bf280344de5b785.png]: https://img-blog.csdnimg.cn/bbd1bbdb5da547358bf280344de5b785.png