Matarialize中文社区-RabbitMQ镜像队列（Mirrored Queue）的使用详解

RabbitMQ镜像队列（Mirrored Queue）的使用详解

激流勇进 2024-10-10 15:41:15

110 镜像队列,Mirrored Queue 专栏

什么是镜像队列？

在RabbitMQ中，镜像队列（Mirrored Queue）是一种特殊的队列类型，它的消息会在多个节点上复制。镜像队列提供了高可用性，确保在主节点宕机时，消费者可以从其他镜像节点继续消费消息。镜像队列适合需要高可用性的生产环境。

镜像队列的工作原理

主队列与镜像队列：在RabbitMQ中，镜像队列在一个或多个节点上创建副本。当主队列接收消息时，所有的副本也会接收到相同的消息。
节点选择：镜像队列的主节点和副本节点可以在RabbitMQ集群中的不同节点上配置，以增加冗余和容错能力。
故障转移：如果主队列的节点宕机，RabbitMQ会自动将副本节点提升为新的主节点，从而保证服务的连续性。

镜像队列的使用

1. 安装与配置RabbitMQ

确保你已安装RabbitMQ并启用管理插件。可以使用以下命令启用管理插件：

rabbitmq-plugins enable rabbitmq_management

接着，可以通过浏览器访问 http://localhost:15672 来管理RabbitMQ。

2. 创建镜像队列

可以使用RabbitMQ的管理界面或通过代码创建镜像队列。下面我们通过Node.js创建镜像队列。

安装依赖

首先，确保你的Node.js项目中安装了 amqplib：

npm install amqplib

创建镜像队列的代码示例

const amqp = require('amqplib');

async function createMirroredQueue() {
    const queueName = 'mirrored_queue';
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    // 设置队列参数，声明镜像队列
    await channel.assertQueue(queueName, {
        durable: true,
        // 将队列设置为镜像队列
        arguments: {
            'x-ha-policy': 'all' // 所有节点都会有这个队列的副本
        }
    });

    console.log(`创建镜像队列: ${queueName}`);

    // 关闭连接
    await channel.close();
    await connection.close();
}

createMirroredQueue().catch(console.error);

3. 向镜像队列发送消息

现在我们可以向镜像队列发送消息。以下是发送消息的示例代码：

async function sendMessage() {
    const queueName = 'mirrored_queue';
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    const message = 'Hello, Mirrored Queue!';
  
    // 发送消息
    channel.sendToQueue(queueName, Buffer.from(message), { persistent: true });
    console.log(`发送消息: ${message}`);

    // 关闭连接
    await channel.close();
    await connection.close();
}

sendMessage().catch(console.error);

4. 从镜像队列接收消息

接下来，我们需要从镜像队列中接收消息。以下是接收消息的代码示例：

async function receiveMessage() {
    const queueName = 'mirrored_queue';
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    await channel.assertQueue(queueName, { durable: true });

    console.log(`等待接收消息来自 ${queueName}...`);

    channel.consume(queueName, (msg) => {
        if (msg !== null) {
            console.log(`接收到消息: ${msg.content.toString()}`);
            // 手动确认消息
            channel.ack(msg);
        }
    }, { noAck: false });
}

receiveMessage().catch(console.error);

5. 镜像队列的所有属性及方法

在创建镜像队列时，我们可以使用一些特定的属性和方法：

durable：设置为true以确保队列在RabbitMQ重启后仍然存在。
arguments：可以设置队列的特定参数，例如：
- x-ha-policy：决定队列的镜像策略。可选值为：
  - all：所有节点都持有队列的副本。
  - nodes：指定某些节点持有副本。