python之路——RabbitMQ

  RabbitMQ  

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

RabbitMQ也分为发送端和接收端，先来看一个简单发送接收例子：

#发送端----------->
import pika

c = pika.BlockingConnection(pika.ConnectionParameters("localhost"))  #先实例化一个socket
chanel1 = c.channel()   #声明一个管道
chanel1.queue_declare(queue="c3")

chanel1.basic_publish(exchange = "",
                     routing_key = "c3",   #queue的名字
                     body = "hello world!!!"  #发送的信息
)

print("Send Done")
c.close()

接收端-------------->
import pika,time

c = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
chanel1 = c.channel()   #声明一个管道
chanel1.queue_declare(queue="c3")

def callback(ch,method,porperties,body):
    print(ch,method,porperties)
    time.sleep(10)
    print("----->",body)

chanel1.basic_consume(#消费消息
                    callback,  #如果收到消息，就调用callback函数来处理消息
                    queue = "c3",
                    # no_ack = True   #no acknowledgement 确认信号,生产者收到确认信号后才会删除消息
)
print("Recv Done")
chanel1.start_consuming()

在这种情况下，直接实现简单的发送和接收，其中接收端没有声明no_ack = True，即使接收端断开重连，消息依然存在在队列里，接收端依然会接收到消息。如果加上no_ack = True，接收端会轮流接收发送端发送的消息。

  （队列/消息）持久化  

持久化指的是，即使断开RabbitMQ服务，队列和消息依然存在，可以使用durable=True使队列持久化，properties=pika.BasicProperties(delivery_mode = 2)可以使消息持久化，单单只使用消息持久化而不使用队列持久化是没有的，看代码：

发送端----------->
import pika

c = pika.BlockingConnection(pika.ConnectionParameters("localhost"))  #先实例化一个socket
chanel1 = c.channel()   #声明一个管道
chanel1.queue_declare(queue="c4",durable=True)   #durale=True 使队列持续化，即使关闭RabbitMQ，再打开队列依然存在

chanel1.basic_publish(exchange = "",
                     routing_key = "c4",   #queue的名字
                     body = "hello world!!!",  #发送的信息
                     properties = pika.BasicProperties(delivery_mode=2)   #消息持久化，但是如果没有持久化队列则没用
)

print("Send Done")
c.close()

接收端------------->
import pika,time

c = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
chanel1 = c.channel()   #声明一个管道
chanel1.queue_declare(queue="c4",durable=True)  #durale=True 使队列持续化，即使关闭RabbitMQ，再打开队列依然存在

def callback(ch,method,porperties,body):
    print(ch,method,porperties)
    time.sleep(10)
    print("----->",body)
    chanel1.basic_ack(delivery_tag=method.delivery_tag)  #这行代码表示：手动确认

chanel1.basic_qos(prefetch_count=1)  #消费端一次处理一条信息，处理完再处理下一条
chanel1.basic_consume(#消费消息
                    callback,  #如果收到消息，就调用callback函数来处理消息
                    queue = "c4",
                    # no_ack = True   #no acknowledgement 确认信号,生产者收到确认信号后才会删除消息
)
print("Recv Done")
chanel1.start_consuming()

  PublishSubscribe(消息发布订阅)  

前面写的例子都是一对一的形式，即消息只能发送到指定的queue里，但有些时候你想让你的消息被所有的Queue收到，类似广播的效果，这时候就要用到exchange了，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

  fanout  ---------->实时接收广播：

发送端

import pika,sys

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

# channel.queue_declare(queue="fanout_queue",durable=True)
channel.exchange_declare(exchange="logs",type="fanout")   #声明广播的类型 fanout

message = " ".join(sys.argv[1:]) or "info: hello world!!"
channel.basic_publish(exchange = "logs",
                      # routing_key = "fanout_queue",
                      routing_key = "",
                      body = message,
                      )
print("Send Done")
connection.close()

发送端要先声明exchange的名称和类型，此时的routing_key = “” 值，不用声明队列

接收端

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# author :Japhi

import pika

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

# channel.queue_declare(queue="fanout_queue",durable=True)
channel.exchange_declare(exchange="log",type="fanout")
result = channel.queue_declare(exclusive=True) #不指定queue的名字，Rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
queue_name = result.method.queue

channel.queue_bind(exchange="logs",queue=queue_name)   #监听（广播）exchange=“logs”，实时接收消息

print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
    print(" recv",body)

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

def callback(ch, method, properties, body):
    print("Recv:",body)
    print("Recv Done")
    ch.basic_ack(delivery_tag=method.delivery_tag)


channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
                      queue='fanout_queue')

channel.start_consuming()

接收端还要额外生成一个随机队列。　　

有选择的接收消息(exchange type=direct)

RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。

发送端

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# author :Japhi

import pika,sys

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

channel.exchange_declare(exchange="direct_log",type="direct")

severity = sys.argv[1] if len(sys.argv) > 1 else "info"
msg = " ".join(sys.argv[2:]) or "hello world!!"
channel.basic_publish(exchange = "direct_log",
                      routing_key = severity,
                      body = msg
)
print("send done:",severity,msg)
connection.close()

接收端

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# author :Japhi

import pika,sys

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

channel.exchange_declare(exchange="direct_log",type="direct")

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

severities = sys.argv[1:]
print(severities)
if not severities:    #首先要通过用户外部输入routing_key
    sys.stderr.write("exit")
    sys.exit()

for severity in severities:
    channel.queue_bind(exchange="direct_log",
                       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()

更细致的消息过滤(exchange type=topic)

发送端

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# author :Japhi

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

接收端

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# author :Japhi

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[0:]   #需要用户外部输入
print(binding_keys)
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()

使用cmd进入接收端，外部输入info japhi，运行发送端，接收端收不到任何数据；外部输入anonymous.info使，接收端就可以收到数据：[x] 'anonymous.info':b'Hello World!'

  RPC  

前面讲的都是一个发送端，一个接收端，有没有可能实现接收端接收到消息后返回数据给发送端实现数据的交互呢？答案是可以的

发送端使用rpc_queue队列发消息给接收端，接收端可以利用另外一个队列reply_to来返回消息给发送端

客户端

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# author :Japhi

import pika,time
import uuid

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) #不指定queue的名字，Rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
        self.callback_queue = result.method.queue
        self.channel.basic_consume(self.on_response, no_ack=True,
                                   queue=self.callback_queue)

    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue',
                                   properties=pika.BasicProperties(
                                       reply_to=self.callback_queue,
                                       correlation_id=self.corr_id,
                                   ),
                                   body=str(n))
        while self.response is None:
            self.connection.process_data_events()
            print("no msg.....")
            time.sleep(0.5)
        return int(self.response)

fibonacci_rpc = FibonacciRpcClient()

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

服务端

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# author :Japhi

import pika
import time

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

channel = connection.channel()

channel.queue_declare(queue='rpc_queue')


def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n - 1) + fib(n - 2)


def on_request(ch, method, props, body):
    n = int(body)

    print(" [.] fib(%s)" % n)
    response = fib(n)

    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id= 
                                                         props.correlation_id),
                     body=str(response))
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(on_request, queue='rpc_queue')

print(" [x] Awaiting RPC requests")
channel.start_consuming()

客户端发送fib(5)，服务端收到数据，调用fib函数，返回结果13给客户端，客户端输出结果