python队列 rabbitmq介绍使用

孤独海岸线

　　解释RabbitMQ，就不得不提到AMQP（Advanced Message Queuing Protocol）协议。 AMQP协议是一种基于网络的消息传输协议，它能够在应用或组织之间提供可靠的消息传输。RabbitMQ是该AMQP协议的一种实现，利用它，可以将消息安全可靠的从发 送方传输到接收方。简单的说，就是消息发送方利用RabbitMQ将信息安全的传递给接收方。
　　RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。
　　MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消 息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

1 RabbitMQ安装
　　for Linux：



安装配置epel源
$ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm
安装erlang
$ yum -y install erlang
安装RabbitMQ
$ yum -y install rabbitmq-server

　　注意：service rabbitmq-server start/stop

for mac：
　　http://my.oschina.net/u/998693/blog/547873

for windows
　　http://www.cnblogs.com/liujianzuo888/articles/5757094.html
　　安装Rabbit MQ
　　Rabbit MQ 是建立在强大的Erlang OTP平台上，因此安装Rabbit MQ的前提是安装Erlang。通过下面两个连接下载安装3.2.3 版本：

• 下载并安装 Eralng OTP For Windows (vR16B03)
  
• 运行安装 Rabbit MQ Server Windows Installer (v3.2.3)
  

　　默认安装的Rabbit MQ 监听端口是5672
　　如何激活Rabbit MQ's Management Plugin



rabbitmq-server/sbin/
rabbitmqctl.bat add_user liujianzuo 123
rabbitmqctl.bat set_permissions -p /  liujianzuo ".*" ".*" ".*"
rabbitmqctl.bat set_user_tags liujianzuo administrator
rabbitmq-plugins.bat enable rabbitmq_management
net stop RabbitMQ && net start RabbitMQ




使用Rabbit MQ 管理插件，可以更好的可视化方式查看Rabbit MQ 服务器实例的状态，你可以在命令行中使用下面的命令激活：
"C:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.2.3\sbin\rabbitmq-plugins.bat" enable rabbitmq_management
要重启服务才能生效，可以执行
net stop RabbitMQ && net start RabbitMQ
下面我们使用rabbitmqctl控制台命令（位于C:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.2.3\sbin>）来创建用户，密码，绑定权限等。
Microsoft Windows [版本 6.3.9600]
(c) 2013 Microsoft Corporation。保留所有权利。
c:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.2.3\sbin 的目录
2014/11/01  15:04    <DIR>          .
2014/11/01  15:04    <DIR>          ..
2014/01/23  22:57               817 rabbitmq-echopid.bat
2014/01/23  22:57             1,900 rabbitmq-plugins.bat
2014/01/23  22:57             4,356 rabbitmq-server.bat
2014/01/23  22:57             7,123 rabbitmq-service.bat
2014/01/23  22:57             1,621 rabbitmqctl.bat
5 个文件         15,817 字节
2 个目录 96,078,618,624 可用字节
c:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.2.3\sbin>rabbitmqctl.ba
t list_users
Listing users ...
guest   [administrator]
...done.
c:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.2.3\sbin>rabbitmqctl.ba
t list_vhosts
Listing vhosts ...
/
...done.
c:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.2.3\sbin>rabbitmqctl.ba
t add_user geffzhang zsy@2014
Creating user "geffzhang" ...
...done.
c:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.2.3\sbin>rabbitmqctl.ba
t list_users
Listing users ...
geffzhang       []
guest   [administrator]
...done.
c:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.2.3\sbin>rabbitmqctl.ba
t set_user_tags geffzhang administrator
Setting tags for user "geffzhang" to [administrator] ...
...done.
c:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.2.3\sbin>rabbitmqctl.ba
t set_permissions -p /  geffzhang ".*" ".*" ".*"
Setting permissions for user "geffzhang" in vhost "/" ...
...done.
c:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.2.3\sbin>rabbitmqctl.ba
t list_users
Listing users ...
geffzhang       [administrator]
guest   [administrator]
...done.
使用浏览器打开http://localhost:15672 访问Rabbit Mq的管理控制台，使用刚才创建的账号登陆系统：

安装API



pip install pika
or
easy_install pika
or
源码
https://pypi.python.org/pypi/pika

　　使用API操作RabbitMQ

基于Queue实现生产者消费者模型





#!/usr/bin/env python
# -*- coding:utf-8 -*-
import Queue
import threading

message = Queue.Queue(10)

def producer(i):
while True:
message.put(i)

def consumer(i):
while True:
msg = message.get()

