【学习笔记】RabbitMQ-6 消息的可靠性投递2

参考资料

• RabbitMQ官方网站
• RabbitMQ官方文档
• 噼咔噼咔-动力节点教程

十一、队列Queue的消息属性

11.1 具体属性

1. Type：类型，只需要关注classic
2. Name：名称
3. Durability：持久化
4. Auto Delete：是否自动删除
5. Arguments：参数

11.2 自动删除

• 当最后一个消费者断开连接后，队列将会删除
• 一般不会用，存在数据丢失的风险

11.2 自定义参数

11.2.1 Message TTL 消息存活时间

• 官方解释
• 定义了一个消息再被推送后，再被丢弃之前，可以存活的时间
• 单位：毫秒
• 参数格式 ：“x-message-ttl” ： number
• 如果消息在指定的时间段内未被消费，则该消息将被标记为过期并被丢弃。
• 这可以确保不再需要的消息不会一直存在于队列中，从而占据资源。

11.2.2 Auto expire 队列自动到期时间

• 官方解释x-expires

• 在自动删除队列之前，队列可以闲置多长时间（How long a queue can be unused for before it is automatically deleted）

• 定义了队列自动过期的时间。

• 参数格式 “x-expires”：number

• 如果队列在指定的时间段内未被使用，则该队列将被自动删除。

• 这可以确保不再需要的队列不会一直存在于RabbitMQ服务器上，从而占据资源。

11.2.3 Overflow behaviour 溢出行为

• 官方解释 queue overflow behaviour
• overflow behavior参数用于定义当队列达到最大容量时的处理方式。
• 参数格式：“x-overflow”：string
• 在RabbitMQ中，有以下几种可选的overflow behavior：
  • drop-head: 当队列达到最大容量时，新的消息将被丢弃，并且最早进入队列的消息会被删除，以腾出空间给新的消息。
  • reject-publish: 当队列达到最大容量时，新的消息将被拒绝发送，并且发布者将收到一个拒绝通知。这种方式可以让发布者有机会处理无法发送的消息。
  • reject-publish-dlx: 当队列达到最大容量时，新的消息将被拒绝发送，并且发布者将收到一个拒绝通知。同时，被拒绝的消息会被发送到一个死信交换器（Dead-Letter Exchange），以便后续进行处理。
  • reject-subscribe: 当队列达到最大容量时，新的订阅者将无法成功订阅该队列，并且会收到一个拒绝通知。这种方式可以让订阅者有机会处理无法接收的消息。

11.2.4 Single active consumer单一消费者模式

• Single active consumer（单一活动消费者）是一种消息队列的消费模式。
• 在这种模式下，一个队列只能有一个活动的消费者来处理消息，而其他消费者处于非活动状态。
• 参数格式“x-single-active-consumer”：bool
• 使用Single active consumer模式可以确保消息的顺序性和可靠性。
  • 当只有一个消费者活动时，消息将按照顺序被处理，避免了多个消费者并发处理消息可能引起的顺序混乱问题。
  • 此外，由于只有一个消费者在处理消息，可以减少并发操作带来的资源竞争和冲突。
• Single active consumer模式也存在一些限制。
  • 于只有一个消费者在处理消息，如果该消费者出现故障或变得不可用，整个消息队列将无法被处理。
  • 因此，在设计系统时需要考虑到这种单点故障的风险，并采取相应的容错和监控机制来保证系统的可用性。

11.2.5 Dead letter exchange死信交换机和 Dead letter routing key死信路由key

两者配合可以指定DLX发送的交换机和键，之前已经研究过就不做赘述了

11.2.6 Max length队列最大信息数 和Max length bytes 队列磁盘大小

• Max length参数用于限制队列中消息的最大数量。当队列中的消息数达到最大值时，新的消息将被拒绝并返回给发布者。

• Max length bytes参数用于限制队列中消息的最大总大小（磁盘空间。当队列中所有消息的大小总和达到最大值时，新的消息将被拒绝并返回给发布者。

这两个参数可以用于控制队列的大小，以避免队列过度增长导致系统资源耗尽。在设置这些参数时，需要根据具体的业务需求和系统资源情况进行权衡和调整。

• 注意，Max length bytes的单位是byte。比如要设置1G的最大磁盘空间

  'x-max-length-bytes': 1073741824
  1

11.2.7 Maximum priority最高优先级

• 当消息被发布到队列时，可以为消息设置一个优先级，优先级越高的消息将会被先处理。
• 使用Maximum priority参数可以限制队列中消息的最大优先级，超过该优先级的消息将被丢弃或者被发送到死信交换器（Dead-Letter Exchange）。

11.2.8 Lazy mode懒模式

• 用于延迟队列的内存分配，从而减少内存的使用量。

• 在Lazy mode下，队列的消息不会立即被写入磁盘，而是先被存储在内存中，直到内存达到一定的阈值时才会被写入磁盘。

  默认情况下，消息会一直存在内存中，占用过多内存可能会导致运行过慢、消息丢失、系统崩溃等问题

