python如何使用Rabbitmq

目录

一、Rabbitmq介绍

二、Rabbitmq的使用场景

1、异步处理

2、服务解耦

3、流量削峰

4、日志收集

5、发布订阅

6、任务调度

三、python如何使用Rabbitmq

1、安装依赖

2、基础使用

3、消息确认

4、消息持久化

5、公平调度

6、发布订阅

7、关键字发布

---

一、Rabbitmq介绍

RabbitMQ是一个开源的消息中间件，基于AMQP（Advanced Message Queue Protocol，高级消息队列协议）协议实现。RabbitMQ被广泛应用于各种应用场景，如异步任务处理、日志传输、实时消息推送等。在微服务架构中，RabbitMQ是一个常见的消息中间件选择，它可以帮助服务之间实现解耦和异步通信，提高系统的可扩展性和稳定性。RabbitMQ提供了一个简单的用户页面，用户可以监控和管理消息、队列、交换器、绑定等资源。通过管理界面，用户可以直观地了解系统的运行状态，并进行相应的配置和管理操作。

二、Rabbitmq的使用场景

1、异步处理

• 在Web应用中，当用户提交表单时，可以将表单处理任务发送给RabbitMQ，由后台服务异步处理，从而提高用户界面的响应速度。
• 在电商系统中，用户下单后，订单处理、库存更新、支付通知等任务可以异步执行，避免阻塞主线程。

2、服务解耦

• 在微服务架构中，不同服务之间通过RabbitMQ进行通信，可以降低服务之间的耦合度，提高系统的可扩展性和可维护性。
• 当某个服务需要升级或维护时，可以通过RabbitMQ实现服务的平滑过渡，而不会影响其他服务的正常运行。

3、流量削峰

• 在高并发场景中，RabbitMQ可以作为一个缓冲层，接收并存储大量的请求，然后按照设定的速率将请求转发给后端服务，从而避免后端服务因过载而崩溃。
• 通过RabbitMQ的限流和队列机制，可以有效地控制请求的速率和数量，保护后端服务的稳定性。

4、日志收集

• RabbitMQ可以用于收集分散在各个服务器上的日志信息，将它们集中到一个或多个日志处理服务中，进行统一的分析和处理。
• 通过RabbitMQ，可以实现日志的实时收集、分析和报警，提高系统的运维效率和故障排查能力。

5、发布订阅

• 在需要向多个客户端推送消息的场景中，如实时通知、消息推送等，可以使用RabbitMQ的Fanout交换器将消息广播给所有绑定的队列。
• 通过RabbitMQ的消息广播机制，可以实现实时、可靠的消息推送服务，提高用户体验。

6、任务调度

• RabbitMQ可以与其他任务调度框架（如Quartz）结合使用，实现定时任务、延迟任务等复杂任务调度需求。
• 通过RabbitMQ的任务调度功能，可以灵活地控制任务的执行时间和频率，提高系统的自动化程度和运行效率。

三、python如何使用Rabbitmq

Rabbitmq网址：https://www.rabbitmq.com/tutorials

1、安装依赖

安装第三方库pika：pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pika

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pika

2、基础使用

生产者模型代码如下

