仲裁队列 Quorum Queue

仲裁队列（Quorum Queue）是RabbitMQ 3.8引入的新型队列类型，基于Raft共识算法实现多节点数据一致性，替代传统的镜像队列提供更可靠的高可用方案。

定义

仲裁队列使用Raft协议在多个节点间复制队列数据。写入操作需获得多数节点（Quorum）确认后才返回成功，确保在网络分区或部分节点宕机时数据不丢失且保持一致。

Maven依赖

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.20.0</version>
</dependency>

配置与示例

声明仲裁队列

import com.rabbitmq.client.*;

import java.util.HashMap;
import java.util.Map;

public class QuorumQueueExample {
    private static final String QUEUE_NAME = "quorum.orders";
    private static final String EXCHANGE_NAME = "quorum_exchange";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // 声明仲裁队列（关键参数 x-queue-type = quorum）
        Map<String, Object> queueArgs = new HashMap<>();
        queueArgs.put("x-queue-type", "quorum");
        
        // 可选：设置最大复制节点数（默认3）
        // queueArgs.put("x-quorum-initial-group-size", 3);
        
        channel.queueDeclare(QUEUE_NAME, true, false, false, queueArgs);

        // 声明交换机并绑定
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true);
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "orders");

        // 发送持久化消息
        String message = "order-QQ-001";
        channel.basicPublish(EXCHANGE_NAME, "orders",
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            message.getBytes("UTF-8"));
        System.out.println("发送: " + message);

        // 消费消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String msg = new String(delivery.getBody(), "UTF-8");
            System.out.println("收到: " + msg);
            // 仲裁队列必须显式ACK
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };

        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
    }
}

查看仲裁队列状态

# 查看队列类型
rabbitmqctl list_queues name type

# 输出示例：
# quorum-orders    quorum
# classic-queue    classic

# 查看仲裁队列的Raft成员
rabbitmqctl eval 'rabbit_amqp_queue:list_quorum_queues().'

# 查看仲裁队列统计信息
rabbitmqctl list_queues name messages message_stats

Raft共识机制

Client 写入请求
      │
      ▼
┌──────────────────────────────────────────────┐
│                  Leader (Node-A)              │
│  1. 接收写入请求                              │
│  2. 追加到本地Raft Log                        │
│  3. 向Follower复制日志                        │
└──────────────────────────────────────────────┘
      │              │              │
      ▼              ▼              ▼
┌─────────┐    ┌─────────┐    ┌─────────┐
│ Node-A  │    │ Node-B  │    │ Node-C  │
│ Leader  │    │ Follower│    │ Follower│
│ ✓ ACK   │    │ ✓ ACK   │    │         │
└─────────┘    └─────────┘    └─────────┘
      ▲              │
      └───── Quorum ─┘
       (多数确认: 2/3)
      │
      ▼
  返回Client成功

• 写入需获得多数（Quorum）节点确认才返回成功
• 3节点集群容忍1个节点故障，5节点容忍2个
• Leader负责处理所有读写，Follower仅同步数据
• Leader选举基于Raft协议，自动完成

仲裁队列不支持Lazy Queue模式（x-queue-mode=lazy）。仲裁队列本身已将消息持久化到磁盘，无需额外Lazy模式。若需降低内存使用，应增加节点数而非改变模式。

注意事项

1. 最小节点数：仲裁队列至少需要3个节点（1 Leader + 2 Follower）才能形成有效Quorum。

2. 故障容限：N个节点的仲裁队列最多容忍(N-1)/2个节点故障。3节点容忍1个，5节点容忍2个。

3. 性能特征：写入延迟取决于最慢的多数节点。跨机房部署时网络延迟直接影响写入吞吐量。

4. 不兼容特性：仲裁队列不支持优先级队列（x-max-priority）、DLX（x-dead-letter-exchange）等经典队列特性。需使用插件或代码层处理。

5. 升级建议：RabbitMQ官方推荐新项目使用仲裁队列替代镜像队列。经典镜像队列将在未来版本中移除。

6. 消费者连接：消费者连接任意节点均可，Raft Leader自动处理消费请求并同步到Follower。

要点总结

• 仲裁队列基于Raft共识算法，写入需多数节点确认才返回成功
• 通过x-queue-type=quorum参数声明，默认复制3个节点
• 3节点集群容忍1节点故障，5节点容忍2节点
• 不支持Lazy模式、优先级队列等经典队列特性
• 官方推荐替代镜像队列，新项目应优先使用仲裁队列