2.6 队列 (Queues)

文档摘要

2.6 队列 (Queues) 2.6 队列 (Queues): RabbitMQ 消息传递的核心载体在 RabbitMQ 的世界里，队列 (Queues) 是消息传递的核心组件。它们就像消息的邮局，负责接收生产者 (Producers) 发送的消息，并将这些消息存储起来，直到消费者 (Consumers) 连接并准备好处理它们。理解队列的概念和特性对于掌握 RabbitMQ 至关重要。 2.6.1 队列的概念与作用概念: 队列本质上是一个消息缓冲区。它可以被看作是一个有序的消息集合，遵循 FIFO (First-In, First-Out) 原则，即先进入队列的消息先被取出和处理。

2.6 队列 (Queues)

2.6 队列 (Queues): RabbitMQ 消息传递的核心载体

在 RabbitMQ 的世界里，队列 (Queues) 是消息传递的核心组件。它们就像消息的邮局，负责接收生产者 (Producers) 发送的消息，并将这些消息存储起来，直到消费者 (Consumers) 连接并准备好处理它们。理解队列的概念和特性对于掌握 RabbitMQ 至关重要。

2.6.1 队列的概念与作用

概念:

队列本质上是一个消息缓冲区。它可以被看作是一个有序的消息集合，遵循 FIFO (First-In, First-Out) 原则，即先进入队列的消息先被取出和处理。在 RabbitMQ 中，队列是消息的最终目的地，生产者将消息发送到交换机 (Exchange)，交换机根据路由规则将消息路由到一个或多个队列中。消费者则从队列中订阅并接收消息进行处理。

作用:

队列在 RabbitMQ 中扮演着至关重要的角色，主要体现在以下几个方面：

解耦 (Decoupling): 队列作为生产者和消费者之间的中间层，实现了生产者和消费者的解耦。生产者无需知道消费者的存在，也无需关心消费者如何处理消息。消费者也无需直接与生产者交互，只需从队列中获取消息即可。这种解耦提高了系统的灵活性和可维护性。
异步处理 (Asynchronous Processing): 生产者将消息发送到队列后即可立即返回，无需等待消费者处理完成。消费者可以在稍后的时间异步地从队列中取出消息进行处理。这大大提高了系统的响应速度和吞吐量。
流量削峰 (Traffic Shaping/Buffering): 当生产者产生消息的速度远高于消费者处理消息的速度时，队列可以起到缓冲的作用，平滑消息流量，避免系统因突发流量而崩溃。
消息持久化 (Message Persistence): RabbitMQ 允许将队列和消息设置为持久化，即使 RabbitMQ 服务器重启，队列和消息也不会丢失，保证了消息的可靠性。
消息路由与过滤 (Message Routing and Filtering): 通过交换机 (Exchange) 和绑定 (Binding) 机制，消息可以根据不同的路由规则被路由到不同的队列，实现消息的灵活路由和过滤。
消息确认 (Message Acknowledgement): RabbitMQ 提供了消息确认机制，确保消息被消费者正确处理，避免消息丢失。

2.6.2 队列的特性详解

RabbitMQ 队列具有多种重要的特性，这些特性决定了队列的行为和适用场景。

1. 持久性 (Durable Queues)

默认情况下，RabbitMQ 的队列是非持久化的，这意味着当 RabbitMQ 服务器重启时，队列会被删除。为了保证队列的可靠性，我们可以将队列声明为持久队列 (Durable Queues)。

特性: 持久队列的元数据（队列名称、属性等）会被存储在磁盘上。当 RabbitMQ 服务器重启后，队列的元数据会被重新加载，队列得以恢复。
声明方式: 在声明队列时，将 durable 参数设置为 true。
代码示例 (Python - pika):


import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# 声明持久队列， durable=True
channel.queue_declare(queue='durable_queue', durable=True)
print(" [*] Waiting for messages. To exit press CTRL+C")
def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()}")
channel.basic_consume(queue='durable_queue', on_message_callback=callback, auto_ack=True)
channel.start_consuming()

2. 自动删除 (Auto-delete Queues)

自动删除队列会在满足特定条件时自动被删除。

