python - 如何在 RabbitMQ 中创建延迟队列？

Question

使用 Python、Pika 和 RabbitMQ 创建延迟(或 parking )队列的最简单方法是什么？我见过类似的questions ，但对于 Python 没有。

我发现在设计应用程序时这是一个有用的想法，因为它允许我们限制需要再次重新排队的消息。

总有可能您收到的消息超出您的处理能力，可能是 HTTP 服务器速度较慢，或者数据库压力过大。

我还发现，在对丢失消息零容忍的情况下出现问题时，它非常有用，而重新排队无法处理的消息可能会解决这个问题。它还可能导致消息反复排队的问题。可能导致性能问题，并记录垃圾邮件。

Answer 1

我发现这在开发我的应用程序时非常有用。因为它为您提供了一种替代方法，而不是简单地重新排队您的消息。这可以轻松降低代码的复杂性，并且是 RabbitMQ 中众多强大的隐藏功能之一。

步骤

首先我们需要设置两个基本 channel ，一个用于主队列，一个用于延迟队列。在最后的示例中，我包含了几个不需要的附加标志，但使代码更可靠；例如confirm delivery、delivery_mode和durable。您可以在 RabbitMQ manual 中找到更多信息。 .

设置完 channel 后，我们添加一个绑定(bind)到主 channel ，我们可以使用它来将消息从延迟 channel 发送到我们的主队列。

channel.queue_bind(exchange='amq.direct',
                   queue='hello')

接下来，我们需要配置延迟 channel ，以便在消息过期后将消息转发到主队列。

delay_channel.queue_declare(queue='hello_delay', durable=True,  arguments={
  'x-message-ttl' : 5000,
  'x-dead-letter-exchange' : 'amq.direct',
  'x-dead-letter-routing-key' : 'hello'
})

• x-message-ttl (消息 - 生存时间)

  这通常用于自动删除在特定持续时间后排队，但通过添加两个可选参数，我们可以更改此行为，而是让此参数确定消息将在延迟队列中停留多长时间(以毫秒为单位)。

• x-dead-letter-routing-key

  这个变量允许我们将消息转移到不同的队列一旦它们过期，而不是删除的默认行为它完全。

• x-dead-letter-exchange

  此变量确定使用哪个 Exchange 将消息从 hello_delay 传输到 hello 队列。

发布到延迟队列

当我们完成所有基本 Pika 参数的设置后，您只需使用基本发布将消息发送到延迟队列。

delay_channel.basic_publish(exchange='',
                      routing_key='hello_delay',
                      body="test",
                      properties=pika.BasicProperties(delivery_mode=2))

执行脚本后，您应该会看到在 RabbitMQ 管理模块中创建的以下队列。

示例。

import pika

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

# Create normal 'Hello World' type channel.
channel = connection.channel()
channel.confirm_delivery()
channel.queue_declare(queue='hello', durable=True)

# We need to bind this channel to an exchange, that will be used to transfer 
# messages from our delay queue.
channel.queue_bind(exchange='amq.direct',
                   queue='hello')

# Create our delay channel.
delay_channel = connection.channel()
delay_channel.confirm_delivery()

# This is where we declare the delay, and routing for our delay channel.
delay_channel.queue_declare(queue='hello_delay', durable=True,  arguments={
  'x-message-ttl' : 5000, # Delay until the message is transferred in milliseconds.
  'x-dead-letter-exchange' : 'amq.direct', # Exchange used to transfer the message from A to B.
  'x-dead-letter-routing-key' : 'hello' # Name of the queue we want the message transferred to.
})

delay_channel.basic_publish(exchange='',
                      routing_key='hello_delay',
                      body="test",
                      properties=pika.BasicProperties(delivery_mode=2))

print " [x] Sent"