标签：
本文是 OpenStack 中的 RabbitMQ 使用研究 两部分中的第一部分，将介绍 RabbitMQ 的基本概念，即 RabbitMQ 是什么。第二部分将介绍其在 OpenStack 中的使用。
RabbitMQ 是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。
AMQP 是一个定义了在应用或者组织之间传送消息的协议的开放标准 （an open standard for passing business messages between applications or organizations），它最新的版本是 1.0。AMQP 目标在于解决在两个应用之间传送消息存在的下列问题：
AMQP 使用异步的、应用对应用的、二进制数据通信来解决这些问题。
RabbitMQ 是 AMQP 的一种实现，它包括Server （服务器端）、Client （客户端） 和 Plugins （插件）。RabbitMQ 服务器是用 Erlang 语言编写的，其最新版本是刚刚（2015/02/11）发布的 3.4.4，而 OpenStack Juno 中使用的 Server 是 2014年3月发布的 3.2.4 版本。现在 RabbitMQ 支持的 AMQP 版本依然是0.9.1。
其基本概念参见下图：
RabbitMQ 官网 和其它网站上有很多文章来描述其基本概念。简单说明如下：
Queue （队列）：一个存储Exchange 发来的消息的缓冲，并将消息主动发送给Consumer，或者 Consumer 主动来获取消息。见 1.4 部分的描述。
消息结构：
消息的几个重要属性：
delivery_mode: 将其值设置为 2 将用于消息的持久化，持久化的消息会被保存到磁盘上来防止其丢失。下面章节 3 有描述。
消息的确认/删除机制：
Consumer 处理消息可能会失败，那么 RabbitMQ 怎么知道什么时候来删除 queue 中的消息呢？它使用两种机制：
第二种情况下，如果 RabbitMQ 没收到确认，它会把消息重新放进队列（re-queued）并添加标识 ‘redelivered‘ 表明该消息之前已经发送过 ，如果没有Consumer的话，消息将保持到有下一个 Consumer 为止。
Consumer 可以主动告诉 RabbitMQ 消息处理失败了（拒绝消息），并告知RabbitMQ 是删除消息还是重新放进队列。
exchange 有几个重要的属性：
RabbitMQ 默认会为每一种类型生成一个或者两个的默认的 exchange：
队列同样有类似于 exchange 的 name、durable、auto_delete 和 exclusive 等属性，并且含义相同。
Exchange 会将消息分发（copy）到符合要求的所有队列中。
Consumer 可以主动获取或者被动接受Queue里面的消息：
一个 Queue 允许有多个 Consumer，比如利用 RabbitMQ 来实现一个简单的 load balancer。这时候，消息会在这些 Consumer 之间根据 channel 的 prefetch level 做分发（请参见AQMP： QoS or message prefetching），如果该值一样的话，消息会被平均分发给这些Consumer。
RabbitMQ 提供Cli rabbitmqctl [-n <node>] [-q] <command> [<command options>] 来进行管理和配置。常用到的命令有：
Exchange 根据它自身的类型 type、消息的属性 routing_key 或者 headers，以及 Binding 的属性 binding_key 来转发消息。
| Exchange 的类型 Type | 使用的消息属性 | 使用的Binding 属性 | 转发模式 |
| Fanout | - (忽略消息的转发属性） | - （忽略binding的转发属性） |
Exchange 将消息转发到所有与它有 binding 关系的队列中。 这种方法转发效率较高。OpenStack 大量使用这种类型的 exchange。 |
| Direct | routing_key （任意的字符串，比如 "abc"） | binding_key （任意的字符串，比如 "abc"） | Exchange 只将消息转到 binding 的 binding_key 等于消息的 routing_key 的队列中。 |
| Topic | routing_key （以 "." 分割的多单词字符串，比如 abc.efg.hij） | binding_key （包含 "#" 和 "*" 的以 “.” 分割的多单词字符串，比如 *.efg.*） |
Exchange 只将消息转到消息的 routing_key 和 binding 的 binding_key 匹配的队列中。匹配规则如下： （1）两者以"."分割的单词数目相同 （2）"*"可代表一个单词 （3）"#“可代表零个或多个单词 |
| Headers | headers （消息头） | binding_key |
Exchange 只将消息转到消息的 headers 和 binding 的 binding_key 匹配的队列中。匹配规则待研究。 OpenStack不使用该类型的exchange。 |
参考文档：
https://www.rabbitmq.com/getstarted.html 这里有详细的阐述和示例源代码。
http://www.cnblogs.com/starof/p/4173413.html 这里有官网文档的中文版。
消息的持久化意味着在 RabbitMQ 被重启后，消息依然还在。要实现持久化，得实现几个相关组件的持久化：
（1）. 交换机的持久化，需要将其 durable 属性设为 true。chan.exchange_declare(exchange="sorting_room", type="direct", durable=True, auto_delete=False,)
（2）. 队列的持久化，需要将其 durable 属性设置为 true。chan.queue_declare(queue="po_box", durable=True, exclusive=False, auto_delete=False)
可以使用 RabbitMQ 来实现 RPC 机制，这里说说其实现原理：
过程：
（1）. 客户端 Client 设置消息的 routing key 为 Service 的队列 op_q；设置消息的 reply-to 属性为返回的 response 的目标队列 reponse_q，设置其 correlation_id 为以随机UUID，然后将消息发到 exchange。比如 channel.basic_publish(exchange=‘‘, routing_key=‘op_q‘, properties=pika.BasicProperties(reply_to = reponse_q, correlation_id = self.corr_id),body=request)
（2）. Exchange 将消息转发到 Service 的 op_q
（3）. Service 收到该消息后进行处理，然后将response 发到 exchange，并设置消息的 routing_key 为原消息的 reply_to 属性，以及设置其 correlation_id 为原消息的 correlation_id 。
ch.basic_publish(exchange=‘‘, routing_key=props.reply_to, properties=pika.BasicProperties(correlation_id = props.correlation_id), body=str(response))
（4）. Exchange 将消息转发到 reponse_q
（5）. Client 逐一接受 response_q 中的消息，检查消息的 correlation_id 是否为等于它发出的消息的correlation_id，是的话表明该消息为它需要的response。
这里有详细的阐述。
常用的Python AMQP SDK包括：
#创建Connection和Channel连接到 RabbitMQ 服务器 conn = amqp.Connection(host="localhost:5672", userid="guest", password="1111", virtual_host="/", insist=False) chan = conn.channel() #创建 queue result = chan.queue_declare(queue="debug", durable=True, exclusive=False, auto_delete=False) #创建 exchange result = chan.exchange_declare(exchange="sorting_room2", type="topic", durable=True, auto_delete=False,) #创建 binding result = chan.queue_bind(queue="debug", exchange="sorting_room2", routing_key="*.debug") #回调函数，当有 message 到达 queue 后，该函数会被调用 def recv_callback(msg): print ‘Received: ‘ + msg.body + ‘ from channel #‘ + str(msg.channel.channel_id)
# lChannel.basic_ack(msg.delivery_tag) #如果no_ack=False的话，可以需要发回一个确认
#启动一个 consumer，consumer_tag 是该 consumer 的一个唯一标识符
#no_ack = True 表示该 consumer 不会发回确认
chan.basic_consume(queue=‘debug‘, no_ack=True, callback=recv_callback, consumer_tag="debugtag")
#等待有消息发到 queue while True: chan.wait()
#终止该 consumer chan.basic_cancel("testtag") #关闭 connection 和 channel chan.close() conn.close()
from amqplib import client_0_8 as amqp import sys
#创建 connection 和 channel conn = amqp.Connection(host="localhost:5672", userid="guest", password="1111", virtual_host="/", insist=False) chan = conn.channel()
#创建 message msg = amqp.Message(sys.argv[1]) msg.properties["delivery_mode"] = 2
#发送 message chan.basic_publish(msg,exchange="sorting_room2",routing_key=(sys.argv[2]))
#关闭 connection 和 channel chan.close() conn.close()
RabbitMQ 支持使用插件来支持 Management， Federation， Shovel 和 STOMP。所有的插件都在这里。
它提供 HTTP-based API 和 browser-based UI 以及 CLI 来管理 RabbitMQ。它的GUI的访问地址是 http://<rabbitmq server ip>:15672/#/traces。它的GUI上，提供了一个 overview，还可以通过它来管理connection、channel、exchange 和 queue，以及 virtual host，tracing 和 policy等。
该机制提供了一个查看被转发的消息的途径。当打开 firehose 的时候，RabbitMQ 会自动建立 amq.rabbitmq.trace 和 amq.rabbitmq.log 两个exchange。你可以编程创建queue 从这两个 exchange 里面获取 trace 和 log，从而观察每一个被处理的消息。这里有一个开源代码实现。
rabbitmq_tracing 插件在 management 插件增加了消息追踪的方法，它是从 firehose 中获取数据。在激活了 rabbitmq-management，firehose 和 rabbitmq_tracing，你可以在 management GUI 中追踪消息：
自此，RabbitMQ 基本上算熟悉了，接下来可以开始分析 OpenStack 中是如何使用 RabbitMQ 了。
探索 OpenStack 之（14）：OpenStack 中 RabbitMQ 使用研究 （上半部分）
标签：
原文地址：http://www.cnblogs.com/sammyliu/p/4293011.html
