上节介绍了通过广播简单的将日志推送到所有的consumer。本节介绍通过定义规则，让consumer有选择的接受日志。例如：只接收error级别的日志，不接受info级别的日志。
- Bindings
binding是关联 exchange 和 queue 的，也可以认为是此 queue 对此 exchange 的消息感兴趣。
Bindings有另外一个参数: routing_key，此参数依赖于 exchange type，如果是 fanout，则忽略此参数。
channel.queue_bind(exchange=exchange_name
queue=queue_name,
routing_key='black')
- Direct exchange
之前的日志系统，是将消息全部广播到所有的consumer，现在需要对于日志的严重性级别来做规则。例如：需要将 critical error 级别的消息写入磁盘，而不写入 warning 和 info 级别的日志消息。
fanout exchange，仅仅是在广播消息，缺乏灵活性。使用 direct exchange代替，routing算法很简单 -- 一个队列中的消息准确匹配到 routing_key 的值。示例图：
如图，direct exchange "X" 和两个队列绑定。第一个队列绑定orange，第二个队列绑定black和green。if消息的 routing_key 为orage，则exchange会将消息路由到队列Q1，elif消息的 routing_key 为black或green，则exchange会将消息路由到队列Q2，else消息将会被丢弃。
- Multiple bindings
绑定多个队列，类似 fanout exchange，将消息推送到所有匹配的队列。routing_key为black的消息将同时交付到Q1和Q2队列。
- Emitting logs
- 创建exchange
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
- 发送消息
channel.basic_publish(exchange='direct_logs',
routing_key=severity, # severity可以是info、warning、error
body=message)
- Subscribing
与之前不同的是，为接受的日志的每个级别，创建一个新的binding。
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
for severity in serverities:
channel.queue_bind(exchange='direct_logs',
queue=queue_name,
routing_key=severity)
- 完整样例
- 编辑 emit_log_direct.py 程序，用来提交日志
#!/usr/bin/env python3
# coding=utf-8
import pika
import sys
message = ' '.join(sys.argv[1:]) or 'Hello World!'
# 获取用户终端输入的severity
severity = sys.argv[1] if len(sys.argv) > 2 else 'info'
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()
# 使用 direct exchange，名称为"direct_logs"
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
channel.basic_publish(exchange='direct_logs',
routing_key=severity,
body=message,
)
print("[x] Sent '%s'" % message)
connection.close()
- 编辑 receive_logs_direct.py 程序，用来接受日志
#!/usr/bin/env python3
# coding=utf-8
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()
# 声明exchange，使用 direct type， 名称为"direct_logs"
channel.exchange_declare(exchange="direct_logs", exchange_type="direct")
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
# 获取终端输入的severity，可指定多个，之后使用for循环接收
severities = sys.argv[1:]
if not severities:
sys.stderr.write('Usage: %s [info] [warning] [error]\n' % __name__)
sys.exit(1)
def callback(ch, method, properties, body):
print('[x] Received %r' % body)
# 循环接收用户指定的severities
for severity in severities:
channel.queue_bind(exchange='direct_logs',
queue=queue_name,
routing_key=severity)
channel.basic_consume(callback, queue=queue_name, no_ack=True)
print('[*] Waiting for messgaes. To exit press CTRL+C')
channel.start_consuming()
- 执行
# 提交日志
> python3 emit_log_direct.py test_info
> python3 emit_log_direct.py warning test_warning
# consumer1接收日志
> python3 receive_logs_direct.py info warning
[*] Waiting for messgaes. To exit press CTRL+C
[x] Received b'test_info'
[x] Received b'warning test_warning'
# consumer2接收日志
> python3 receive_logs_direct.py warning
[*] Waiting for messgaes. To exit press CTRL+C
[x] Received b'warning test_warning'
参考文档： http://www.rabbitmq.com/tutorials/tutorial-four-python.html
