如何使用RabbitMQ Quorum Queue实现数据复制及Spring Boot AMQP代码适配

1. 队列定义参数修改 #

Quorum队列要求声明时显式指定队列类型为quorum，对应两种常见声明方式的修改如下：

• 以Queue Bean形式声明队列
  原经典队列的简化写法：

@Bean
public Queue classicQueue() {
    return new Queue("test-queue", true);
}

修改为Quorum队列写法：

@Bean
public Queue quorumQueue() {
    return QueueBuilder.durable("test-queue")
            .quorum() // 底层自动注入x-queue-type=quorum核心参数
            // 可选配置：指定副本数量，默认值为5，集群节点不足5时自动取集群节点总数
            .withArgument("x-quorum-initial-group-size", 3)
            // 可选配置：限制队列最大存储字节数，避免队列溢出
            .withArgument("x-max-length-bytes", 1024 * 1024 * 1024)
            .build();
}

• 以@RabbitListener注解直接声明队列
  原经典队列写法：

@RabbitListener(queuesToDeclare = @Queue("test-queue"))

修改为Quorum队列写法：

@RabbitListener(queuesToDeclare = @Queue(name = "test-queue", durable = "true",
        arguments = @Argument(name = "x-queue-type", value = "quorum")))

2. 生产端配置调整 #

配合Quorum队列的高可用特性，建议开启生产者确认机制进一步降低消息丢失概率，在application.yml中添加配置：

spring:
  rabbitmq:
    publisher-confirm-type: correlated # 开启消息到达Broker的确认回调
    publisher-returns: true # 开启消息不可路由的返回回调
    template:
      mandatory: true # 消息不可路由时返回给生产者，不直接丢弃

生产端可对应添加确认回调逻辑，对发送失败的消息做降级处理。

3. 消费端配置调整 #

Quorum队列不支持无ACK模式，建议开启手动ACK确保消息消费成功再确认，在application.yml中添加配置：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual # 手动ACK，业务逻辑执行完成后主动确认消息
        prefetch: 10 # 限制单消费者预取消息数，避免单节点负载过高
        retry:
          enabled: true # 开启消费重试
          max-attempts: 3
          initial-interval: 1000ms

消费端逻辑需要对应调整：业务执行成功调用channel.basicAck()确认消息，执行失败可根据场景选择basicNack()让消息重入队列，或者投递到死信队列做后续处理。

4. 适配注意事项 #

• Quorum队列默认强制持久化，不支持非持久化、排他、TTL自动过期、优先级等经典队列特性，用到上述特性的业务需要提前调整逻辑
• 已存在的经典队列无法直接转为Quorum队列，需要先删除原有经典队列再重新声明，或者直接使用新的队列名称迁移流量

内容的提问来源于stack exchange，提问作者user666