特性: 当最后一个消费者取消订阅队列 (通过 basic.cancel) 或者与队列断开连接时，并且队列中不再有未消费的消息，自动删除队列会被 RabbitMQ 服务器自动删除。
声明方式: 在声明队列时，将 auto_delete 参数设置为 true。
适用场景: 适用于临时队列，例如用于响应式请求/回复模式中的临时回复队列。
代码示例 (Python - pika):


import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# 声明自动删除队列， auto_delete=True
channel.queue_declare(queue='auto_delete_queue', auto_delete=True)
print(" [*] Waiting for messages. To exit press CTRL+C")
def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()}")
channel.basic_consume(queue='auto_delete_queue', on_message_callback=callback, auto_ack=True)
channel.start_consuming()

3. 独占性 (Exclusive Queues)

独占队列只允许被声明它的连接所访问。

特性: 独占队列只能被声明它的连接上的消费者消费。其他连接无法访问该队列。当声明队列的连接关闭时，独占队列会被自动删除（即使它不是自动删除队列）。
声明方式: 在声明队列时，将 exclusive 参数设置为 true。
适用场景: 适用于需要确保队列只能被特定消费者访问的场景，例如安全性要求较高的场景。
代码示例 (Python - pika):


import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# 声明独占队列， exclusive=True
channel.queue_declare(queue='exclusive_queue', exclusive=True)
print(" [*] Waiting for messages. To exit press CTRL+C")
def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()}")
channel.basic_consume(queue='exclusive_queue', on_message_callback=callback, auto_ack=True)
channel.start_consuming()

4. 消息持久性 (Message Persistence)

队列的持久性只保证队列的元数据在服务器重启后不会丢失，但默认情况下，发送到队列的消息是非持久化的，服务器重启时消息会丢失。为了保证消息的可靠性，我们需要将消息也设置为持久化。

特性: 持久消息会被写入磁盘，当 RabbitMQ 服务器重启后，持久消息会被重新加载，不会丢失。
发送方式: 在发送消息时，需要设置消息属性的 delivery_mode 为 2 (表示持久化)。
代码示例 (Python - pika):


import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='persistent_queue', durable=True) # 队列需要持久化
message = 'This is a persistent message!'
# 发送持久消息， delivery_mode=2
channel.basic_publish(exchange='',
                      routing_key='persistent_queue',
                      body=message,
                      properties=pika.BasicProperties(
                         delivery_mode=2, # make message persistent
                      ))
print(f" [x] Sent '{message}'")
connection.close()

注意: 消息持久化会降低消息吞吐量，因为它涉及到磁盘 I/O 操作。需要根据实际应用场景权衡性能和可靠性。

5. 消息确认 (Message Acknowledgement)

为了确保消息被消费者正确处理，RabbitMQ 提供了消息确认机制 (Acknowledgement，简称 ACK)。

特性: 消费者在成功处理完消息后，需要向 RabbitMQ 服务器发送 ACK 确认。如果消费者在处理消息过程中发生异常或者未发送 ACK，RabbitMQ 会认为消息处理失败，并将消息重新放入队列，等待其他消费者重新消费 (或者发送到死信队列，如果配置了死信队列)。
确认模式: 消息确认模式有两种：
- 自动确认 (Automatic Acknowledgement): 消费者一旦接收到消息，RabbitMQ 就会立即将消息标记为已发送。即使消费者处理消息失败，消息也无法被重新消费，容易造成消息丢失。
- 手动确认 (Manual Acknowledgement): 消费者在成功处理完消息后，需要手动调用 channel.basic_ack() 方法发送 ACK 确认。只有收到 ACK 后，RabbitMQ 才会将消息从队列中移除。手动确认模式可以保证消息的可靠性，但需要消费者代码显式处理 ACK。
代码示例 (Python - pika - 手动确认):


import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
print(" [*] Waiting for messages. To exit press CTRL+C")
def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()}")
    time.sleep(body.count(b'.')) # 模拟耗时任务
    print(" [x] Done")
    ch.basic_ack(delivery_tag=method.delivery_tag) # 手动发送 ACK
channel.basic_qos(prefetch_count=1) # 每次只预取一条消息，公平分发
channel.basic_consume(queue='task_queue', on_message_callback=callback, auto_ack=False) # auto_ack=False 关闭自动确认
channel.start_consuming()

