RabbitMQ：带有主题交换的持久消息

Question

我对 RabbitMQ 很陌生。

我已经建立了一个“主题”交换。消费者可以在发布者之后启动。我希望消费者能够接收在启动之前发送且尚未消耗的消息。

交换器使用以下参数进行设置：

exchange_type => 'topic'
durable => 1
auto_delete => 0
passive => 0

使用此参数发布消息：

delivery_mode => 2

消费者使用 get() 从交换器检索消息。

不幸的是，在任何客户端启动之前发布的任何消息都会丢失。我使用了不同的组合。

我想我的问题是交换不保存消息。也许我需要在发布者和消费者之间有一个队列。但这似乎不适用于通过密钥路由消息的“主题”交换。

我应该如何进行？我使用 Perl 绑定 Net::RabbitMQ （应该没关系）和 RabbitMQ 2.2.0。

Answer 1

如果在发布消息时没有连接的使用者可用于处理消息，则需要一个持久队列来存储消息。

交换器不存储消息，但队列可以。令人困惑的部分是，交换可以被标记为“持久”，但这真正意味着，如果您重新启动代理，交换本身仍将存在，但它确实不存在。 > 意味着发送到该交换器的任何消息都会自动保留。

鉴于此，这里有两个选项：

1. 在启动发布商自行创建队列之前，执行管理步骤。您可以使用 Web UI 或命令行工具来执行此操作。确保将其创建为持久队列，以便即使没有活动的使用者，它也能存储路由到它的任何消息。
2. 假设您的消费者被编码为始终在启动时声明（因此自动创建）他们的交换和队列（并且他们将它们声明为持久的），只需在启动任何发布者之前运行所有消费者至少一次 。这将确保正确创建所有队列。然后，您可以关闭消费者，直到真正需要它们为止，因为队列将持久存储路由到它们的任何未来消息。

我会选择#1。需要执行的步骤可能并不多，您始终可以编写所需步骤的脚本，以便可以重复这些步骤。另外，如果所有消费者都将从同一个队列中提取数据（而不是每个都有一个专用队列），那么这实际上是最小的管理开销。

队列是需要适当管理和控制的东西。否则，您最终可能会遇到流氓消费者声明持久队列，使用它们几分钟，但再也不会。不久之后，您将拥有一个永久增长的队列，并且没有任何东西可以减少其大小，以及即将到来的经纪人灾难。

Answer 2

正如 Brian 所提到的，交换器不存储消息，主要负责将消息路由到另一个交换器或队列。如果交换器未绑定到队列，则发送到该交换器的所有消息都将“丢失”。

您不需要在发布者脚本中声明固定的客户端队列，因为这可能不可扩展。队列可以由发布者动态创建，并使用交换器到交换器绑定在内部路由。

RabbitMQ 支持交换到交换的绑定，这将允许拓扑灵活性、解耦和其他好处。您可以在 RabbitMQ Exchange 到 Exchange 绑定处阅读更多信息[AMPQ]

RabbitMQ Exchange 到 Exchange 绑定

如果不存在使用队列的消费者，则用于创建具有持久性的交换到交换绑定的示例 Python 代码。

#!/usr/bin/env python
import pika
import sys
 
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
 
 
#Declares the entry exchange to be used by all producers to send messages. Could be external producers as well
channel.exchange_declare(exchange='data_gateway',
exchange_type='fanout',
durable=True,
auto_delete=False)
 
#Declares the processing exchange to be used.Routes messages to various queues. For internal use only
channel.exchange_declare(exchange='data_distributor',
exchange_type='topic',
durable=True,
auto_delete=False)
 
#Binds the external/producer facing exchange to the internal exchange
channel.exchange_bind(destination='data_distributor',source='data_gateway')
 
##Create Durable Queues binded to the data_distributor exchange
channel.queue_declare(queue='trade_db',durable=True)
channel.queue_declare(queue='trade_stream_service',durable=True)
channel.queue_declare(queue='ticker_db',durable=True)
channel.queue_declare(queue='ticker_stream_service',durable=True)
channel.queue_declare(queue='orderbook_db',durable=True)
channel.queue_declare(queue='orderbook_stream_service',durable=True)
 
#Bind queues to exchanges and correct routing key. Allows for messages to be saved when no consumer is present
channel.queue_bind(queue='orderbook_db',exchange='data_distributor',routing_key='*.*.orderbook')
channel.queue_bind(queue='orderbook_stream_service',exchange='data_distributor',routing_key='*.*.orderbook')
channel.queue_bind(queue='ticker_db',exchange='data_distributor',routing_key='*.*.ticker')
channel.queue_bind(queue='ticker_stream_service',exchange='data_distributor',routing_key='*.*.ticker')
channel.queue_bind(queue='trade_db',exchange='data_distributor',routing_key='*.*.trade')
channel.queue_bind(queue='trade_stream_service',exchange='data_distributor',routing_key='*.*.trade')

Answer 3

您的情况似乎是“消息持久性”。

从 RabbitMQ 教程文档 中，您需要标记两个 队列和消息作为持久的（下面的代码为C#版本。对于其他语言，您可以更喜欢此处）。

1. 首先，在Publisher中，您需要确保队列能够在RabbitMQ节点重新启动后继续存在。为此，我们需要将其声明为持久的：

channel.QueueDeclare(queue: "hello",
                     durable: true,
                     ....);

2. 其次，在 Consumer 中，您需要将您的消息标记为持久 - 通过设置 IBasicProperties.SetPercient为真。

var properties = channel.CreateBasicProperties();
properties.Persistent = true;