使用 Python、Pika 和 RabbitMQ 创建延迟（或停车）队列的最简单方法是什么？我见过类似的问题，但没有针对 Python。
在设计应用程序时，我发现这是一个有用的想法，因为它允许我们限制需要再次重新排队的消息。
总有可能您收到的消息超出您的处理能力，可能是 HTTP 服务器很慢，或者数据库压力过大。
我还发现，在对丢失消息零容忍的情况下出现问题时，它非常有用，而重新排队无法处理的消息可能会解决这个问题。它还可能导致消息反复排队的问题。可能导致性能问题，并记录垃圾邮件。
问问题
34303 次
6 回答
99
在开发我的应用程序时，我发现这非常有用。因为它为您提供了一种替代方法，而不是简单地重新排队您的消息。这可以轻松降低代码的复杂性，并且是 RabbitMQ 中许多强大的隐藏功能之一。
脚步
首先我们需要设置两个基本通道，一个用于主队列，一个用于延迟队列。在最后的示例中，我包含了几个不需要的附加标志，但使代码更可靠；例如和。confirm delivery_ 您可以在 RabbitMQ手册中找到有关这些的更多信息。delivery_modedurable
在我们设置了通道之后，我们向主通道添加了一个绑定，我们可以使用它来将消息从延迟通道发送到我们的主队列。
channel.queue_bind(exchange='amq.direct',
queue='hello')
接下来，我们需要配置我们的延迟通道，以便在消息过期后将消息转发到主队列。
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 （消息 - 生存时间）
这通常用于在特定持续时间后自动删除队列中的旧消息，但通过添加两个可选参数，我们可以更改此行为，而是让此参数以毫秒为单位确定消息将在延迟队列中保留多长时间。
-
这个变量允许我们在消息过期后将消息转移到不同的队列，而不是完全删除它的默认行为。
-
此变量确定使用哪个 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"
于 2013-06-09T21:58:47.780 回答
17
您可以使用 RabbitMQ 官方插件：x-delayed-message。
首先，下载ez 文件并将其复制到Your_rabbitmq_root_path/plugins
其次，启用插件（不需要重启服务器）：
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
最后，使用“x-delay”标头发布您的消息，例如：
headers.put("x-delay", 5000);
注意：
它不能确保您的消息的安全性，因为如果您的消息在您的 rabbitmq-server 停机期间过期，不幸的是消息丢失了。所以当你使用这个方案时要小心。
在rabbitmq-delayed-message-exchange中享受它和更多信息
于 2015-11-03T05:24:24.207 回答
8
仅供参考，如何在 Spring 3.2.x 中执行此操作。
<rabbit:queue name="delayQueue" durable="true" queue-arguments="delayQueueArguments"/>
<rabbit:queue-arguments id="delayQueueArguments">
<entry key="x-message-ttl">
<value type="java.lang.Long">10000</value>
</entry>
<entry key="x-dead-letter-exchange" value="finalDestinationTopic"/>
<entry key="x-dead-letter-routing-key" value="finalDestinationQueue"/>
</rabbit:queue-arguments>
<rabbit:fanout-exchange name="finalDestinationTopic">
<rabbit:bindings>
<rabbit:binding queue="finalDestinationQueue"/>
</rabbit:bindings>
</rabbit:fanout-exchange>
于 2014-06-09T21:22:28.097 回答
4
NodeJS 实现。
从代码中一切都非常清楚。希望它能节省一些人的时间。
var ch = channel;
ch.assertExchange("my_intermediate_exchange", 'fanout', {durable: false});
ch.assertExchange("my_final_delayed_exchange", 'fanout', {durable: false});
// setup intermediate queue which will never be listened.
// all messages are TTLed so when they are "dead", they come to another exchange
ch.assertQueue("my_intermediate_queue", {
deadLetterExchange: "my_final_delayed_exchange",
messageTtl: 5000, // 5sec
}, function (err, q) {
ch.bindQueue(q.queue, "my_intermediate_exchange", '');
});
ch.assertQueue("my_final_delayed_queue", {}, function (err, q) {
ch.bindQueue(q.queue, "my_final_delayed_exchange", '');
ch.consume(q.queue, function (msg) {
console.log("delayed - [x] %s", msg.content.toString());
}, {noAck: true});
});
于 2016-04-13T20:33:55.533 回答
1
Rabbit 队列中的消息可以通过 2 种方式延迟 - 使用 QUEUE TTL - 使用消息 TTL 如果队列中的所有消息要延迟固定时间，请使用队列 TTL。如果每条消息必须延迟不同的时间，请使用消息 TTL。我已经使用 python3 和 pika 模块对其进行了解释。pika BasicProperties 参数“过期”（以毫秒为单位）必须设置为延迟队列中的延迟消息。设置过期时间后，将消息发布到delayed_queue（“不是消费者等待消费的实际队列”），一旦delayed_queue中的消息过期，消息将使用exchange'amq.direct'路由到实际队列
def delay_publish(self, messages, queue, headers=None, expiration=0):
"""
Connect to RabbitMQ and publish messages to the queue
Args:
queue (string): queue name
messages (list or single item): messages to publish to rabbit queue
expiration(int): TTL in milliseconds for message
"""
delay_queue = "".join([queue, "_delay"])
logging.info('Publishing To Queue: {queue}'.format(queue=delay_queue))
logging.info('Connecting to RabbitMQ: {host}'.format(
host=self.rabbit_host))
credentials = pika.PlainCredentials(
RABBIT_MQ_USER, RABBIT_MQ_PASS)
parameters = pika.ConnectionParameters(
rabbit_host, RABBIT_MQ_PORT,
RABBIT_MQ_VHOST, credentials, heartbeat_interval=0)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
channel.queue_declare(queue=queue, durable=True)
channel.queue_bind(exchange='amq.direct',
queue=queue)
delay_channel = connection.channel()
delay_channel.queue_declare(queue=delay_queue, durable=True,
arguments={
'x-dead-letter-exchange': 'amq.direct',
'x-dead-letter-routing-key': queue
})
properties = pika.BasicProperties(
delivery_mode=2, headers=headers, expiration=str(expiration))
if type(messages) not in (list, tuple):
messages = [messages]
try:
for message in messages:
try:
json_data = json.dumps(message)
except Exception as err:
logging.error(
'Error Jsonify Payload: {err}, {payload}'.format(
err=err, payload=repr(message)), exc_info=True
)
if (type(message) is dict) and ('data' in message):
message['data'] = {}
message['error'] = 'Payload Invalid For JSON'
json_data = json.dumps(message)
else:
raise
try:
delay_channel.basic_publish(
exchange='', routing_key=delay_queue,
body=json_data, properties=properties)
except Exception as err:
logging.error(
'Error Publishing Data: {err}, {payload}'.format(
err=err, payload=json_data), exc_info=True
)
raise
except Exception:
raise
finally:
logging.info(
'Done Publishing. Closing Connection to {queue}'.format(
queue=delay_queue
)
)
connection.close()
于 2017-12-15T14:05:46.427 回答
1
取决于您的场景和需求，我会推荐以下方法，
- 使用官方插件，https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/，但是如果延迟消息的总数超过一定数量（https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/issues/72），它将没有高可用性选项，并且在 MQ 重启期间延迟时间用完时会丢失数据。
- 像 NServiceBus 一样实现一组级联延迟队列（https://docs.particular.net/transports/rabbitmq/delayed-delivery）。
于 2019-08-14T13:38:13.903 回答