在上篇教程中,我们搭建了一个工作队列,每个任务只分发给一个工作者(worker)。在本篇教程中,我们要做的跟之前完全不一样 —— 分发一个消息给多个消费者(consumers)。这种模式被称为“发布/订阅”。 为了描述这种模式,我们将会构建一个简单的日志系统。它包括两个程序——第一个程序负责发送日志消息,第二个程序负责获取消息并输出内容。 在我们的这个日志系统中,所有正在运行的接收方程序都会接受消息。我们用其中一个接收者(receiver)把日志写入硬盘中,另外一个接受者(receiver)把日志输出到屏幕上。最终,日志消息被广播给所有的接受(receivers)。 1、 交换机(Exchanges) 前面的教程中,我们发送消息到队列并从中取出消息。现在是时候介绍RabbitMQ中完整的消息模型了。让我们简单的概括一下之前的教程: 发布者(producer)是发布消息的应用程序。 队列(queue)用于消息存储的缓冲。 消费者(consumer)是接收消息的应用程序。 RabbitMQ消息模型的核心理念是:发布者(producer)不会直接发送任何消息给队列。事实上,发布(producer)甚至不知道消息是否已经被投递到队列。 发布者(producer)只需要把消息发送给一个交换机(exchange)。交换机非常简单,它一边从发布者方接收消息,一边把消息推送到队列。交换机必须知道如何处理它接收到的消息,是应该推送到指定的队列还是是多个队列,或者是直接忽略消息。这些规则是通过交换机类型(exchange type)来定义的。 有几个可供选择的交换机类型: 直连交换机(direct):指定的queue才能收到消息。把queue绑定关键字,消息发布端根据关键字将消息发送到exchange,exchange根据关键字将消息发送到指定队列; 主题交换机(topic): 符合条件的queue才能收到消息. 头交换机 (headers): 扇型交换机(fanout): 它把消息发送给它所知道的所有队列
一、新创建一个Exchanges
二、消息发布端
#!/usr/bin/env python import time import pika import datetime import json def get_message(): # 产生消息入口处 for i in range(100): # 生成10条消息 message = json.dumps({'id': "psub_0000%s" % i, "amount": 100 * i, "name": "melon", "createtime": str(datetime.datetime.now())}) producer(message) time.sleep(1) def producer(message): connection = pika.BlockingConnection( pika.ConnectionParameters(virtual_host='/melon_demo', host='82.156.19.94', port=5672, credentials=pika.PlainCredentials('guest', 'guest'))) channel = connection.channel() channel.exchange_declare(exchange='fanout_logs', exchange_type='fanout', internal=False, durable=True) channel.basic_publish(exchange='fanout_logs', routing_key='', body=message) print(" [x] Sent %r" % message) connection.close() if __name__ == "__main__": get_message() # 程序执行入口
三、消息接收端
#!/usr/bin/env python import pika connection = pika.BlockingConnection(pika.ConnectionParameters(virtual_host='/melon_demo', host='82.156.19.94', port=5672, credentials=pika.PlainCredentials('guest', 'guest'))) channel = connection.channel() channel.exchange_declare(exchange='fanout_logs', exchange_type='fanout', internal=False, durable=True) result = channel.queue_declare(queue='', exclusive=True) queue_name = result.method.queue print('========queue_name=======',queue_name) channel.queue_bind(exchange='fanout_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(queue=queue_name, on_message_callback=callback, auto_ack=True) channel.start_consuming()
四、先运行接收端程序,再运行发布端
启动三个接收程序;生成三个queue,一个消息产生程序:三个接收都能同步接收到相同的消息![](/i/l/?n=23&i=blog/1113099/202302/1113099-20230205161158928-1441136741.png)
标签:订阅,pika,RBMQ,exchange,queue,案例,交换机,消息,channel From: https://www.cnblogs.com/1314520xh/p/17093514.html