连接到 OpenStack 中的 RabbitMQ 代理

Question

OpenStack使用RabbitMQ作为消息传递系统。为此，有几个交换机和队列。我发现名为“nova”的“topic”类型的交换用于消息传输。Exchange使用路由密钥将消息路由到队列(http://www.rabbitmq.com/tutorials/amqp-concepts.html). （有用图像位于http://www.rabbitmq.com/img/tutorials/intro/hello-world-example-routing.png-没有足够的声誉发布在这里）OpenStack中有几个队列，如compute、cert、network等。它们使用相同名称的路由密钥。因此，我使用这些路由密钥创建了几个新队列，以便将它们与处理消息的消费者绑定。例如，有一个名为“compute”的队列使用名为“compute”路由键。我创建了使用相同路由密钥的新队列“my_compute”。因为我认为它应该工作，我会收到消息。

我有一些代码连接到exchange，创建我的队列和消费者。

def connect(params):
connection = kombu.Connection(hostname=params['host'])
exchange = kombu.entity.Exchange(name=params['exchange_name'],
                                 type=params['exchange_type'],
                                 durable=params['exchange_durable'],
                                 auto_delete=params['exchange_auto_delete'],
                                 internal=params['exchange_internal'])
queue_list = []
for queue in params['queues_params']:
queue_list.append(kombu.messaging.Queue(name=queue['name'],
                                        exchange=exchange,
                                        routing_key=queue['routing_key'],
                                        channel=connection.channel(),
                                        durable=queue['durable'],
                                        auto_delete=queue['auto_delete']))
consumer = kombu.messaging.Consumer(channel=connection.channel(),
                                    queues=queue_list, 
                                    no_ack=True,
                                    callbacks=[self._process_message])
consumer.consume()
return connection

参数“params”是从json文件获得的映射：

{
"host"                 : "xxx",
"exchange_name"        : "nova",
"exchange_type"        : "topic",
"exchange_durable"     : false,
"exchange_auto_delete" : false,
"exchange_internal"    : false,
"queues_params"        : [
    {
        "name"        : "my_compute",
        "routing_key" : "compute",
        "durable"     : false,
        "auto_delete" : false,
        "arguments"   : [ ]
    },
    {
        "name"        : "my_network",
        "routing_key" : "network",
        "durable"     : false,
        "auto_delete" : false,
        "arguments"   : [ ]
    },
    .
    .
    .

起作用了。但我只收到网络队列的消息。我不知道还有没有其他消息，但看起来有。我说的对吗？还是哪里出了问题？是否有其他消息，我如何获得它们？

Answer 1

在queues_params中的routing_key中，您指定了“network”，这就是您只接收网络消息的原因。您可以使用通配符routing_key“#”，它捕获与相应主题交换相关联的所有消息。如果你愿意的话，你可以看看我的便条。

Answer 2

这段代码在这个开发周期中正在经历一些积极的变化，但目前我断言您正在深入研究。对于大多数nova组件，队列接口被抽象到这些组件使用的RPC公共库之下，并选择主题和队列。

特别是，当RPC代码想要将消息发送到特定的计算、网络或存储主机时，主题也可以是特定于主机的。您将在上面看到的唯一消息是通用广播消息，实际上通常是对有关浮动IP地址信息的请求

如果您想要一个捕获队列中所有消息的例子，那么您肯定应该看看Ceilometer，它就是用来做这件事的，并且利用了也嵌入到nova和相关组件中的通知系统。它不会给你提供与拦截和解释nova、network和cinder消息相同的东西——如果有用的话，这取决于你的总体目标。