目标:
代码实现(direct_product.py)
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
1 # __author__ = 'STEVEN' 2 import pika,sys 3 #开启socket 4 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) 5 #开启一个通道 6 channel = connection.channel() 7 #这里不用再创建队列 8 channel.exchange_declare(exchange='d_logs',exchange_type='direct') 9 #通过命令行运行本代码,指定要发送的消息级别,如果不指定则为info 10 severity = sys.argv[1] if len(sys.argv)>1 else 'info' 11 #通过命令行输入消息级别后的参数 12 mes = ''.join(sys.argv[2:]) or 'hello world' 13 #添加了消息级别,即将消息发布到指定了级别的消息队列 14 channel.basic_publish(exchange='d_logs',routing_key=severity,body=mes) 15 print('[x] send the mes%s to queue'%mes) 16 connection.close()
代码实现(direct_consumer.py)
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
1 # __author__ = 'STEVEN' 2 import pika,sys 3 #建立socket 4 connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) 5 #开启通道 6 channel = connection.channel() 7 #通道先声明exchange类型 8 channel.exchange_declare(exchange='d_logs',exchange_type='direct') 9 #声明queue 10 result = channel.queue_declare(exclusive=True) 11 #获取queue_name 12 queue_name = result.method.queue 13 #从sys中获取消息级别等信息 14 severitis = sys.argv[1:] 15 if not severitis: 16 sys.stderr.write('usage:%s [info] [error] [warning]'%sys.argv[0]) 17 sys.exit(1) 18 #遍历绑定各个exchange和queue以及消息级别 19 for severity in severitis: 20 #绑定queue 21 channel.queue_bind(exchange='d_logs',routing_key=severity,queue=queue_name) 22 #回调函数 23 def callback(ch,method,properties,body): 24 print('[x] receive mess :%s'%body.decode()) 25 #指定消费相关参数 26 channel.basic_consume(callback,queue=queue_name,no_ack=True) 27 print('[*] is waiting for the message') 28 #开启消费 29 channel.start_consuming()
运行注意:
先开启消费端,用命令行指定接受级别