交换机的路由功能
注意！交换机为匿名交换机时，routing_key指的是队列名，这只是一个特例！这个参数的本意是路由键名！
本篇实现这样的功能：
- 生产者投递不同类型的消息到交换机，消息类型用交换机的路由参数指定；
- 消费者根据交换机不同的路由获取并处理消息
具体场景：生产者可以投递不同级别的日志，消费者根据自己的需求去捕获不同类型的日志。如error级别的日志才写入到文件，所有级别的日志都显示到终端。
路由
代码实现
新建 emit_log_direct.py
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs', # 声明交换机
exchange_type='direct') # 交换机类型
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()
新建 receive_logs_direct.py
import pika
import sys
connection = 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)
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()
执行结果
开启消费者1
python receive_logs_direct.py warning error > logs_from_rabbit.log # 将 warning 和 error 级别的日志写入文件
开启消费者2
python receive_logs_direct.py info warning error # 将所有级别的日志输出到终端
开启生产者
python emit_log_direct.py error "Run. Run. Or it will explode."