6. 队列类型

RabbitMQ 主要支持两种类型的队列：

经典队列 (Classic Queues): RabbitMQ.x 之前的默认队列类型。经典队列使用单个进程处理消息，性能受限于单个节点的性能。
Quorum 队列 (Quorum Queues): RabbitMQ.x 引入的新型队列类型，基于 Raft 一致性算法构建，具有更高的可靠性和数据安全保障，即使在节点故障的情况下也能保证消息的可靠传递。Quorum 队列通常用于对数据一致性要求更高的场景。

选择队列类型: 经典队列适用于大多数场景，性能较高。Quorum 队列适用于对数据可靠性要求极高的场景，例如金融交易、关键业务系统等。

7. 队列属性 (Queue Arguments)

在声明队列时，可以设置一些可选的属性 (Arguments) 来定制队列的行为，例如：

x-message-ttl: 消息的 TTL (Time-To-Live)，单位为毫秒。超过 TTL 的消息会被丢弃或进入死信队列 (如果配置了死信队列)。
x-expires: 队列的过期时间，单位为毫秒。超过过期时间且队列不再被使用时，队列会被自动删除。
x-dead-letter-exchange: 死信交换机 (Dead-Letter Exchange，DLX)。当队列中的消息被拒绝 (NACK)、过期或队列达到最大长度时，消息会被重新发布到死信交换机。
x-dead-letter-routing-key: 死信路由键 (Dead-Letter Routing Key，DLK)。消息被重新发布到死信交换机时使用的路由键。
x-max-length: 队列的最大长度，即队列中可以存储的最大消息数量。超过最大长度后，队列头部的消息会被丢弃或进入死信队列。
x-max-length-bytes: 队列的最大长度，以字节为单位。
x-overflow: 队列溢出行为，可选值包括 drop-head (丢弃队列头部消息)、 reject-publish (拒绝发布新消息)、 reject-publish-dlx (拒绝发布新消息并发送到死信队列)。
x-queue-mode: 队列模式，可选值包括 default (默认模式，消息存储在磁盘和内存中) 和 lazy (懒惰模式，消息尽可能存储在磁盘上，减少内存占用)。

代码示例 (Python - pika - 设置队列属性):


import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# 声明队列并设置属性
channel.queue_declare(queue='queue_with_args',
                      arguments={
                          'x-message-ttl': 60000,       # 消息 TTL 为 60 秒
                          'x-dead-letter-exchange': 'dlx_exchange', # 死信交换机
                          'x-max-length': 1000          # 最大长度为 1000 条消息
                      })
print(" [*] Queue with arguments declared.")
connection.close()

2.6.3 队列的工作流程与 Mermaid 图表

以下 Mermaid 图表展示了消息从生产者到消费者，经过交换机和队列的完整流程：

流程详解:

生产者 (Producer) 发布消息 (Publish Message): 生产者应用程序将消息发送到指定的交换机 (Exchange)。在发送消息时，生产者需要指定交换机名称、路由键 (Routing Key) 和消息内容。
交换机 (Exchange) 路由消息 (Route): 交换机接收到消息后，根据交换机类型 (例如 direct, topic, fanout, headers) 和绑定规则 (Bindings)，将消息路由到一个或多个队列 (Queue)。路由规则通常基于消息的路由键和绑定的路由键进行匹配。
队列 (Queue) 存储消息: 队列接收到被路由的消息后，将消息存储在队列中，等待消费者消费。
消费者 (Consumer) 消费消息 (Deliver Message): 消费者应用程序连接到 RabbitMQ 服务器，并订阅指定的队列。当队列中有新的消息时，RabbitMQ 会将消息推送给消费者。
消费者确认 (Acknowledgement): 消费者接收到消息并成功处理后，会向 RabbitMQ 服务器发送消息确认 (ACK)。RabbitMQ 收到 ACK 后，才会将消息从队列中移除。如果消费者未发送 ACK 或者处理消息失败，RabbitMQ 会根据配置重新投递消息。

2.6.4 队列的代码实践 (多语言示例)

除了 Python (pika) 示例，以下提供 Java (RabbitMQ Java Client) 和 JavaScript (amqplib) 的代码示例，展示如何声明队列、发送消息和消费消息。

