Rabbitmq 镜像队列

RabbitMQ 支持高可用性队列（HA Queues），可以在多个节点之间复制队列，确保即使某个节点失败，消息仍然可用。将 RabbitMQ 部署为集群，确保高可用性和负载均衡。

RabbitMQ 的镜像队列集群（Mirrored Queues Cluster）是实现高可用性的一种模式，它通过在集群中的多个节点之间复制队列，确保消息在集群的某些节点宕机时不会丢失，并且队列仍然可以正常工作。镜像队列集群广泛用于需要保证消息可靠性的业务场景，如金融、支付、订单处理等关键任务。

在镜像队列集群中，队列的主副本（主队列）会存在于一个节点上，称为主队列节点（Master）。同时，这个队列会被复制到集群中的其他节点，这些副本称为镜像（Mirrors）。

核心工作机制

• 主队列（Master Queue）：
  • 队列的原始副本，所有的消息发布、消费、ACK 都是在主队列上处理的。
  • 每当有消息发送到主队列时，主队列会将消息同步到其镜像节点。
• 镜像队列（Mirrored Queue）：
  • 镜像队列是主队列的完全副本，存在于集群的其他节点上，会实时同步主队列中的所有消息和状态。这些镜像队列与主队列保持同步，以确保所有消息和状态在每个镜像中都相同。
  • 如果主队列所在的节点发生故障，集群中的其中一个镜像队列会自动提升为新的主队列，并继续提供服务。
• 故障转移（Failover）：
  • 当主队列的节点发生故障时，RabbitMQ 会自动从剩下的镜像队列中选择一个提升为主队列。此过程通常是无缝的，在故障转移期间，消费和生产可能会有短暂的中断，但当新的主队列节点被选定后，消息处理会恢复正常。消费者和生产者可以继续与新的主队列通信。
  • 故障转移后，新的主队列会自动在其他节点上创建新的镜像，以保持高可用性。
• 镜像的自动管理：
  • RabbitMQ 可以根据策略自动管理镜像队列的数量和分布。可以配置镜像策略，控制镜像队列的创建、复制的节点数量等。

镜像队列集群的配置
镜像队列集群的配置通常通过 策略（Policy） 来实现。可以通过 RabbitMQ 管理界面或命令行工具 rabbitmqctl 来定义镜像队列的策略。
使用 rabbitmqctl 命令创建一个策略，将队列的镜像复制到集群的所有节点上。

rabbitmqctl set_policy ha-all "^ha\." '{"ha-mode":"all"}'

• ha-all: 策略名称。
• ^ha.: 队列名称的正则表达式，所有以 ha. 开头的队列都会应用此策略。
• {“ha-mode”:“all”}: 表示所有节点都会有该队列的镜像。
• ha-mode：
  • all: 将队列镜像到集群中的所有节点上。
  • exactly: 将队列镜像到集群中的指定数量的节点上。
  • nodes: 选择特定的节点进行队列镜像。
• ha-sync-mode:
  • automatic: 当新的镜像节点加入时，自动同步主队列中的消息到新镜像节点。
  • manual: 需要手动执行同步操作。

如果不希望将队列镜像到集群的所有节点，而是只希望将其复制到指定数量的节点，可以使用 ha-mode 的 exactly 选项。
假设有 3 个节点的 RabbitMQ 集群（节点 A、B、C），可以配置队列只在两个节点上进行镜像，如下：

rabbitmqctl set_policy ha-two "^ha\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

• 这会将所有以 ha. 开头的队列镜像到集群的任意两个节点上。
• ha-mode: 使用 exactly 来指定镜像的数量。
• ha-params: 表示该队列会被镜像到集群中两个节点上。
• ha-sync-mode: 使用 automatic 模式，表示镜像队列会自动与主队列同步。

也可以选择将镜像队列只复制到特定节点上，通过 nodes 参数指定节点。

rabbitmqctl set_policy ha-nodes "^ha\." '{"ha-mode":"nodes","ha-params":["rabbit@node1","rabbit@node2"]}'

• ha-mode: 使用 nodes 模式，将队列镜像到 rabbit@node1 和 rabbit@node2 上。

镜像队列会增加集群的网络带宽和存储开销，因为每一条消息都要在多个节点之间同步。应根据业务场景合理设置镜像的节点数量，平衡高可用性和性能。

优点：

• 高可用性：当某个节点故障时，镜像队列可以迅速接管，确保队列和消息的可用性。
• 无缝故障转移：RabbitMQ 会自动进行故障转移，无需额外的操作或维护。
• 消息可靠性：通过在多个节点上同步队列和消息，确保了消息不会因为节点故障而丢失。

缺点：