• 使用Lazy mode可以在一定程度上降低队列的内存使用量，提高系统的性能和可扩展性。

• 然而，由于消息需要等待一定时间才能被写入磁盘，因此在使用Lazy mode时需要注意消息的持久化和数据丢失的风险。如果需要确保消息不会丢失，需要将消息设置为持久化，并且在队列中启用持久化模式。

• 参数配置格式 “x-queue-mode” ： “lazy”

11.2.9 Master locator主定位器

用于集群，暂不涉及

RabbitMQ的Master locator用于在集群中查找当前拥有某个队列的主节点（Master Node）。在RabbitMQ集群中，一个队列可以被复制到多个节点上，其中一个节点被指定为主节点，其他节点为从节点。主节点负责处理所有的读写请求，而从节点只负责复制主节点上的数据。

十二、消息的可靠性投递

12.1 概念理解

消息的可靠性投递是指在消息传递过程中，确保消息能够被成功地传递到目标节点并被消费者正确地处理。

在消息传递中，可能会发生网络故障、节点宕机等问题，这些问题可能会导致消息丢失或重复传递，从而影响系统的可靠性和稳定性。

12.2 如何提高消息的可靠性

注意

消息的可靠性投递就是要保证消息投递过程中每一个环节都要成功，那么这肯定会牺牲些性能，性能与可靠性是无法兼得的

如果业务实时一致性要求不是特别高的场景，可以牺牲一些可靠性来换取性能。

消息的传递模型如下

12.2.1 消息发送时的确认模式

对应第一步的生产者到交换机的这一步，确保消息成功发送到交换机

12.2.2 消息返回模式

对应的是交换机到队列这一步，确保消息成功从交换机发送到队列中

当然在实际生产环境下，我们不会出现这种问题，必须对所有的name和key进行严格测试才能上线(很少有这种问题 ) ;

1.2.2.3 持久化机制

交换机和队列的持久化机制，保证服务器故障或重启后，消息仍然存在

• 队列持久化
• 交换机持久化
• 消息持久化

12.2.4 消费时的确认模式

启动手动确认模式，前面有提到。

只有当消费者确保业务运行成功后，再确认接受消息。

12.2.5 集群、镜像队列，保证高可用

• 设置rabbitMQ的集群
• 设置镜像队列

十三、消息的幂等性

同一个消息，第一次接收，正常处理业务；

如果同一个消息，第二次接收，那就不能再处理了，否则就重复处理了。

13.1 概念

13.1.1 幂等性

概念：对一个资源，无论请求一次还是多次，该数据本身造成的影响应该是相同的，不能因为重复的请求而造成资源重复造成影响。

（个人理解：有点类似数据库的一致性

13.1 举例:接口幂等性

接口幂等性是指：一个接口用同样的参数反复调用，不会造成业务错误，那么这个接口就是具有幂等性的；

这些接口必须要要做幂等设计：

• 注册接口：多次请求注册，也只能生成同一个新账号
• 发送短信验证码接口：多次请求也只发送一次短信
• 比如同一个订单我支付两次，但是只会扣款一次，第二次支付不会扣款，这说明这个支付接口是具有幂等性的;

13.2 数据库的幂等性

问：sql语句中哪些语句是非幂等的？

• select：多次查询结果一样：幂等
• delete：多次删除结果也是一样：幂等
• update：不进行计算的更新操作：幂等
• insert：除非指定id，多次的插入会导致多条数据，所以是非幂等的。

所以需要对insert的进行幂等操作，比如数据库自带的约束。或者在插入前进行查询，避免重复插入。

13.2 如何避免消息的重复消费问题？（消息消费时的幂等性

13.1 全局唯一id +Redis

当生产者发送消息时，设置一个全局唯一的messageId，消费者拿到消息后，使用redis的setnx指令存储messageId，如果存储成功（返回为1）说明消息还为被处理，如果返回为0，则说明已经被处理过了。

关于SETNX

SETNX 指令常用于实现分布式锁的场景，通过尝试设置一个特定的键来获取锁，如果设置成功则表示获取到了锁，否则表示锁已经被其他客户端持有。

SETNX key value
1

当执行 SETNX 指令时，会按照以下步骤进行处理：

1. 如果键 key 不存在，则设置键 key 的值为 value。
2. 如果键 key 已经存在，则不进行任何操作，返回 0。
3. 如果设置成功，则返回 1。

13.3 ⭐️代码实现

13.3.1 环境准备

• redis

• 项目依赖新增：

  spring-boot-starter-redis
  1

13.3.2 实例思路整理

生产者：

1. 模拟一个订单对象，其中一个属性为订单id（唯一）
2. 发送两个订单，他们的订单id相同（模拟非幂等情况

消费者：

1. 开启手动确认模式：spring.rabbitmq.listener.simple.acknowledge-mode = manual
2. 正常监听队列，接收到订单信息后，将订单id存入redis
3. 判断redis的存储结果，如果正常存入redis，则确认接受信息并继续进行消费者业务。否则拒绝信息。

代码略