RabbitMQ（python实现）学习之三：Routing（接收端接收固定类型消息）
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 if len(sys.argv) > 1 else 'info'#手动输入发送消息的routing_key
message = ' '.join(sys.argv) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',
routing_key=severity,
body=message)
print " 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
if not severities:
print >> sys.stderr, "Usage: %s " % \
(sys.argv,)
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 " %r:%r" % (method.routing_key, body,)
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()
页:
[1]