• 性能损耗：由于镜像队列需要同步消息和状态，增加了网络、磁盘和内存的负载，可能会影响消息处理的性能。
• 配置复杂性：在集群规模增大时，需要考虑如何合理配置镜像策略，避免不必要的资源消耗；维护镜像队列集群需要更复杂的集群管理和监控，尤其是在节点数量增加时。
• 分区问题：如果集群中存在网络分区，可能会导致队列数据不一致，镜像队列可能在不同分区中各自成为主队列，带来数据冲突问题。
• 扩展性受限：随着集群中节点数量的增加，同步的负载会成倍增加，因此不适合非常大规模的集群。

Spring Boot 项目中配置镜像队列
在 Spring Boot 项目中配置 RabbitMQ 镜像队列（Mirrored Queues）进行消息的收发，需要在 RabbitMQ 中设置队列为高可用队列，并在 Spring Boot 应用中进行相应的配置。

镜像队列的核心特性是，消息会被复制到集群中的多个节点（主节点 + 副节点），从而实现高可用性。其工作原理如下：

• 主节点（Master Node）负责处理所有入队和出队的操作。
• 副节点（Mirror Node）仅同步主节点的数据，并在主节点不可用时，自动切换成为新的主节点。

当主节点挂掉后，RabbitMQ 会通过仲裁机制，自动选择一个副节点作为新的主节点。此时，消息消费者和生产者会自动与新的主节点进行通信，而无需手动更改配置。

在 Spring Boot 中，可以通过配置 RabbitMQ 的镜像队列集群，使应用程序在主节点宕机后自动切换到副节点。为此，最常见的做法是配置 RabbitMQ 的多个节点地址或使用一个负载均衡的 DNS 入口来实现故障切换。

使用多个节点地址 (addresses)
Spring Boot 提供了 addresses 配置项，可以指定多个 RabbitMQ 节点地址。当其中一个节点不可用时，应用程序会自动尝试连接其他节点，而无需手动修改配置文件。这是最简单和直接的方式来实现高可用性。

spring:rabbitmq:addresses: host1:5672,host2:5672,host3:5672  # 配置多个 RabbitMQ 集群节点port: 5672username: guestpassword: guestvirtual-host: /listener:simple:concurrency: 3max-concurrency: 10connection-timeout: 15000queues:- name: mirroredQueuedurable: trueexclusive: falseauto-delete: falsearguments:x-ha-policy: all

在这个配置中，addresses 通过逗号分隔的多个 host:port 配置，告诉 Spring Boot 的 RabbitMQ 客户端尝试连接多个节点。如果 host1 宕机，客户端会自动尝试连接 host2，然后是 host3。当其中某个节点不可用时，Spring Boot 会自动切换到下一个可用节点，避免手动修改配置文件。

如果 RabbitMQ 集群通过 DNS 提供了一个负载均衡的入口地址，可以使用该 DNS 入口来实现节点的自动切换。负载均衡 DNS 可以自动将请求路由到可用的 RabbitMQ 节点，并在节点故障时自动进行切换。

spring:rabbitmq:host: rabbitmq-cluster.example.com  # 指定 RabbitMQ 集群的负载均衡 DNS 名称port: 5672                          # RabbitMQ 默认端口username: guestpassword: guestvirtual-host: /listener:simple:concurrency: 5max-concurrency: 10connection-timeout: 15000

在这个配置中，host 被设置为 RabbitMQ 集群的 DNS 名称（例如 rabbitmq-cluster.example.com）。这个 DNS 名称应该指向 RabbitMQ 集群中的所有节点，并能够处理主节点宕机时的自动切换。

创建一个 RabbitMQ 配置类，配置高可用队列（镜像队列）：

• 当镜像队列被正确配置（如 x-ha-policy: all），消息会自动复制到副节点。
• 当主节点故障时，副节点会被提升为新的主节点，继续处理消息的发送和消费。
• Spring Boot 中的 RabbitMQ 连接使用 spring-rabbit 和 AMQP 协议库，它内置了自动重连机制，当连接的主节点宕机时，Spring Boot 应用程序会尝试重新连接到集群中的其他节点，重新建立连接。

@Configuration
@EnableRabbit
public class RabbitConfig {@Beanpublic Queue mirroredQueue() {// 创建一个高可用队列（镜像队列）return new Queue("mirroredQueue", true, false, false, Map.of("x-ha-policy", "all"));}
}

创建一个消息发送者，用于发送消息到镜像队列：

@Service
public class MessageProducer {@Autowiredprivate RabbitTemplate rabbitTemplate;public void sendMessage(String message) {rabbitTemplate.convertAndSend("mirroredQueue", message);System.out.println("Sent message: " + message);}
}

创建一个消息接收者，用于接收来自镜像队列的消息：

@Component
public class MessageListener {@RabbitListener(queues = "mirroredQueue")public void receiveMessage(String message) {System.out.println("Received message: " + message);}
}

监控：使用 RabbitMQ 管理插件或其他监控工具，监控队列的状态、消费者的处理情况和系统的性能指标。
告警：设置告警机制，当消息堆积或消费者处理速度缓慢时，及时通知相关人员。