Python之RabbitMQ

RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。

MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消 息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

RabbitMQ安装

安装配置epel源
   $ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm
 
安装erlang
   $ yum -y install erlang
 
安装RabbitMQ
源码安装-http://www.rabbitmq.com/download.html
yum安装-$ yum -y install rabbitmq-server
注意：service rabbitmq-server start/stop

安装API

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

 使用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(5):
    t = threading.Thread(target=producer, args=(i,))
    t.start()

for i in range(2):
    t = threading.Thread(target=consumer, args=(i,))
    t.start()
'''
生产者和消费者启动后，生产者像Q里上传消息，消费者下载消息 
'''

对于RabbitMQ来说，生产和消费不再针对内存里的一个Queue对象，而是某台服务器上的RabbitMQ Server实现的消息队列。rabbitmq就是单独的一个进程充当队列

#!/usr/bin/env python
import pika

# ######################### 生产者 #########################

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) #链接rabbitmq,默认端口
channel = connection.channel() # 创建一个隧道（服务端和客户端不能直接和rabbitmq直接通信,这样会在多交互的情况下容易乱）

channel.queue_declare(queue='hello')# 在隧道中创建和客户端交互的队列

channel.basic_publish(exchange='', # 讲消息发送到交换器，由交换器发送到指定的队列中，默认为交换器
                      routing_key='hello', # 向队列中发送数据
                      body='Hello World!') # 发送的数据内容
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): # 接受消息
    print(" [x] Received %r" % body)

channel.basic_consume(callback,#接收到消息后交给回调函数处理
                      queue='hello',#队列名称
                      no_ack=True) #no_ack=true表示在消费者接收到消息后断开链接了，队列里不在存储这个消息，为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会重新将该任务添加到队列中。下面将消费者稍作修改

#!/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): # 接受消息
    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()# 启动 接收消息
'''
在消费者启动后接受到消息，不等待10秒直接关闭，那么在下次启动还会继续接收这个消息，等待10秒后返回ok再次启动，这个消息才被消费
'''

2、durable   消息不丢失

durable表示持久化q，当rabbitmq挂掉后，之前定义的非持久化的q就会失效，消费者如果不重新生成q的话就会报错，delivery_mode=2表示消息持久化，实例

生产者

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# make message persistent
channel.queue_declare(queue='hello1',durable=True)#durable定义持久化q（这里为什么不用之前的hello，是因为之前已经定义了hello为不持久化，所以你必须重新生成一个q，设置持久化或者删除之前的hello）

channel.basic_publish(exchange='',
                      routing_key='hello1',
                      body='Hello World!',
                      properties=pika.BasicProperties(
                        delivery_mode=2, #持久化消息
                      ))
print(" [x] Sent 'Hello World!'")
connection.close()

消费者

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# make message persistentchannel.queue_declare(queue='hello1',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='hello1',
                      no_ack=False)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

3、消息公平分发

如果Rabbit只管按顺序把消息发到各个消费者身上，不考虑消费者负载的话，很可能出现，一个机器配置不高的消费者那里堆积了很多消息处理不完，同时配置高的消费者却一直很轻松。为解决此问题，可以在各个消费者端，配置perfetch=1,意思就是告诉RabbitMQ在我这个消费者当前消息还没处理完的时候就不要再给我发新消息了

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) #消息的公平分发参数 =1 表示最多同时处理一个消息

channel.basic_consume(callback,
                      queue='hello',
                      no_ack=False)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

PublishSubscribe(消息发布订阅)　

之前的例子都基本都是1对1的消息发送和接收，即消息只能发送到指定的queue里，但有些时候你想让你的消息被所有的Queue收到，类似广播的效果，这时候就要用到exchange了，

exchange type = fanout

Exchange在定义的时候是有类型的，以决定到底是哪些Queue符合条件，可以接收消息

fanout: 所有bind到此exchange的queue都可以接收消息
direct: 通过routingKey和exchange决定的那个唯一的queue可以接收消息
topic:所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息

　　 表达式符号说明：#代表一个或多个字符，*代表任何字符
      例：#.a会匹配a.a，aa.a，aaa.a等
          *.a会匹配a.a，b.a，c.a等
     注：使用RoutingKey为#，Exchange Type为topic的时候相当于使用fanout　

headers: 通过headers 来决定把消息发给哪些queue

发布者

exchange type = fanout

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='logs', #定义交换器的名称
                         type='fanout') #fanout 所有绑定到这个交换器的q都能接收消息

channel.basic_publish(exchange='logs', # 不在向q中发送消息，直接向定义的交换器里发送消息
                      routing_key='',
                      body="hello")
print(" [x] Sent %r hello" )
connection.close()

订阅者

#_*_coding:utf-8_*_
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名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
queue_name = result.method.queue # 获取q名称

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()

关键字发送

 exchange type = direct

之前事例，发送消息时明确指定某个队列并向其中发送消息，RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。

