配置持久化队列和交换机以实现可靠消息传递

当您的应用程序使用 RabbitMQ 处理作业、订单或通知时，代理重启不应清除队列中等待处理的工作。持久化队列、持久化交换机、持久化绑定和持久化消息是使 RabbitMQ 在重启后保持可靠性的关键组件。

本指南将介绍您需要的设置、人们通常出错的地方，以及如何在信任生产环境之前验证行为。

理解持久性与持久化

在配置之前，区分与消息存活相关的两个主要概念至关重要：

队列持久性： 指队列定义本身。持久化队列定义在代理重启后仍然存在。如果队列声明为非持久化，则在代理停止时会被删除。
交换机持久性： 指交换机定义。持久化交换机在重启后仍然存在；非持久化交换机在代理停止时会被移除。
绑定持久性： 持久化交换机和持久化队列之间的绑定会随持久化拓扑一起恢复。涉及临时实体的绑定会随这些实体一起消失。
消息持久化： 指单个消息的处理方式。持久化消息由代理写入磁盘，确保即使队列本身是持久化的，消息也能在代理重启后存活。标记为瞬态（非持久化）的消息仅保存在内存中，在重启期间可能会丢失。

要使消息在代理重启后存活，队列必须是持久化的，并且消息必须以持久化方式发布。在正常的路由发布中，交换机和绑定也需要恢复，以便生产者和消费者可以继续使用相同的拓扑。

步骤 1：声明持久化队列

队列在创建时必须明确声明为持久化。这告诉 RabbitMQ 将队列元数据保存到磁盘，以便在代理重新上线时自动重新创建。

此配置通常通过连接到代理的客户端库（AMQP 客户端）完成。以下是常见工具中声明队列的示例。

使用 `rabbitmqadmin` CLI（或类似工具）的示例

使用命令行工具声明队列时，需要将 durable 参数设置为 true。

# 声明名为 'high_priority_tasks' 的持久化队列的命令
rabbitmqadmin declare queue name=high_priority_tasks durable=true

使用 Python（pika 库）的示例

在编程上下文中，channel.queue_declare() 方法中的 durable 参数必须设置为 True。

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

queue_name = 'order_processing_queue'

channel.queue_declare(
    queue=queue_name,
    durable=True  # <-- 在此处设置持久性
)

print(f"队列 '{queue_name}' 已声明为持久化。")
# connection.close() # （在其他操作后关闭连接）

队列声明警告： 如果队列已存在，并且您尝试使用不同的属性重新声明（例如，从非持久化更改为持久化），RabbitMQ 将引发错误（Precondition Failed 或类似错误），因为现有队列无法更改其持久性状态。

步骤 2：声明持久化交换机

将消息路由到队列的交换机也应声明为持久化，如果您依赖它们在代理重启后存活。如果交换机是非持久化的，它将在重启后被删除，并且与其关联的任何绑定也会丢失。

使用 Python（pika 库用于交换机声明）的示例

与队列类似，交换机在声明时需要将 durable 参数设置为 True。

import pika

# 假设连接和通道已建立

exchange_name = 'critical_events_exchange'

channel.exchange_declare(
    exchange=exchange_name,
    exchange_type='direct',
    durable=True
)

print(f"交换机 '{exchange_name}' 已声明为持久化。")

步骤 3：发布持久化消息

仅声明持久化队列和交换机只能确保拓扑存活。要确保消息本身存活，发布者必须将消息属性标记为持久化。

发布时，将 delivery_mode 属性设置为 2（表示持久化）。

示例：发布持久化消息（Pika）

在 channel.basic_publish 调用中，使用 properties 参数设置消息持久化。

import pika
from pika import BasicProperties

# ... 通道设置 ...

message_body = "此订单绝不能丢失！"
exchange = 'critical_events_exchange'
routing_key = 'urgent'

channel.basic_publish(
    exchange=exchange,
    routing_key=routing_key,
    body=message_body,
    properties=BasicProperties(
        delivery_mode=2  # <-- 传递模式 2 = 持久化
    )
)

print("消息已持久化发布。")

最佳实践：发布者确认： 虽然持久化在代理重启期间保存数据，但它不能保证代理在发布者应用程序崩溃之前收到消息。为了最大可靠性，始终将持久化/持久化配置与发布者确认配对使用，以从代理处接收消息已安全写入磁盘的确认。

步骤 4：绑定持久化组件

一旦创建了持久化队列和持久化交换机，必须将它们绑定在一起。绑定定义了路由逻辑。如果交换机是持久化的，与其关联的绑定通常也应该是持久化的，以确保路由结构在代理重启后立即生效。

# ... 通道设置 ...
exchange_name = 'critical_events_exchange'
queue_name = 'order_processing_queue'
routing_key = 'urgent'

channel.queue_bind(
    exchange=exchange_name,
    queue=queue_name,
    routing_key=routing_key
)

print(f"已在 {exchange_name} 和 {queue_name} 之间建立绑定。")

在 RabbitMQ 中，持久化队列和持久化交换机之间的绑定是持久化的。如果任一方是瞬态的，绑定无法比该实体存活更久。

可靠性检查清单总结

要实现针对代理故障的端到端消息可靠性，请确保所有三个组件都正确配置：

组件	所需配置	目的
队列	`durable=True`	在代理重启后存活（元数据已保存）。
交换机	`durable=True`	在代理重启后存活（拓扑已保存）。
绑定	持久化队列绑定到持久化交换机	重启后恢复路由关系。
消息	`delivery_mode=2`（持久化）	在代理重启后存活（数据写入磁盘）。

要点

RabbitMQ 中的持久化并非单一开关。声明持久化队列和交换机，绑定持久化实体，以 delivery_mode=2 发布消息，并启用发布者确认，以便您的发布者知道 RabbitMQ 已接受消息。然后重启一个非生产代理，验证队列、绑定和未消费的持久化消息是否仍然存在。