简介

MessageQueue用于解决跨进程、跨线程、跨应用、跨网络的通信问题。

RabbitMQ使用erlang开发，在windows上使用时要先安装erlang。

官方的示例比较容易理解，可以点这里去看看。

结构

生产者 ---> exchange ---> queue ---> 消费者。

生产者负责提供消息，exchange负责分发消息到指定queue，queue存储消息（默认临时，可设置持久化），消费者接收处理消息。

基本模型

流程：

1. 建立到rabbitmq的连接
2. 建立通道
3. 声明使用的队列（生产者和消费者都要声明，因为不能确定两者谁先运行）
4. 生产/消费
5. 持续监听/关闭连接

python中使用pika模块来处理与rabbitmq的连接。

# 生产者
import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')
)
channel = connection.channel()
r = channel.queue_declare(queue='name', exclusive=False, durable=False) # exclusive设置True是随机生成一个queue名字并返回，durable设置True是队列持久化
queue_name = r.method.queuechannel.basic_publish(exchange = '', # 使用默认分发器routing_key = queue_name,properties = pika.BasicProperties(delivery_mode = 2 # 这个参数用于设置消息持久化),body = [data]
)connection.close()# 消费者
import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')
)
channel = connection.channel()
r = channel.queue_declare(queue='name', exclusive=False, durable=False)
queue_name = r.method.queuedef callback(channel, method, properties, body):pass# channel.basic_ack(delivery_tag = method.delivery_tag) 在回调函数最后调用手工应答，表示消息处理完毕，queue可以删除消息了channel.basic_consume(callback, # 这是个回调函数，接收生产者发来的bodyqueue = queue_name,no_ack = True # 设置True表示消息一经获取就被从queue中删除，如果这时消费者崩溃，则这条消息将永久丢失，所以去掉这个属性，在回调函数中手工应答比较安全
)channel.basic_qos(prefetch_count = [num]) # 设置消费者的消费能力，数字越大，则说明该消费者能力越强，往往与设备性能成正比channel.start_consuming() # 阻塞模式获取消息
# connection.process_data_events() 非阻塞模式获取消息

发布订阅模型

类似收音机广播，订阅者只要打开收音机就能收听信息，但接收不到它打开之前的消息。

包括发布订阅模型以及接下来的一些模型，都是通过exchange和routing_key这两个属性来控制的。直接用官网的源码来做注释。

流程：

1. 发布者设置发布频道
2. 收听者配置频道信息
3. 收听者通过随机queue绑定频道接收消息

# 发布者
#!/usr/bin/env python
import pika
import sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()# 创建一个命名exchange，并设置其type为fanout，表示广播
channel.exchange_declare(exchange='logs',exchange_type='fanout')# 从命令行接收输入
message = ' '.join(sys.argv[1:]) or "info: Hello World!"# 由于是广播模式，任意消费者只要设置同样的exchange，就能以任意queue来接收消息，所以这里routing_key置空
channel.basic_publish(exchange='logs',routing_key='',body=message)
print(" [x] Sent %r" % message)
connection.close()# 收听者
#!/usr/bin/env python
import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()# 这里使用同样的exchange配置，就像调节收音机频道
channel.exchange_declare(exchange='logs',exchange_type='fanout')# 在基础模型中提到过，设置exclusive=True表示生成随机的queue
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue# 生成了queue，还要将它与exchange进行绑定，这样消息才能通过exchange进入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()

路由/级别模型

将消息发送到指定的路由处，类似于logging模块的分级日志消息。

主要利用channel.queue_bind(routing_key=[route])这个方法，来为queue增加路由。

流程：

1. 生产者向指定路由发送消息
2. 消费者绑定路由
3. 根据路由接收到不同的消息

# 生产者
#!/usr/bin/env python
import pika
import sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()# 同样使用命名exchange，主要是type为direct
channel.exchange_declare(exchange='direct_logs',exchange_type='direct')# 将命令行输入的路由作为接收消息的queue的属性，只有匹配的才能接收到消息
severity = sys.argv[1] if len(sys.argv) > 2 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 sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()channel.exchange_declare(exchange='direct_logs',exchange_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)# 对该消费者的queue绑定路由
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()

细目模型/更细致的划分

这个模型比前几种更强大，但是原理与路由模型是相同的。

如果routing_key='#'，它就相当于发布订阅模式，向所有queue发送消息，如果routing_key值中不包含*，#，则相当于路由模型。

该模型主要是实现了模糊匹配。

# 生产者
#!/usr/bin/env python
import pika
import sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()channel.exchange_declare(exchange='topic_logs',exchange_type='topic')routing_key = sys.argv[1] if len(sys.argv) > 2 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 sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()channel.exchange_declare(exchange='topic_logs',exchange_type='topic')result = channel.queue_declare(exclusive=True)
queue_name = result.method.queuebinding_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()

RPC模型

前面的几种模型都只能是一端发消息，另一端接收，RPC模型实现的就是单端收发功能。

主要是通过两个队列实现，一个发，一个收。

流程：

1. 客户端发送消息到约定队列，并且附带返回队列的名称和验证id
2. 服务器接到消息，将处理过的消息发送给指定队列并附带验证id
3. 客户端接到消息先验证id，通过则处理消息

# 服务器
#!/usr/bin/env python
import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))channel = connection.channel()channel.queue_declare(queue='rpc_queue')def fib(n):if n == 0:return 0elif n == 1:return 1else: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()# 客户端
#!/usr/bin/env python
import pika
import uuidclass 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)self.callback_queue = result.method.queueself.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 = bodydef call(self, n):self.response = Noneself.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()return int(self.response)fibonacci_rpc = FibonacciRpcClient()print(" [x] Requesting fib(30)")
response = fibonacci_rpc.call(30)
print(" [.] Got %r" % response)