RabbitMQ（python实现）学习之三：Routing（接收端接收固定类型消息）

samsungsamsung

　　1.1本博文简介
　　前面降到了将消息广播给多个Consumer，本博文主要是对消息进行固定点的传输，每个Consumer从消息队列中获取固定类型的消息。
　　1.2Bindings
　　在前面的代码中，咱们写过这样的代码：



channel.queue_bind(exchange=exchange_name,queue=queue_name)
　　一个binding 是exchange与queue的关联纽带。
　　binding可以附加一个额外的参数routing_key，来避免basic_publish的混乱。一下是具体代码：



channel.queue_bind(exchange=exchange_name,queue=queue_name,routing_key='black')
　　1.3Direct exchange（direct类型的exchange）
　　前面博文中讲到，消息被广播至所有的consumers，现在我们想根据自己需要，对消息进行分类，不同的消息传送给不同的consumer。
　　前面我们将exchange的类型设置为fanout，这种类型灵活性不足，只会无脑的广播。
　　下面我们将exchange的type设置为direct，此时，到达exchange的消息，根据routing_key的不同，将消息发送给相应的队列queue，这要就把消息进行了定制化接收。具体模型如下：

　　由图中我们可以看到，exchange的type为direct，有两个的queue。Q1通过routing_key=orange的binding与exchange相连，而Q2通过两个binding，routing_key=gren和routing_key=black，与exchange相连。
　　对于routing_key=orange的消息，会发送至Q1，而对于routing_key=black和routing_key=green的消息会被发送至Q2，其他的routing_key的消息会被遗弃掉。
　　1.4代码实现
　　首先创建一个exchange，代码如下



channel.exchange_declare(exchange ='direct_logs',type='direct')
　　接下来我们将要发送消息：



channel.basic_publish(exchange='direct_logs',routing_key=severity,body=message)
　　severity后面程序会讲到，取值为‘info’、‘warning’、‘error’
　　
　　订阅消息，我们要创建binging来接收消息：



result = channel.queue_declare(exclusive=True)
queue_name=result.method.queue
for severity in severities:
channel.queue_bind(exchange='direct_logs',queue=queue_name,routing_key=severity)
　　1.5完整代码实现
　　模型如下：

　　send.py代码：



#!/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'  #手动输入发送消息的routing_key
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()
　　receive.py代码实现：



#!/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:
print >> sys.stderr, "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,)
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()