import pika# 生产者模型connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接本地的rabbitmq
channel = connection.channel()  # 连接通道
channel.queue_declare(queue='list')  #  创建一个名叫list的队列
channel.basic_publish(exchange='',routing_key='list',  # 向那个队列发布信息body='lol',# 发布的信息)
connection.close()  # 关闭连接

消费者代码如下

import pika# 消费者模型connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接本地的rabbitmq
channel = connection.channel()  # 连接通道
channel.queue_declare(queue='list')  #  为确保队列存在，创建一个list队列def callback(ch, method, properties, body):print(f" [x] 消费者接收到了任务 {body}")channel.basic_consume(queue='list',on_message_callback=callback,auto_ack=True)  # 接收信息：queue表示监听的队列，on_message_callback表示接收到信息执行的函数，auto_ack表示默认执行回复
channel.start_consuming()   # 开启永无止境的循环监听该队列

3、消息确认

在队列中，执行任务可能需要几秒钟，您可能想知道如果 使用者启动一个长任务，并在完成之前终止。 使用我们当前的代码，一旦 RabbitMQ 将消息传递给消费者，它就会 立即将其标记为删除。在这种情况下，如果您终止 worker，则它刚刚处理的消息丢失了。调度的消息 对于这个尚未处理的特定 worker 来说，也会丢失。为了确保消息永远不会丢失，RabbitMQ 支持消息确认。ack由 consumer 告诉 RabbitMQ 已收到特定消息， 处理，并且 RabbitMQ 可以自由删除它。

这是由消费者来进行改变的，代码如下：

需要将auto_ack改为False，然后在回调函数里加入ch.basic_ack(delivery_tag=method.delivery_tag)

import pika# 消费者模型connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接本地的rabbitmq
channel = connection.channel()  # 连接通道
channel.queue_declare(queue='list')  #  创建一个队列def callback(ch, method, properties, body):print(f" [x] 消费者接收到了任务 {body}")ch.basic_ack(delivery_tag=method.delivery_tag) # 给队列回复进行确认channel.basic_consume(queue='list2',on_message_callback=callback,auto_ack=False)  # auto_ack改为False
channel.start_consuming()  # 开启永无止境的循环监听该队列

4、消息持久化

如果 RabbitMQ 服务器 停止。

当 RabbitMQ 退出或崩溃时，它会忘记队列和消息 除非你告诉它不要这样做。需要做两件事来确保 消息不会丢失：我们需要将队列和消息都标记为 耐用。然和把队列和消息保存在磁盘里。

这是有生产者来进行改变的，代码如下：

durable=True在创建队列时声明持久化，delivery_mode=pika.DeliveryMode.Persistent让信息做持久化

import pika# 生产者模型connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接本地的rabbitmq
channel = connection.channel()  # 连接通道
channel.queue_declare(queue='list2',durable=True)  #  创建一个队列,durable=True表示队列支持持久化
channel.basic_publish(exchange='',routing_key='list2',  # 向那个队列发布信息body='lol',# 发布的信息properties=pika.BasicProperties(delivery_mode=pika.DeliveryMode.Persistent  # 让信息做持久化))
connection.close()  # 关闭连接

5、公平调度

您可能已经注意到，调度仍然没有完全工作 正如我们想要的那样。例如，在有两个 worker 的情况下，当所有 奇数消息很重，偶数消息很轻，一个 worker 将是 一直很忙，另一个几乎不做任何工作。井 RabbitMQ 对此一无所知，仍会 dispatch 消息均匀。

发生这种情况是因为 RabbitMQ 只是在消息 进入队列。它不看未确认的数量 消息。它只是盲目地调度每 n 条消息 到第 n 个消费者。

为了解决这个问题，我们可以使用带有 setup 的 channel 方法。它使用协议方法告诉 RabbitMQ 不要一次向 worker 提供多条消息。或者，换句话说，不要调度 向 worker 发送新消息，直到它处理并确认 上一个。相反，它会将其分派给下一个仍然不忙的 worker。

这是由消费者来进行改变的，代码如下：

channel.basic_qos(prefetch_count=1)进行闲置派发

import pika# 消费者模型connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接本地的rabbitmq
channel = connection.channel()  # 连接通道
channel.queue_declare(queue='list')  #  为确保队列存在，创建一个list队列def callback(ch, method, properties, body):print(f" [x] 消费者接收到了任务 {body}")channel.basic_qos(prefetch_count=1)  # 闲置派发
channel.basic_consume(queue='list',on_message_callback=callback,auto_ack=True)  # 接收信息：queue表示监听的队列，on_message_callback表示接收到信息执行的函数，auto_ack表示默认执行回复
channel.start_consuming()   # 开启永无止境的循环监听该队列

6、发布订阅

发布订阅模式是将信息发布给所有的订阅者，其特点就是有交换机。

channel.exchange_declare(exchange='m1',exchange_type='fanout')声明一个交换机，类型为fanout，是将信息发给所有的订阅者

发布者代码如下：

import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接rabbitmq
channel = connection.channel()  # 连接通道channel.exchange_declare(exchange='m1',exchange_type='fanout')  # fanout：将信息发给所有的队列
channel.queue_declare(queue='')  #  创建一个队列,durable=True表示队列支持持久化
channel.basic_publish(exchange='m1',routing_key='',  # 队列名称body='lol',)# 发布的数据
connection.close()  # 关闭连接

订阅者代码如下：

import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接rabbitmq
channel = connection.channel()channel.exchange_declare(exchange='m1',exchange_type='fanout',)  # fanout：将信息发给所有的队列
result=channel.queue_declare(exclusive=True,queue='')  #  创建一个随机队列
queue_name=result.method.queue  # 拿到队列名字
print(queue_name)
channel.queue_bind(exchange='m1',queue=queue_name)  # 对exchange和queue进行绑定def callback(ch, method, properties, body):print(f" [x] 消费者接收到了任务 {body}")channel.basic_consume(queue=queue_name,on_message_callback=callback,auto_ack=True)
channel.start_consuming()  # 开启永无止境的循环监听该队列

7、关键字发布

关键字发布就是在发布订阅模式基础上，将不同信息发布给不同的订阅者。

channel.exchange_declare(exchange='m2',exchange_type='direct') ，声明一个交换机，交换机的类型为direct，将信息发布给指定的订阅者。

发布者代码如下：

import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接rabbitmq
channel = connection.channel()  # 连接通道channel.exchange_declare(exchange='m2',exchange_type='direct')  # direct：将信息发布给指定的订阅者
channel.queue_declare(queue='')  #  创建一个队列,durable=True表示队列支持持久化
channel.basic_publish(exchange='m2',routing_key='hhq',  # 队列名称body='cf',)# 发布的数据
connection.close()  # 关闭连接

订阅者代码如下：

import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接rabbitmq
channel = connection.channel()channel.exchange_declare(exchange='m2',exchange_type='direct')  # exchange:交易所的名称，fanout：将信息发给所有的队列
result=channel.queue_declare(exclusive=True,queue='')  #  创建一个随机队列
queue_name=result.method.queue  # 拿到队列名字
print(queue_name)
channel.queue_bind(exchange='m2',queue=queue_name,routing_key='hhq')  # 对exchange和queue进行绑定def callback(ch, method, properties, body):print(f" [x] 消费者接收到了任务 {body}")channel.basic_consume(queue=queue_name,on_message_callback=callback,auto_ack=True)
channel.start_consuming()  # 开启永无止境的循环监听该队列