RabbitMQ direct广播模式

xuxiaohui9216

一、前言

　　之前我们谈到的是1对多以广播的方式，发送给所有的消费者。那如果消费者可以进行过滤，有选择的进行接收我想要的消息。下面我们就来学习第二种广播方式，即  direct广播模式
　　RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。

二、direct广播模式逻辑图

三、direct广播模式代码实现

3.1、生产者(publisher)



import pika,sys

connection = pika.BlockingConnection(pika.ConnectionParameters
("localhost"))
channel = connection.channel()
#定义direct类型的exchange
channel.exchange_declare(exchange="direct_logs",
exchange_type="direct")
#定义重要程度，定义什么级别的日志
severity = sys.argv[1] if len(sys.argv) > 1 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()

　　注意了：这边routing_key是只的你过滤的条件

3.2、消费者(consumer)



import pika,sys

connection = pika.BlockingConnection(pika.ConnectionParameters
("localhost"))
channel = connection.channel()
#定义direct类型的exchange
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,properites,body):
"回调函数"
print(" [x] %r:%r" % (method.routing_key, body))
#消费消息
channel.basic_consume(callback,queue=queue_name,no_ack=True)

channel.start_consuming()

　　注意：绑定队列时，需要循环遍历绑定。

三、执行实验

3.1、符合条件的执行结果
　　①服务端不输入参数执行->客户端能正常收到消息

　　②服务端输入参数->客户端能正常收到消息
　　服务端：                                                                                            客户端：

　　结论：客户端设置只允许warning和info的级别的收到，其他的收不到

3.2、不符合符合条件的执行结果
　　服务端：                                                                                                客户端

　　结论：客户端不接受不满足info条件的消息。