Python-RabbitMQ消息队列的发布与订阅
Posted Sch01aR#
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Python-RabbitMQ消息队列的发布与订阅相关的知识,希望对你有一定的参考价值。
RabbitMQ消息队列的发布与订阅类似于广播,一端发送消息,多个客户端可以同时接收到消息
fanout:所有绑定到exchange的queue都可以接收消息
消息发布端
# -*- coding:utf-8 -*- __author__ = "MuT6 Sch01aR" import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host=\'127.0.0.1\')) channel = connection.channel() channel.exchange_declare(exchange=\'fanout_logs\', exchange_type=\'fanout\') message = \'Hello World!\' channel.basic_publish(exchange=\'fanout_logs\', routing_key=\'\', body=message) print(\'发送消息:\', message) connection.close()
消息订阅端
# -*- coding:utf-8 -*- __author__ = "MuT6 Sch01aR" import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host=\'127.0.0.1\')) channel = connection.channel() channel.exchange_declare(exchange=\'fanout_logs\', exchange_type=\'fanout\') result = channel.queue_declare(exclusive=True) # 不指定queue名字,rabbitmq会随机分配一个名字 # exclusive=True会在使用此queue的消息订阅端断开后,自动将queue删除 queue_name = result.method.queue print(\'当前queue名称:\', queue_name) channel.queue_bind(exchange=\'fanout_logs\', queue=queue_name) def callback(ch, method, properties, body): print(ch, method, properties) print(\'收到数据:\', body) channel.basic_consume(callback, queue=queue_name, no_ack=True, ) print(\'开始等待消息\') channel.start_consuming()
消息发布端需要在消息订阅端运行之后运行,不然消息订阅端收不到消息
开启3个消息订阅端和一个消息发布端
消息发布端发布消息
3个消息订阅端会同时接收到消息
direct:指定的queue才能收到消息
把queue绑定关键字,消息发布端根据关键字将消息发送到exchange,exchange根据关键字将消息发送到指定队列
消息发布端
# -*- coding:utf-8 -*- __author__ = "MuT6 Sch01aR" import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters(host=\'127.0.0.1\')) channel = connection.channel() channel.exchange_declare(exchange=\'direct_logs\', exchange_type=\'direct\') severity = sys.argv[1] if len(sys.argv) > 1 else \'info\' # 关键字 message = \'Hello World!\' channel.basic_publish(exchange=\'direct_logs\', routing_key=severity, body=message) print(\'级别 [%s] 发送数据 [%s]\' % (severity, message)) connection.close()
消息订阅端
# -*- coding:utf-8 -*- __author__ = "MuT6 Sch01aR" import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters(host=\'127.0.0.1\')) channel = connection.channel() channel.exchange_declare(exchange=\'direct_logs\', exchange_type=\'direct\') result = channel.queue_declare(exclusive=True) queue_name = result.method.queue severities = sys.argv[1:] # 关键字 for severity in severities: channel.queue_bind(exchange=\'direct_logs\', queue=queue_name, routing_key=severity) def callback(ch, method, properties, body): print(ch, method, properties) print(\'收到来自%s的消息:%s\' % (method.routing_key, body)) channel.basic_consume(callback, queue=queue_name, no_ack=True) print(\'开始等待消息\') channel.start_consuming()
开启3个消息订阅端,分别引用关键字info、warning,warning、error,error、info,开启一个消息发布端,引用关键字wanrning
只有引用关键字warning的消息订阅端才收到消息
没有引用关键字warning的就收不到消息
topic:符合条件的queue才能收到消息
消息订阅端指定条件,符合条件的queue才能被消息订阅端收到消息
#会匹配任意个任意字符串,*会匹配一个任意字符串
单个#会匹配全部字符串,单个*无法匹配
消息发布端
# -*- coding:utf-8 -*- __author__ = "MuT6 Sch01aR" import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters(host=\'127.0.0.1\')) channel = connection.channel() channel.exchange_declare(exchange=\'topic_logs\', exchange_type=\'topic\') severity = sys.argv[1] if len(sys.argv) > 1 else \'info\' message = \'Hello World!\' channel.basic_publish(exchange=\'topic_logs\', routing_key=severity, body=message) print(\'给 [%s] 发送消息 [%s]\' % (severity, message)) connection.close()
消息订阅端
# -*- coding:utf-8 -*- __author__ = "MuT6 Sch01aR" import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters(host=\'127.0.0.1\')) channel = connection.channel() channel.exchange_declare(exchange=\'topic_logs\', exchange_type=\'topic\') result = channel.queue_declare(exclusive=True) queue_name = result.method.queue keys = sys.argv[1:] for key in keys: channel.queue_bind(exchange=\'topic_logs\', queue=queue_name, routing_key=key) def callback(ch, method, properties, body): print(ch, method, properties) print(\'来自 [%s] 的信息 [%s]\' % (method.routing_key, body)) channel.basic_consume(callback, queue=queue_name, no_ack=True) print(\'开始接收消息\') channel.start_consuming()
开启3个消息订阅端和一个消息发布端
3个消息订阅端,一个*.exe匹配以.exe结尾的,一个#匹配全部字符串,一个python.exe只匹配字符串为python.exe的
消息发布端引用关键字qq.exe
很显然,匹配以.exe结尾和匹配全部字符串的消息订阅端将会收到消息,只匹配字符串为python.exe的不会收到消息
以上是关于Python-RabbitMQ消息队列的发布与订阅的主要内容,如果未能解决你的问题,请参考以下文章