for i in range(12):
t = threading.Thread(target=producer, args=(i,))
t.start()
for i in range(10):
t = threading.Thread(target=consumer, args=(i,))
t.start()
内存queue消费者 生产者　　对于RabbitMQ来说，生产和消费不再针对内存里的一个Queue对象，而是某台服务器上的RabbitMQ Server实现的消息队列。



channel.queue_declare(queue='hello')  #创建队列名称为hello 客户端也可以创建，如果服务端没有创建客户端就创建，客户端创建了服务端就不用创建
# 这里的exchange为空时exchange不工作，单独的客户端服务端就只用一个队列来通信  注意body主体信息
# noack是当客户端取出消息时候 回调函数处理数据时候如果时间长，我们这个参数就是不让其等待，数据不再队列，出问题，数据就没了。如果要求数据性高我们就设置为false 可以让回调函数处理完了，即使中间崩溃了，下次启动取数据还会取的到。



routing_key='hello', # 这里的定义队列的名字 这时候是单向一个队列 交换机是没有作用的




#!/usr/bin/env python
import pika
# ######################### 服务端生产者 #########################
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))   #链接rabbitmq
channel = connection.channel() # 生成socket句柄
channel.queue_declare(queue='hello')  #创建队列名称为hello 客户端也可以创建，如果服务端没有创建客户端就创建，客户端创建了服务端就不用创建
channel.basic_publish(exchange='',
routing_key='hello',
body='Hello World!') # 这里的exchange为空时exchange不工作，单独的客户端服务端就只用一个队列来通信  注意body主体信息
print(" [x] Sent 'Hello World!'")  
connection.close()
#!/usr/bin/env python
import pika
# ########################## 客户端消费者 ##########################
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))  #客户端链接 rabbitmq
channel = connection.channel() #客户端创建句柄
channel.queue_declare(queue='hello')  #客户端创建，如果服务端没有的话就生效
def callback(ch, method, properties, body):  #这里的参数body是服务端的body内容主体
print(" [x] Received %r" % body)
channel.basic_consume(callback,
queue='hello',
no_ack=True)  # noack是当客户端取出消息时候 回调函数处理数据时候如果时间长，我们这个参数就是不让其等待，数据不再队列，出问题，数据就没了。如果要求数据性高我们就设置为false 可以让回调函数处理完了，即使中间崩溃了，下次启动取数据还会取的到。
print('

Waiting for messages. To exit press CTRL+C')
channel.start_consuming()


重要参数介绍：
　　（1）acknowledgment 消息不丢失
　　no-ack ＝ False，如果消费者遇到情况(its channel is closed, connection is closed, or TCP connection is lost)挂掉了，那么，RabbitMQ会重新将该任务添加到队列中。
　　（2）durable   消息不丢失  rabbitmq 的持久化





持久化
你花了大量的时间来创建队列、交换机和绑定，然后，服务器程序挂了。你的队列、交换机和绑定怎么样了？还有，放在队列里面但是尚未处理的消息们呢？
如果你是用默认参数构造的这一切的话，那么，他们都灰飞烟灭了。RabbitMQ重启之后会干净的像个新生儿。你必须重做所有的一切，亡羊补牢，如何避免将来再度发生此类杯具？
队列和交换机有一个创建时候指定的标志durable。durable的唯一含义就是具有这个标志的队列和交换机会在重启之后重新建立，它不表示说在队列当中的消息会在重启后恢复。那么如何才能做到不只是队列和交换机，还有消息都是持久的呢？
但是首先需要考虑的问题是：是否真的需要消息的持久化？如果需要重启后消息可以回复，那么它需要被写入磁盘。但即使是最简单的磁盘操作也是要消耗时间的。所以需要衡量判断。
当你将消息发布到交换机的时候，可以指定一个标志“Delivery Mode”（投递模式）。根据你使用的AMQP的库不同，指定这个标志的方法可能不太一样。简单的说，就是将Delivery Mode设置成2，也就是持久的（persistent）即可。一般的AMQP库都是将Delivery Mode设置成1，也就是非持久的。所以要持久化消息的步骤如下：
将交换机设成 durable。
将队列设成 durable。
将消息的 Delivery Mode 设置成2 。
绑定（Bindings）怎么办？绑定无法在创建的时候设置成durable。没问题，如果你绑定了一个durable的队列和一个durable的交换机，RabbitMQ会自动保留这个绑定。类似的，如果删除了某个队列或交换机（无论是不是durable），依赖它的绑定都会自动删除。
注意：
RabbitMQ 不允许你绑定一个非坚固（non-durable）的交换机和一个durable的队列。反之亦然。要想成功必须队列和交换机都是durable的。
一旦创建了队列和交换机，就不能修改其标志了。例如，如果创建了一个non-durable的队列，然后想把它改变成durable的，唯一的办法就是删除这个队列然后重现创建。因此，最好仔细检查创建的标志。
在Windows上安装Rabbit MQ 指南，最好的是这篇《Rabbit MQ Windows Installation guide》，其中还包括了使用.NET RabbitMQ.Client Nuget 包访问Rabbit MQ的示例代码。
持久化　　生产者：1  服务端 设置持久化创建队列时候durable=True 让rabbitmq队列 即使rabbitmq服务器崩溃也没事  2 设置 属性  投递模式  delivery_mode=2, # make message persistent



#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
channel = connection.channel()
# make message persistent
channel.queue_declare(queue='hello', durable=True)
channel.basic_publish(exchange='',
routing_key='hello',
body='Hello World!',
properties=pika.BasicProperties(
delivery_mode=2, # make message persistent
))
print(" [x] Sent 'Hello World!'")
connection.close()

　　消费者： 取完消息消费者崩溃 队列消息不消失　　 1 ch.basic_ack(delivery_tag = method.delivery_tag)   2 no_ack=False



#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
channel = connection.channel()
# make message persistent
channel.queue_declare(queue='hello', durable=True)

def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
import time
time.sleep(10)
print 'ok'
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_consume(callback,
queue='hello',
no_ack=False)
print('

Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

　　（3）消息获取顺序  可以有多个消费者 依次取
　　默认消息队列里的数据是按照顺序被消费者拿走，例如：消费者1 去队列中获取 奇数 序列的任务，消费者1去队列中获取 偶数 序列的任务。
　　channel.basic_qos(prefetch_count=1) 表示谁来谁取，不再按照奇偶数排列   这里可以利用消费者睡得性能好谁先来就先拿就行。类似权重



#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
channel = connection.channel()
# make message persistent
channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
import time
time.sleep(10)
print 'ok'
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
queue='hello',
no_ack=False)
print('

Waiting for messages. To exit press CTRL+C')
channel.start_consuming()


二 上述 exchange 为空时单独一个生产者一个消费者 下面我们就介绍下exchange 的常用三种工作模式　　

3  exchange工作模型（fanout，direct，topic）

3.1 发布订阅  # 可以有更多个订阅者客户端

　　发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中。
　　exchange type = fanout
　　发布者：



#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs',  
type='fanout')  #订阅发布模式 ，由路由器名称logs来选择队列分发消息
message = ' '.join(sys.argv[1:]) or "info: Hello World!" # 消息为实行脚本或者不跟参数
channel.basic_publish(exchange='logs',
routing_key='',
body=message) # 绑定路由器 跟发送信息
print(" [x] Sent %r" % message)
connection.close()
发布者

　　订阅者：



#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs',
type='fanout') #路由器
result = channel.queue_declare(exclusive=True) #生成队列
queue_name = result.method.queue #生成队列的名称 自己的规则
channel.queue_bind(exchange='logs',
queue=queue_name) #将队列名称与路由器绑定
print('

Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r" % body)
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)  #传入队列名称跟回调函数
channel.start_consuming()
订阅者
订阅者


  3.2 关键字发送