发布者

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' #这句话拆分成下面
# if len(sys.argv) > 1:
#     severity = sys.argv[1]
# else:
#     severity = "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()
'''
chen@chen:~/python/Learning/day10$ python3.5 rb_p2.py 
 [x] Sent 'info':'Hello World!'
chen@chen:~/python/Learning/day10$ python3.5 rb_p2.py info nihao 
 [x] Sent 'info':'nihao'

'''

订阅者

#!/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]
" % 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)) # method.routing_key获取关键字信息

channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

channel.start_consuming()
'''
chen@chen:~/python/Learning/day10$ python3.5 rb_c2.py
Usage: rb_c2.py [info] [warning] [error]

chen@chen:~/python/Learning/day10$ python3.5 rb_c2.py info
 [*] Waiting for logs. To exit press CTRL+C
 [x] 'info':b'nihao'
 [x] 'info':b'Hello World!'
 [x] 'info':b'nihao'

chen@chen:~/python/Learning/day10$ python3.5 rb_c2.py info error
 [*] Waiting for logs. To exit press CTRL+C
 [x] 'info':b'nihao'
 [x] 'error':b'nihao'

'''

6、模糊匹配

 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')

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]...
" % 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()

#!/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()

Remote procedure call (RPC)

###########sub订阅端###########
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost')) # 链接主机
channel = connection.channel() # 创建隧道
channel.queue_declare(queue='rpc_queue') #创建q名称

def fib(n): # 计算函数
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n-1) + fib(n-2)
#服务端接收到客户端的命令后进行计算。计算完在通过初始定义的q返回
def on_request(ch, method, props, body): # 接收客户端的数据
    n = int(body) # 接收到client的数据
    print("props",props.reply_to) # props 接收到的是client生成的随机Q
    print("reply_to",props.correlation_id) # 客户端通过UUID生产的随机Q
    print(" [.] fib(%s)" % n)
    response = fib(n) # 调用计算函数
    # 向client发送计算完的数据（发布方）
    ch.basic_publish(exchange='', # 使用默认交换器
                     routing_key=props.reply_to, #client自动生成的随机q,准确的说是关键字（返回q）
                     properties=pika.BasicProperties(correlation_id =props.correlation_id),body=str(response)) #props.correlation_id 表示客户端发送的另一个随机Q
    ch.basic_ack(delivery_tag = method.delivery_tag) # 消息持久化

channel.basic_qos(prefetch_count=1) #公平分发（表示如果我这个数据没处理完，将不在接收新的数据）
channel.basic_consume(on_request, queue='rpc_queue') # 接收rpc队列中的数据（订阅方），on_request表示回调函数CALLBACK，回调函数中处理了数据，又充当了发布方

print(" [x] Awaiting RPC requests") # 打印开始接收消息
channel.start_consuming() # 开始接收

#############pub发送端##########
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
import uuid # uuid模块在这里用作生产一串随机数当做返回的Q

class FibonacciRpcClient(object):
    def __init__(self): # 初始化配置
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(
                host='localhost')) # 创建链接
        self.channel = self.connection.channel() # 建立隧道

        result = self.channel.queue_declare(exclusive=True) #生产随机Q ，链接断开后自动删除这个Q
        self.callback_queue = result.method.queue#获取随机Q（这里指的是发送给server端的关键字，server通过这个关键字当Q）
        print(self.callback_queue)
        self.channel.basic_consume(self.on_response, no_ack=True,queue=self.callback_queue) #（订阅方），用于接收Q中的数据 【这里订阅方是写到构造方法了，初始化的时候就先订阅这个Q。】

    def on_response(self, ch, method, props, body): #回调方法
        if self.corr_id == props.correlation_id: # 判断如果接收过来的Q等于UUID的随机Q则数据正确
            self.response = body

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4()) # 设定一个随机数
        print(self.corr_id)
        self.channel.basic_publish(exchange='', # 交换器
                                   routing_key='rpc_queue', #client的 发布方的Q
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue, # reply_to 表示回调队列（Q）,客户端通过我传给你的这个Q给我返回消息，这个Q在初始化的时候已经给服务端了
                                         correlation_id = self.corr_id, # 传入UUID，服务端返回的时候带上UUID，客户端就能确认是自己本次要接收的消息了，（为什么加这个呢？个人理解：如果由N个客户端向rpcQ里发消息，都要求返回，或者你发了多个消息都要求返回，那返回来的结果你如何判断一一对应呢）
                                         ),# 持久化消息
                                   body=str(n))
        while self.response is None:
            self.connection.process_data_events() # 如果response的结果为NONE，就一直等待接收
        return int(self.response) # 返回server1计算的结果

fibonacci_rpc = FibonacciRpcClient()

print(" [x] Requesting fib(30)")
response = fibonacci_rpc.call(30)
print(" [.] Got %r" % response)