使用 RabbitMQ 有没有办法在没有出队操作的情况下查看队列内容？

Posted 2023-03-10

【中文标题】使用 RabbitMQ 有没有办法在没有出队操作的情况下查看队列内容？

【英文标题】：Using RabbitMQ is there a way to look at the queue contents without a dequeue operation?

【发布时间】：2011-06-09 16:11:37

【问题描述】：

作为学习 RabbitMQ 和 python 的一种方式，我正在开展一个项目，该项目允许我在多台计算机之间分发 h264 编码。基础知识已经完成，我有一个在 Linux 或 Mac 上运行的守护进程，它附加到队列、接受作业并使用 HandBrakeCLI 对其进行编码，并在编码完成后确认消息。我还构建了一个简单的工具来将项目推送到队列中。

现在我想扩展将项目推送到队列中的工具的功能，以便我可以查看队列中的内容。我知道能够查看队列中有多少项目，但我希望能够获取实际消息，以便我可以显示正在等待编码的电影或电视节目。这个想法是队列管理器将在作业完成时从编码器客户端接收消息，然后刷新队列列表。

我知道有一种复杂的方法可以使队列管理器的列表与实际工作队列保持同步，但我希望这是“持久的”，因为我应该能够关闭队列管理器并稍后重新打开它以查看队列。

【问题讨论】：

我通过 twitter 收到了这个 - “不 - RabbitMQ 的队列是纯 FIFO 结构，没有窥视。但是，看看 basic.consume/get with acks”

【参考方案1】：

您想要做的是浏览队列，虽然我从this 得知 RabbitMQ 还不支持。

【讨论】：

不直接支持，但是如果您声明一个没有自动确认的队列并且不确认您收到的消息，那么您可以看到其中的所有内容。查看完之后，在通道上发送 CANCEL，或者断开连接并重新连接以使所有消息重新排队。这确实会在消息头中增加一个数字，但不会影响消息。

@MichaelDillon - 该评论应该是一个答案。

@cdeszaq：是的，你是对的。我把它写成答案，提供更多信息。

【参考方案2】：

不直接支持队列浏览，但是如果您声明一个没有自动确认的队列并且不确认您收到的消息，那么您可以看到其中的所有内容。查看完之后，在通道上发送 CANCEL，或者断开连接并重新连接以使所有消息重新排队。这确实会在消息头中增加一个数字，但不会影响消息。

我构建了一个应用程序，其中消息排序不是非常重要，并且我经常以这种方式扫描队列。如果发现问题，我会将消息转储到文件中，修复它们并重新提交。

如果您只需要偶尔查看一两条消息，您可以使用 RabbitMQ 管理插件来做到这一点。

此外，如果您只需要消息计数，则可以在每次声明队列时或在 basic.get 命令上获取。

【讨论】：

只是为了扩展消息计数，您可以声明一个带有passive=True的队列，如果它不存在则不会创建队列。如果是这样，您将获得消息和消费者的数量。

RabbitMQ 有一个名为 Firehose Tracer 的专有扩展，它可以访问所有发布的消息。详情在这里：rabbitmq.com/firehose.html

【参考方案3】：

@MichaelDillon 根据你的回答让别人的生活更轻松，我在这里举一个 no_ack 的例子：

#!/usr/bin/env python
import pika

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

channel.queue_declare(queue='Q.hello')


def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    # ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(callback, queue='Q.hello')

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

【讨论】：

ch.basic_ack(delivery_tag=method.delivery_tag) 在这种情况下有效地用作“从队列中删除”命令。 +1

这非常有用，但我无法找到一种不会无限期阻塞的方法。基本上我只想在队列中查看固定数量的消息，而不需要坐等更多消息进来。有什么想法吗？