　　exchange type = direct
　　之前事例，发送消息时明确指定某个队列并向其中发送消息，RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。
　　一个路由器可以有多个关键字
　　发布者：



#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
type='direct')
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',
routing_key=severity,
body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()
生产者

　　消费者：



#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
type='direct')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
severities = sys.argv[1:]
if not severities:
sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
sys.exit(1)
for severity in severities:
channel.queue_bind(exchange='direct_logs',
queue=queue_name,
routing_key=severity)
print('

Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()
消费者


     3.3 模糊匹配

　　exchange type = topic
　　在topic类型下，可以让队列绑定几个模糊的关键字，之后发送者将数据发送到exchange，exchange将传入”路由值“和 ”关键字“进行匹配，匹配成功，则将数据发送到指定队列。

• # 表示可以匹配 0 个 或 多个 单词
  
• *  表示只能匹配 一个 单词
  

发送者路由值              队列中
old.boy.python          old.*  -- 不匹配
old.boy.python          old.#  -- 匹配

　　生产者：



#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',
type='topic')
routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='topic_logs',
routing_key=routing_key,
body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()
生产者
生产者

　　消费者：



#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',
type='topic')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
binding_keys = sys.argv[1:]
if not binding_keys:
sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
sys.exit(1)
for binding_key in binding_keys:
channel.queue_bind(exchange='topic_logs',
queue=queue_name,
routing_key=binding_key)
print('

Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()
消费者
消费者