Python操作RabbitMQ初体验（一）

由于想用Python实现一套分布式系统，来管理和监控CDN的内容与运行状态，误打误撞认识了RabbitMQ，推荐的人很多，如余锋《我为什么要选择RabbitMQ》等等。

在MQ这个词汇映入眼帘的时候，我花了好几个小时去阅读什么是MQ，即Message Queue(消息队列）。顾名思义，消息队列，装满消息的队列，队列，数据结构名词，具备先进先出特性的一种数据结构。消息队列无非是用来传递消息的而已，那么其意义何在，应用场景又在哪里，具备什么特性，及其独特优势是什么，为什么要用，这些都是浮现在我脑海中的一连串问题。

经过一大串搜索，终于肤浅的理解消息队列是应用对应用的信息传递的通信方式。比如需要分析多台服务器的日志，完全可以每台服务器都用一个进程往一个Mysql数据库的表里面写入数据，即所需要的信息，然后再写几个进程，读取表里数据，进行数据分析不是挺好，不过很快，这种设计的丑陋之处就显现出来了…… 你想要多个程序从一个队列当中取数据来处理？没问题，我们硬编码程序的个数好了……什么？还要能够允许程序动态地增加和减少的时候动态进行压力分配？这是很流行的《RabbitMQ+Python入门经典 兔子和兔子窝》里面的例子。想想也是，当我的CDN传输来一大堆数据的时候，数据的分发，处理，所有的一切都会是问题。但是其实还是没明白，Rabbit到底是怎么样实现这些东西的。

概念上说，RabbitMQ是AMPQ（高级消息协议队列）的标准实现，据说不熟悉AMQP，会看不懂RabbitMQ的文档。但是也只能建大的理解关键概念了。整个RabbitMQ的实现原理模型见下图，其实就是一个带路由任务分发队列的生产者与消费者模型。如图所示，即生产者生产出相应的信息，发送给路由器，路由器根据信息中的关键Key信息，将信息分发到不同的消息队列中，再由消费者去不同的消息队列中读取数据的过程。

Broker：简单来说就是消息队列服务器实体。
　　Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
　　Queue：消息队列载体，每个消息都会被投入到一个或多个队列。
　　Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
　　Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
　　vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
　　producer：消息生产者，就是投递消息的程序。
　　consumer：消息消费者，就是接受消息的程序。
　　channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

消息队列的使用过程大概如下：

（1）客户端连接到消息队列服务器，打开一个channel。
　　（2）客户端声明一个exchange，并设置相关属性。
　　（3）客户端声明一个queue，并设置相关属性。
　　（4）客户端使用routing key，在exchange和queue之间建立好绑定关系。
　　（5）客户端投递消息到exchange。

了解了RabbitMQ大概流程与优势之后，我开始在自己的Fedora上进行RabbitMQ的试用。

根据RabbitMQ官网介绍，安装需要从此处下载 

rpm --import http://www.rabbitmq.com/rabbitmq-signing-key-public.asc yum install rabbitmq-server-3.3.0-1.noarch.rpm

chkconfig rabbitmq-server on

/sbin/service rabbitmq-server stop/start

grep beam.smp /var/log/audit/audit.log | audit2allow -M mypolsemodule -i mypol.pp/sbin/service rabbitmq-server start

pip install pika

#!/usr/bin/env pythonimport pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(              
 'localhost'))channel = connection.channel()

channel.queue_declare(queue='sayhello')

channel.basic_publish(exchange='',                        
routing_key='hello',                                    
body='hello world')print "Sent ‘hello world'"

connection.close()

channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):    print " [x] Received %r" % (body,)

channel.basic_consume(callback,                      
queue='hello',                      
no_ack=True)print ' [*] Waiting for messages. To exit press CTRL+C'channel.start_consuming()