Matarialize中文社区-RabbitMQ的队列类型

RabbitMQ的队列类型

激流勇进 2024-08-11 19:20:45

371 标准队列,死信队列,镜像队列专栏

在 RabbitMQ 中，队列是消息存储的核心组件，不同类型的队列可以满足不同的业务需求。以下是一些常见的 RabbitMQ 队列类型和它们的应用场景：

1. 标准队列（Classic Queue）

描述：这是 RabbitMQ 中最常见的队列类型，也是默认的队列类型。消息按照先进先出的顺序进行处理。
应用场景：适用于一般消息传递的场景，没有特殊的消息顺序或持久化需求。

2. 镜像队列（Mirrored Queue）

描述：镜像队列会在 RabbitMQ 集群的多个节点上同步消息。即使一个节点出现故障，其他节点也可以继续处理消息。
应用场景：适用于对高可用性有要求的场景，确保消息在节点故障时不丢失。

3. 死信队列（Dead Letter Queue, DLQ）

描述：死信队列用于处理无法正常消费的消息。这些消息可能由于多次重试失败、消息过期或者队列满载等原因被转移到死信队列中。
应用场景：适用于需要对无法处理的消息进行特殊处理或监控的场景。

配置示例：

const queueOptions = {
    deadLetterExchange: 'dlx', // 死信交换器
    deadLetterRoutingKey: 'dlx.routingKey' // 死信路由键
};
channel.assertQueue('myQueue', queueOptions);

4. 延时队列（Delayed Queue）

描述：延时队列允许消息在指定的时间之后才被消费者接收。这通常通过使用死信交换器或 RabbitMQ 插件（如 rabbitmq-delayed-message-exchange 插件）来实现。
应用场景：适用于需要在一段时间后才处理消息的场景，例如订单的延时支付、任务的定时执行。

使用 `x-delayed-message` 插件的配置示例：

安装插件（如果尚未安装）：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

配置延时队列：

const exchange = 'my-delayed-exchange';
const exchangeOptions = {
    arguments: {
        'x-delayed-type': 'direct' // 使用 direct 类型交换器
    }
};
channel.assertExchange(exchange, 'x-delayed-message', exchangeOptions);

const queue = 'myQueue';
const queueOptions = { durable: true };
channel.assertQueue(queue, queueOptions);

channel.bindQueue(queue, exchange, 'routingKey');

// 发送消息并设置延迟时间
const msg = 'Hello after delay';
const msgOptions = {
    headers: { 'x-delay': 5000 } // 延迟5秒
};
channel.publish(exchange, 'routingKey', Buffer.from(msg), msgOptions);

5. 优先级队列（Priority Queue）

描述：优先级队列允许消息根据优先级进行处理，高优先级的消息会被优先消费。
应用场景：适用于需要区分不同优先级任务的场景，比如在紧急任务和普通任务同时存在时，优先处理紧急任务。

配置示例：

const queueOptions = {
    arguments: {
        'x-max-priority': 10 // 设置最大优先级
    }
};
channel.assertQueue('priorityQueue', queueOptions);

const msg = 'High priority message';
const msgOptions = {
    priority: 9 // 设置消息优先级
};
channel.sendToQueue('priorityQueue', Buffer.from(msg), msgOptions);

6. 持久化队列（Durable Queue）

描述：持久化队列能够在 RabbitMQ 服务重启后保留队列中的消息。通过将队列和消息标记为持久化，确保消息不因服务重启而丢失。
应用场景：适用于需要确保消息不会丢失的重要任务。

配置示例：

channel.assertQueue('durableQueue', { durable: true });
const msg = 'Persistent message';
channel.sendToQueue('durableQueue', Buffer.from(msg), { persistent: true });

7. 分片队列（Quorum Queue）

描述：分片队列基于 Raft 共识算法，提供高可用性和数据一致性的保障。消息被存储在多个节点的日志中，以确保数据的安全性和一致性。
应用场景：适用于对一致性和可用性要求非常高的场景，特别是在分布式环境中。

配置示例：

const queueOptions = {
    arguments: {
        'x-queue-type': 'quorum' // 设置为 Quorum 队列
    }
};
channel.assertQueue('quorumQueue', queueOptions);

8. RPC 队列

描述：在远程过程调用（RPC）模式下，RabbitMQ 可以用作异步调用的消息传递媒介。客户端发送请求消息，服务器处理后返回响应消息。
应用场景：适用于分布式系统中需要异步 RPC 调用的场景。

配置示例：

// 客户端发送请求
channel.assertQueue('', { exclusive: true }, (err, q) => {
    const correlationId = generateUuid(); // 唯一的请求标识
    const msg = 'RPC request';

    channel.consume(q.queue, (msg) => {
        if (msg.properties.correlationId === correlationId) {
            console.log('Received:', msg.content.toString());
        }
    }, { noAck: true });

    channel.sendToQueue('rpc_queue', Buffer.from(msg), {
        correlationId: correlationId,
        replyTo: q.queue
    });
});