OpenStack中的消息队列2--以Quantum为例

wendu

基础

# A queue is not bound by any limits, it can store as many messages as you like ‒ it's essentially an infinite buffer.
# If we send a message to non-existing location, RabbitMQ will just trash(丢弃) the message.
# Creating a queue using queue_declare(python) is idempotent ‒ we can run the command as many times as we like, and only one will be created.
# By default, RabbitMQ will send each message to the next consumer, in sequence on one queue, called round-robin
# It's a common mistake to miss the basic_ack（消息确认）. It's an easy error, but the consequences are serious. Messages will be redelivered when your client quits (which may look like random redelivery), but RabbitMQ will eat more and more memory as it won't be able to release any unacked messages.
# RabbitMQ doesn't allow you to redefine an existing queue with different parameters
# we can set ‘durable’ to queue and set ‘delivery_mode’ to messages.
# on consumer side, channel.basic_qos(prefetch_count=1) don't dispatch a new message to a worker until it has processed and acknowledged the previous one(Pay attention: If all the workers are busy, your queue can fill up)
# exchange type: direct, topic, headers and fanout
a direct exchange is simple - a message goes to the queues whose binding key exactly matches the routing key of the message. Also, we can bind multiple queues with the same binding key, just like the fanout type exchange.
Topic exchange is powerful and can behave like other exchanges. When a queue is bound with "#" binding key - it will receive all the messages, regardless of the routing key - like in fanout exchange. When special characters "*" and "#" aren't used in bindings, the topic exchange will behave just like a direct one.
# 在client端, 一个无参的queueDeclare()创建一个非持久化的，自动删除的，具有唯一名字的临时队列
# once we disconnect the consumer, the queue which has an exclusive flag should be deleted.
# 只有持久化的queue才能绑定到持久化的exchange上
# The AMQP protocol predefines a set of 14 properties that go with a message. Most of the properties are rarely used, with the exception of the following:
delivery_mode: Marks a message as persistent (with a value of 2) or transient (any other value).
content_type: Used to describe the mime-type of the encoding. For example for the often used JSON encoding it is a good practice to set this property to: application/json.
reply_to: Commonly used to name a callback queue.
correlation_id: Useful to correlate RPC responses with requests

命令

rabbitmqctl list_queues
rabbitmqctl list_exchanges
rabbitmqctl list_bindings
rabbitmqctl list_queues name messages_ready messages_unacknowledged  列出队列中待处理和未确认的消息个数
rabbitmq-server –detached  启动RabbitMQ server
rabbitmqctl stop
rabbitmqctl status
rabbitmqctl cluster_status

RabbitMQ在OpenStack中的使用

之前曾今写过一篇Blog介绍这个话题，此次以G版中的quantum（OVS）为例再简单介绍一下。

quantum-agent

ovs-agent启动时初始化消息队列的代码在setup_rpc函数中：

由上面的代码中consumers变量定义了三个字符串：
q-agent-notifier-port-update
q-agent-notifier-network-delete
q-agent-notifier-tunnel-update
进而，由Kombu实现，创建了三个FanoutConsumer对象（这三个对象的生命周期随进程的结束而结束），分别监听三个绑定在不同fanout类型exchange上的queue。也就是说，系统中有至少有三个fanout类型的exchange，名称分别为：
q-agent-notifier-port-update_fanout
q-agent-notifier-network-delete_fanout
q-agent-notifier-tunnel-update_fanout
每个exchange上绑定有一个queue，名称分别为(其中的UUID是系统生成的唯一的标识号)：
q-agent-notifier-port-update_fanout_UUID
q-agent-notifier-network-delete_fanout_UUID
q-agent-notifier-tunnel-update_fanout_UUID
其实这里的名称只是作为一个标识，因为发往fanout类型的exchange的消息是不需要定义routing-key的。

上述的exchange和queue名称也可以在controller节点上通过使用rabbitmqctl list_bindings命令查看。

在ovs plugin中，以update_port为例：

进入port_update:

这里，plugin组装消息，创建一个临时的FanoutPublisher（发送完消息，该对象被关闭），将消息发送到一个exchange，exchange的名称为”<topic>_fanout”，也就是” q-agent-notifier-port-update_fanout”。也就是说，fanout_cast()函数中的topic参数其实是定义了fanout类型exchange的名称。于是上述的agent就能收到消息并处理了。整个结构如下图：

quantum-plugin

同样的，在plugin启动时，会创建TopicConsumer，监听名称为q-plugin的队列，该队列以binding-key=’q-plugin’绑定在名称为’openstack’、类型为topic的exchange上。
self.conn.create_consumer(self.topic, self.dispatcher, fanout=False)
上述代码中的topic其实是定义了队列的名称和binding-key。这一点与agent不同。

dhcp-agent/l3-agent

dhcp-agent和l3-agent需要与quantum-plugin交互以获取网络的信息，并且需要向plugin定时上报状态。以report_state为例：

这里的topic是’q-plugin’。可以看到report_state方法是同步调用。同步调用相对来说比较复杂：
1. 在G版，所有的同步调用都使用同一个direct类型的exchange，名称为’reply_UUID’，上面绑定有一个queue，名称也为’reply_UUID’，当系统进行第一次call调用时，创建exchange和queue，创建动作在ReplyProxy对象中。
2. 消息中加入’ _reply_q’字段（= reply_UUID），这样TopicConsumer就可以根据消息的属性，知道该把消息返回到那个exchange。
3. 为此次同步调用创建一个MulticallProxyWaiter对象，注册到ReplyProxy对象中，该对象有一个消息队列，并在该队列上阻塞，直到该队列中收到消息。
4. ReplyProxy对象维护一个字典，key为msg_id，value为MulticallProxyWaiter对象，当queue中收到消息，根据消息的’ _msg_id’字段找到注册到字典的MulticallProxyWaiter对象，将消息放入该对象的消息队列中。这样MulticallProxyWaiter对象就可以拿到消息同步返回了。

当plugin收到消息，调用响应的方法处理后，看到消息中有’ _reply_q’字段，于是就会创建一个临时的DirectPublisher向’ _reply_q’表示的exchange发送响应。

当然，dhcp-agent与l3-agent一样，都需要接受调度，所以在进程启动时也会创建一个FanoutConsumer(监听名为’ dhcp_agent_fanout_UUID’，绑定在名为’ dhcp_agent_fanout’、类型为fanout的exchange上的queue)和两个TopicConsumer(一个监听名为’ dhcp_agent’，以binding-key=’dhcp_agent’绑定在名为’openstack’、类型为topic的exchange上的queue，另一个监听名为’dhcp_agent_HOSTMANE’，以binding-key=’dhcp_agent_HOSTMANE’绑定在同一个exchange上的queue)

njsuntop

我抢、我抢、我抢沙发~

123sw

床上运动也可以减肥的，你们都不知道吗？

zhendeaini123

美女未抱身先走,常使色狼泪满襟。。。。。。

posir

所有刻骨铭心的爱都灵魂游离于床上的瞬间！

忧郁者

啥时硬件也可以COPY就好了!

q666123

帮帮顶顶！！