python使用rabbitMQ介绍三（发布订阅模式）

Posted 2020-12-29 Stitch

一、模式介绍

在前面的例子中，消息直接发送到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 来控制。

 

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host=\'localhost\'))
channel = connection.channel()

channel.exchange_declare(exchange=\'logs\',
                         exchange_type=\'fanout\')

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

channel.queue_bind(exchange=\'logs\',
                   queue=queue_name)

print(\' [*] Waiting for logs. To exit press CTRL+C\')


def callback(ch, method, properties, body):
    print(" [x] %r" % body)


channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

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，可以看到两个队列。