一、模式介绍
在前面的例子中,消息直接发送到queue中。
现在介绍的模式,消息发送到exchange中,消费者把队列绑定到exchange上。
发布-订阅模式是把消息广播到每个消费者,每个消费者接收到的消息都是相同的。
一个生产者,多个消费者,每一个消费者都有自己的一个队列,生产者没有将消息直接发送到队列,而是发送到了交换机,每个队列绑定交换机,生产者发送的消息经过交换机,到达队列,实现一个消息被多个消费者获取的目的。需要注意的是,如果将消息发送到一个没有队列绑定的exchange上面,那么该消息将会丢失,这是因为在rabbitMQ中exchange不具备存储消息的能力,只有队列具备存储消息的能力。
队列模型:
二、代码示例
发布者:不再创建队列,发送消息到exchange(交换机)中。exchange_type为fanout。
1 #!/usr/bin/env python 2 import pika 3 4 connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘)) 5 channel = connection.channel() 6 7 channel.exchange_declare(exchange=‘logs‘, 8 exchange_type=‘fanout‘) 9 10 for i in range(20): 11 message = "info: Hello World! {}".format(i) 12 channel.basic_publish(exchange=‘logs‘, 13 routing_key=‘‘, 14 body=message) 15 print(" [x] Sent %r" % message) 16 connection.close()
订阅者
订阅者:每个消费者创建一个匿名队列,绑定到对应的exchange上,接收消息。一旦消费者的connection端口,对应的队列则关闭,不在接收消息。使用exclusive 来控制。
1 #!/usr/bin/env python 2 import pika 3 4 connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘)) 5 channel = connection.channel() 6 7 channel.exchange_declare(exchange=‘logs‘, 8 exchange_type=‘fanout‘) 9 10 result = channel.queue_declare(exclusive=True) 11 queue_name = result.method.queue 12 13 channel.queue_bind(exchange=‘logs‘, 14 queue=queue_name) 15 16 print(‘ [*] Waiting for logs. To exit press CTRL+C‘) 17 18 19 def callback(ch, method, properties, body): 20 print(" [x] %r" % body) 21 22 23 channel.basic_consume(callback, 24 queue=queue_name, 25 no_ack=True) 26 27 channel.start_consuming()
发布者输出:
[x] Sent ‘info: Hello World! 0‘ [x] Sent ‘info: Hello World! 1‘ [x] Sent ‘info: Hello World! 2‘ [x] Sent ‘info: Hello World! 3‘ [x] Sent ‘info: Hello World! 4‘ [x] Sent ‘info: Hello World! 5‘ [x] Sent ‘info: Hello World! 6‘ [x] Sent ‘info: Hello World! 7‘ [x] Sent ‘info: Hello World! 8‘ [x] Sent ‘info: Hello World! 9‘ [x] Sent ‘info: Hello World! 10‘ [x] Sent ‘info: Hello World! 11‘ [x] Sent ‘info: Hello World! 12‘ [x] Sent ‘info: Hello World! 13‘ [x] Sent ‘info: Hello World! 14‘ [x] Sent ‘info: Hello World! 15‘ [x] Sent ‘info: Hello World! 16‘ [x] Sent ‘info: Hello World! 17‘ [x] Sent ‘info: Hello World! 18‘ [x] Sent ‘info: Hello World! 19‘
订阅者输出:
[x] b‘info: Hello World! 0‘ [x] b‘info: Hello World! 1‘ [x] b‘info: Hello World! 2‘ [x] b‘info: Hello World! 3‘ [x] b‘info: Hello World! 4‘ [x] b‘info: Hello World! 5‘ [x] b‘info: Hello World! 6‘ [x] b‘info: Hello World! 7‘ [x] b‘info: Hello World! 8‘ [x] b‘info: Hello World! 9‘ [x] b‘info: Hello World! 10‘ [x] b‘info: Hello World! 11‘ [x] b‘info: Hello World! 12‘ [x] b‘info: Hello World! 13‘ [x] b‘info: Hello World! 14‘ [x] b‘info: Hello World! 15‘ [x] b‘info: Hello World! 16‘ [x] b‘info: Hello World! 17‘ [x] b‘info: Hello World! 18‘ [x] b‘info: Hello World! 19‘
可以看到两个订阅者收到的消息是一样的。
每个订阅者的状态变化不会影响另外的订阅者收到的消息。
三、队列信息
在web管理页面上,查看exchange中的logs,可以看到两个队列。
原文地址:https://www.cnblogs.com/StitchSun/p/9451779.html
时间: 2024-10-01 02:49:08