一、
消息的廣播需要exchange:exchange是一個轉發器,其實把消息發給RabbitMQ里的exchange
fanout: 所有bind到此exchange的queue都可以接收消息,廣播
direct: 通過routingKey和exchange決定的那個唯一的queue可以接收消息
topic:所有符合routingKey(此時可以是一個表達式)的routingKey所bind的queue可以接收消息
headers:通過headers來決定把消息發給哪些queue,用的比較少
原理:消息發布端分別發送INFO,WARNING,ERROR類型的消息,C1 C2 C3訂閱了不同類型的消息
消息發送端:
''' 發布者publisher ''' import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel = connection.channel() channel.exchange_declare(exchange='direct_logs', type='direct') # 1、改成type='direct' # 2、默認發送的消息級別為info,可以帶參數,warning error等 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, # 3、把上面的消息發到這個queue中 body=message) print("send :", message) connection.close()
消息訂閱者:
''' 訂閱者subscriber ''' import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel = connection.channel() channel.exchange_declare(exchange='direct_logs', type='direct') # 4、改exchange的類型 result = channel.queue_declare(exclusive=True) queue_name = result.method.queue # 5、啟動訂閱端的時候,severities存放訂閱端訂閱了哪些級別 # 然后用routing_key把這些級別綁定到queue上,這些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("Wait for logs...") # 6、使用method.routing_key可以得到消息的級別 def callback(ch, method, properties, body): print("received:", method.routing_key, body) channel.basic_consume(callback, queue=queue_name, no_ack=True) channel.start_consuming()
運行過程:
''' 啟動訂閱者1: python subscriber.py info 啟動訂閱者2:python subscriber.py info error 啟動發布者1:python publisher.py info hello 啟動發布者2:python publisher.py error servicesdown 訂閱者1收到消息:info b'hello' 訂閱者2收到消息:info b'hello' error b'servicesdown' '''