一、参考
1. Python操纵 RabbitMQ、Redis、Memcache、SQLAlchemy
2. RabbitMQ Tutorials | RabbitMQ
3. Python中使用RabbitMQ
4. python 操纵RabbitMq详解
5. rabbitmq篇---python的pika 库常用函数及参数说明
二、环境搭建
Docker 安装: win上使用Docker - 掘金 (juejin.cn)
MQ 安装:.Net RabbitMQ - 掘金 (juejin.cn)
控制台输入: pip install pika
三、快速使用
流程参考:RabbitMQ tutorial - "Hello world!" | RabbitMQ
3.1 创建生产者
- import pika
- # 无密码
- # connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
- # 有密码
- credentials = pika.PlainCredentials("guest","guest")
- connection = pika.BlockingConnection(pika.ConnectionParameters(
- host = '127.0.0.1',
- port = 5673, # 注意,默认为5672!5673是因为在docker初始化时设置的
- virtual_host = '/',
- credentials = credentials))
- channel = connection.channel()
- # 创建一个队列
- channel.queue_declare(queue='hello')
- # 发送数据
- channel.basic_publish(exchange='',
- routing_key='hello', # 消息队列名称
- body='Hello World!') # 发送的数据
- print(" [x] Sent 'Hello World!'")
- connection.close()
复制代码 运行该文件后:
3.1.1 durable 持久化
创建队列时,队列默认不进行持久化,可在创建时指定为持久化,即:
- # 创建一个队列 —— 支持持久化
- channel.queue_declare(queue='hello',durable=True)
复制代码 并且还需指定properties`,即:
- # 发送数据
- channel.basic_publish(exchange='',
- routing_key='hello', # 消息队列名称
- body='Hello World!', # 发送的数据
- properties=pika.BasicProperties(
- delivery_mode= 2 # 消息持久化
- ))
复制代码
3.2 创建消费者
- import pika
- credentials = pika.PlainCredentials('guest', 'guest')
- connection = pika.BlockingConnection(pika.ConnectionParameters(
- host = '127.0.0.1',
- port = 5673,
- virtual_host = '/',
- credentials = credentials))
- channel = connection.channel()
- # 申明消息队列,消息在这个队列传递,如果不存在,则创建队列
- channel.queue_declare(queue = 'hello')
- # 定义一个回调函数来处理消息队列中的消息,这里是打印出来
- def callback(ch, method, properties, body):
- print(f"消费者接收到了任务: {body.decode()}")
- # 告诉rabbitmq,用callback来接收消息
- channel.basic_consume(
- queue='hello',
- auto_ack=True,
- on_message_callback=callback
- )
- # 开始接收信息,并进入阻塞状态,队列里有信息才会调用callback进行处理
- channel.start_consuming()
复制代码 3.2.1 ack 消息验证
auto_ack 表现是否进行ack验证,默以为 False;
当为 False 时,消费者发生异常后,消息不会被消费;
当为 True 时,消费者发生异常后,消息仍会被消费。
3.2.2 闲置消费
默认派发次序按照队列次序执行,当队列某个使命处置惩罚时间过久时,就会造成资源浪费
- # 闲置消费
- channel.basic_qos(prefetch_count=1)
复制代码
3.3 结果演示
运行该文件后:
四、发布与订阅
上面方法用于使命的发布与处置惩罚,一个使命只会对应一个处置惩罚者,
对应关系如下:
下图按照特定次序分发使命(轮询、闲置消费等等)
而 发布/订阅 模式,要借助交换机(Exchange),一个使命会交给多个处置惩罚者,常用于通知
关系如下图所示:
4.1 Fanout 模式
流程参考:RabbitMQ tutorial - Publish/Subscribe | RabbitMQ
这种模式下,通报到 exchange 的消息将会转发到全部与其绑定的 queue 上。
- 不须要指定 routing_key ,即使指定了也是无效。
- 须要提前将 exchange 和 queue 绑定,一个 exchange 可以绑定多个 queue,一个queue可以绑定多个exchange。
- 须要先启动 订阅者,此模式下的队列是 consumer 随机天生的,发布者 仅仅发布消息到 exchange ,由 exchange 转发消息至 queue。
4.1.1 生产者-发布者
- import pika
- # 无密码
- # connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
- # 有密码
- credentials = pika.PlainCredentials("guest","guest")
- connection = pika.BlockingConnection(pika.ConnectionParameters(
- host = '127.0.0.1',
- port = 5673,
- credentials = credentials))
- channel = connection.channel()
- # 设置订阅模式
- channel.exchange_declare(exchange='s',durable=True,exchange_type='fanout')
- # 创建一个队列
- # channel.queue_declare(queue='hello',durable=True)
- # 发送数据
- channel.basic_publish(exchange='s',
- routing_key='', # 消息队列名称
- body='Hello World!', # 发送的数据
- )
- print(" [x] Sent 'Hello World!'")
- connection.close()
复制代码 4.1.2 消费者-订阅者
- import pikacredentials = pika.PlainCredentials('guest', 'guest')connection = pika.BlockingConnection(pika.ConnectionParameters( host = '127.0.0.1', port = 5673, virtual_host = '/', credentials = credentials))channel = connection.channel()# 设置订阅模式channel.exchange_declare(exchange='s',durable=True,exchange_type='fanout')# 随机天生队列result = channel.queue_declare(queue='',exclusive=True)queue_name = result.method.queue# 绑定 exchange 和 queue 绑定channel.queue_bind(exchange='s',queue = queue_name)# 申明消息队列,消息在这个队列通报,如果不存在,则创建队列# channel.queue_declare(queue = 'hello')# 定义一个回调函数来处置惩罚消息队列中的消息,这里是打印出来def callback(ch, method, properties, body): ch.basic_ack(delivery_tag = method.delivery_tag) print(f"消费者吸收到了使命: {body.decode()}")# 闲置消费
- channel.basic_qos(prefetch_count=1)
- # 告诉rabbitmq,用callback来吸收消息channel.basic_consume( # queue='m1', queue= queue_name, auto_ack=False, on_message_callback=callback )# 开始吸收信息,并进入阻塞状态,队列里有信息才会调用callback进行处置惩罚channel.start_consuming()
复制代码 相比于之前做出的修改`:
4.1.3 结果演示
创建两个消费者和一个生产者,当生产者发生消息后,两个消费者同时收到
4.2 Routing(Direct) 模式
这种工作模式的原理是 消息发送至 exchange,exchange 根据 路由键(routing_key) 转发到相对应的 queue 上。
- 可以使用默认 exchange =' ' ,也可以自定义 exchange
- 这种模式下不须要将 exchange 和 任何进行绑定,当然绑定也是可以的。可以将 exchange 和 queue ,routing_key 和 queue 进行绑定
- 通报或接受消息时 须要 指定 routing_key
- 须要先启动 订阅者,此模式下的队列是 consumer 随机天生的,发布者 仅仅发布消息到 exchange ,由 exchange 转发消息至 queue。
4.2.1 生产者-发布者
- import pika
- # 无密码
- # connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
- # 有密码
- credentials = pika.PlainCredentials("guest","guest")
- connection = pika.BlockingConnection(pika.ConnectionParameters(
- host = '127.0.0.1',
- port = 5673,
- credentials = credentials))
- channel = connection.channel()
- # 设置订阅模式
- exchange_name = 's1'
- channel.exchange_declare(exchange= exchange_name,durable=True,exchange_type='direct')
- # 创建一个队列
- # channel.queue_declare(queue='hello',durable=True)
- # 发送数据
- data = 'hello world'
- channel.basic_publish(
- exchange = exchange_name,
- routing_key='b', # 消息队列名称
- body= data, # 发送的数据
- )
- print(f" [x] Sent '{data}'")
- connection.close()
复制代码 4.2.2 消费者-订阅者
- import pikacredentials = pika.PlainCredentials('guest', 'guest')connection = pika.BlockingConnection(pika.ConnectionParameters( host = '127.0.0.1', port = 5673, virtual_host = '/', credentials = credentials))channel = connection.channel()# 路由关键词routing_key_1 = 'a'# 设置订阅模式exchange_name = 's1'channel.exchange_declare( exchange= exchange_name, durable = True, exchange_type = 'direct' )# 随机天生队列result = channel.queue_declare(queue='',exclusive=True)queue_name = result.method.queue# 绑定 exchange 和 queue 绑定channel.queue_bind( exchange = exchange_name, queue = queue_name, routing_key = routing_key_1 )# 申明消息队列,消息在这个队列通报,如果不存在,则创建队列# channel.queue_declare(queue = 'hello')# 定义一个回调函数来处置惩罚消息队列中的消息,这里是打印出来def callback(ch, method, properties, body): ch.basic_ack(delivery_tag = method.delivery_tag) print(f"消费者吸收到了使命: {body.decode()}")# 闲置消费
- channel.basic_qos(prefetch_count=1)
- # 告诉rabbitmq,用callback来吸收消息channel.basic_consume( # queue='m1', queue= queue_name, auto_ack=False, on_message_callback=callback )# 开始吸收信息,并进入阻塞状态,队列里有信息才会调用callback进行处置惩罚channel.start_consuming()
复制代码 4.2.3 结果演示
创建两个消费者:
设置消费者一的 routingkey为 a 和 b,消费者二的 routingkey为 a
测试生产者发送不同的 routing_key,消费者的反应环境
4.3 Topics 模式
流程参考: RabbitMQ tutorial - Topics | RabbitMQ
这种模式和第二种模式差不多,exchange 也是通过 路由键 routing_key 来转发消息到指定的 queue 。
不同点是 routing_key 使用正则表达式支持模糊匹配,但匹配规则又与通例的正则表达式不同,
比如“#”是匹配全部,“*”是匹配一个词,词与词之间使用 “.” 分割
4.3.1 生产者-发布者
- import pika
- # 无密码
- # connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
- # 有密码
- credentials = pika.PlainCredentials("guest","guest")
- connection = pika.BlockingConnection(pika.ConnectionParameters(
- host = '127.0.0.1',
- port = 5673,
- credentials = credentials))
- channel = connection.channel()
- # 设置订阅模式
- exchange_name = 's2'
- channel.exchange_declare(exchange= exchange_name,durable=True,exchange_type='topic')
- # 创建一个队列
- # channel.queue_declare(queue='hello',durable=True)
- # 发送数据
- routing_key = 'a.1.2'
- data = 'hello world'
- channel.basic_publish(
- exchange = exchange_name,
- routing_key = routing_key, # 消息队列名称
- body= data, # 发送的数据
- )
- print(f" [x] Sent '{data}'")
- connection.close()
复制代码 4.3.2 消费者-订阅者
- import pikacredentials = pika.PlainCredentials('guest', 'guest')connection = pika.BlockingConnection(pika.ConnectionParameters( host = '127.0.0.1', port = 5673, virtual_host = '/', credentials = credentials))channel = connection.channel()# 路由关键词 : '#' 是匹配全部,'*' 是匹配一个词routing_key_1 = 'a.#'routing_key_2 = 'b.*'# 设置订阅模式exchange_name = 's2'channel.exchange_declare( exchange= exchange_name, durable = True, exchange_type = 'topic' )# 随机天生队列result = channel.queue_declare(queue='',exclusive=True)queue_name = result.method.queue# 绑定 exchange 和 queue 绑定channel.queue_bind( exchange = exchange_name, queue = queue_name, routing_key = routing_key_1 )channel.queue_bind( exchange = exchange_name, queue = queue_name, routing_key = routing_key_2 )# 申明消息队列,消息在这个队列通报,如果不存在,则创建队列# channel.queue_declare(queue = 'hello')# 定义一个回调函数来处置惩罚消息队列中的消息,这里是打印出来def callback(ch, method, properties, body): ch.basic_ack(delivery_tag = method.delivery_tag) print(f"消费者吸收到了使命: {body.decode()}")# 闲置消费
- channel.basic_qos(prefetch_count=1)
- # 告诉rabbitmq,用callback来吸收消息channel.basic_consume( # queue='m1', queue= queue_name, auto_ack=False, on_message_callback=callback )# 开始吸收信息,并进入阻塞状态,队列里有信息才会调用callback进行处置惩罚channel.start_consuming()
复制代码 4.3.3 结果演示
创建两个消费者:
设置消费者一的 routing_key 为 a.# 和 b.*, 指a后面可以以恣意词结尾,b后面最多只能附加一个词
消费者二的 routing_key 为 a.#,指a后面可以以恣意词结尾
测试生产者发送不同的 routing_key,消费者的反应环境
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。 |