1. Java (RabbitMQ Java Client)

声明队列 (Declare Queue):


import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class DeclareQueue {
    private static final String QUEUE_NAME = "java_queue";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 声明队列
            System.out.println(" [x] Queue '" + QUEUE_NAME + "' declared");
        }
    }
}

发送消息 (Send Message):


import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class SendMessage {
    private static final String QUEUE_NAME = "java_queue";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            String message = "Hello RabbitMQ from Java!";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8")); // 发送消息
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}

消费消息 (Receive Message):


import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
public class ReceiveMessage {
    private static final String QUEUE_NAME = "java_queue";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {}); // 消费消息
        while (true) {
            Thread.sleep(1000); // 保持连接
        }
    }
}

2. JavaScript (amqplib)

声明队列 (Declare Queue):


const amqp = require('amqplib');
async function declareQueue() {
  try {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();
    const queueName = 'js_queue';
    await channel.assertQueue(queueName, { durable: false }); // 声明队列
    console.log(` [x] Queue '${queueName}' declared`);
    await channel.close();
    await connection.close();
  } catch (error) {
    console.error("Error:", error);
  }
}
declareQueue();

发送消息 (Send Message):


const amqp = require('amqplib');
async function sendMessage() {
  try {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();
    const queueName = 'js_queue';
    await channel.assertQueue(queueName, { durable: false });
    const message = 'Hello RabbitMQ from JavaScript!';
    channel.sendToQueue(queueName, Buffer.from(message)); // 发送消息
    console.log(` [x] Sent '${message}'`);
    await channel.close();
    await connection.close();
  } catch (error) {
    console.error("Error:", error);
  }
}
sendMessage();

消费消息 (Receive Message):


const amqp = require('amqplib');
async function receiveMessage() {
  try {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();
    const queueName = 'js_queue';
    await channel.assertQueue(queueName, { durable: false });
    console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", queueName);
    channel.consume(queueName, msg => {
      if (msg !== null) {
        console.log(" [x] Received '%s'", msg.content.toString());
        channel.ack(msg); // 手动确认
      }
    }, { noAck: false }); // 关闭自动确认
  } catch (error) {
    console.error("Error:", error);
  }
}
receiveMessage();

2.6.5 队列的最佳实践

队列命名: 使用有意义且规范的队列名称，方便管理和维护。建议采用命名空间或前缀来区分不同应用或模块的队列。
队列持久性: 根据业务需求选择是否需要队列持久性。对于重要的消息，建议使用持久队列和持久消息。
消息确认机制: 对于需要保证消息可靠性的场景，务必使用手动消息确认机制，并妥善处理消息确认和重试逻辑。
队列长度监控: 监控队列的长度，避免队列积压导致系统性能下降或消息丢失。可以设置队列的最大长度限制和溢出策略。
死信队列: 合理配置死信队列，用于处理无法正常消费的消息，例如过期消息、被拒绝的消息等，方便后续分析和处理。
队列类型选择: 根据业务场景选择合适的队列类型 (经典队列或 Quorum 队列)。
prefetch_count: 合理设置 prefetch_count (QoS)，控制消费者每次预取的消息数量，平衡性能和公平性。

2.6.6 总结

队列是 RabbitMQ 中至关重要的核心概念，它提供了消息缓冲、解耦、异步处理、流量削峰等关键功能。深入理解队列的特性、工作流程和最佳实践，能够帮助我们更好地利用 RabbitMQ 构建可靠、高效的消息驱动系统。通过本文的详解和代码示例，相信您已经对 RabbitMQ 队列有了更全面的认识。在实际应用中，请根据具体的业务需求和场景，灵活运用队列的各种特性，并遵循最佳实践，构建健壮的消息队列系统。

作者与出处

原作者：灏天文库

来源：灏天文库

原文： https://aiknowledge.cn/collection/216-rabbitmq%E6%B6%88%E6%81%AF%E9%98%9F%E5%88%97%E5%AE%9E%E6%88%98%E6%95%99%E7%A8%8B

整理：灏天文库整理

由灏天文库平台收录，内容或由平台用户上传，仅供学习交流