如何为消费者订阅 N 个未知队列？

Question

我正在尝试了解如何做到这一点：

一些生产者创建了 N 个队列（比如说foo.1 foo.2 foo.3）。比我在兔子的另一部分有一个消费者需要从所有 N 个（在我的示例 3 中）队列中获取消息。我知道我可以这样做：

(method_frame, header_frame, body) = self.channel.basic_get(queue='foo.1', no_ack=False)
(method_frame, header_frame, body) = self.channel.basic_get(queue='foo.2', no_ack=False)
(method_frame, header_frame, body) = self.channel.basic_get(queue='foo.3', no_ack=False)

但是如果我的消费者不知道这些名字怎么办，我真正想做的是：

(method_frame, header_frame, body) = self.channel.basic_get(queue='foo.*', no_ack=False)

Answer 1

RabbitMQ 管理界面/api 将有权访问服务器上的所有队列。有一个易于使用的 Python 客户端 PyRabbit 可以让您使用 get_queues。从那里你可以做任何你需要的过滤。

Answer 2

这是我看到的： 您需要从提供的数量n 的队列中获取所有消息。根据我的个人经验，我会编写一个 for 循环并通过添加 "foo.%s" % (iteration) 创建一个字符串。

这是我的意思的一个例子：

for i in range(queues):
    str = 'foo.%s' % (i)
    (method_frame, header_frame, body) = self.channel.basic_get(queue=str, no_ack=False)

只要知道队列数，就可以用这个了。

Answer 3

如果您的消费者有办法识别队列，您应该能够通过搜索foo.__dict__ 找到所有队列。

您应该记住，如果您的任何队列设置在班级级别，那么它们将不会出现在 foo.__dict__ 中。在这种情况下，您将不得不编写一个遍历 foo 的 mro 的算法。

或者，如果您可以修改队列的创建，则可以通过使用某种管理器来跟踪它们。

class MyQueue(list):
    queues = {}  # Keeps track of all the queues out there

    @classmethod
    def add_to_producer(cls, obj, name, init_values):
        q = MyQueue(init_values)
        cls.queues[(obj, name)] = q
        setattr(obj, name, q)


class MyProducer(object):

    def __init__(self):
        # Initialize our producer with a couple of queues
        MyQueue.add_to_producer(self, 'a', [1,2])
        MyQueue.add_to_producer(self, 'b', [])


p1 = MyProducer()
p2 = MyProducer()
# Add another queue to p2
MyQueue.add_to_producer(p2, 'c', [4,5,6])

# Go through all of our created queues
for obj, attr_name in MyQueue.queues:
    if obj == p1:
        print 'p1', getattr(obj, attr_name)

    if obj == p2:
        print 'p2', getattr(obj, attr_name)

>>> p1 [1, 2]
>>> p1 []
>>> p2 [4, 5, 6]
>>> p2 [1, 2]
>>> p2 []