Python Web开发：消息队列与异步任务

🕗 发布于 2024-09-20 19:51 python 前端 开发语言 性能优化 fastapi

Python Web开发：消息队列与异步任务

💡 消息队列基础
- 1.1 消息队列的概念及应用场景
- 1.2 常见消息队列系统（如 RabbitMQ、Kafka）
🚀 Celery 异步任务队列
- 2.1 使用 Celery 实现任务的异步处理与调度
- 2.2 Celery 与 Redis、RabbitMQ 的集成
🔐 消息的可靠性
- 3.1 确保消息的幂等性处理，避免重复消费
- 3.2 死信队列的应用
📊 任务监控与重试机制
- 4.1 Celery 的任务状态监控与失败重试

1. 💡 消息队列基础

1.1 消息队列的概念及应用场景

消息队列（Message Queue，简称MQ）是一种用于在分布式系统中异步通信的机制。它的核心作用是通过提供一个缓冲区，帮助解耦应用程序的发送者与接收者。消息队列系统能够保证消息从发送者传递到接收者，即便接收方短暂不可用，消息也不会丢失，确保系统的稳定性和可扩展性。

应用场景：

解耦异步任务：在一些高并发场景中，像订单处理、短信通知、邮件发送等场景，消息队列能够帮助处理大量的请求并提高系统的响应速度。通过异步消息的处理，减少实时响应的压力。
削峰填谷：在业务系统中，会遇到某些时间段并发请求的流量较大，这时可以通过消息队列将流量平缓化，以确保系统的平稳运行。例如，在双十一等促销活动期间，用户的订单请求会大量涌入，系统能够通过消息队列来平稳处理订单请求，避免系统崩溃。
数据的异步同步：例如在分布式系统中，某些业务数据需要在多个微服务中保持一致性。此时，利用消息队列可以保证异步的数据同步，提高服务的可用性。
日志与监控数据的异步采集：在需要记录大量日志信息或监控数据时，消息队列可以帮助减少数据的写入压力，保证系统不会因为大量日志写入而产生瓶颈。

在这些场景下，消息队列可以很好地处理并发、解耦和系统延迟问题，是高并发分布式系统中必不可少的一部分。

1.2 常见消息队列系统（如 RabbitMQ、Kafka）

消息队列系统有很多种，每种系统都有其独特的设计与应用场景。以下是两种最常见的消息队列系统：

RabbitMQ：RabbitMQ 是一个强大的消息代理，基于 AMQP 协议。它支持复杂的路由机制，能够在不同的队列之间灵活的分发消息。RabbitMQ 适合于业务逻辑较为复杂的场景，尤其是需要多种不同的消费者处理消息时。它具有良好的可靠性和消息持久化机制，能确保消息不丢失。

代码示例：如何在 Python 中使用 pika 库来与 RabbitMQ 交互：
```
import pika

# 连接到 RabbitMQ 服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
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()
```
Kafka：Kafka 是一个分布式的消息系统，主要应用于高吞吐量、实时数据处理的场景。Kafka 具有分布式、高可靠性、高吞吐量的特性，通常用于日志收集、大数据分析等场景。Kafka 中的消息是按主题（Topic）进行分类存储，并且它的消费者可以根据需要进行消息的顺序消费和重复读取，极大地增强了灵活性。

代码示例：如何在 Python 中使用 confluent-kafka 库与 Kafka 交互：
```
from confluent_kafka import Producer

# 创建一个 Kafka 生产者
conf = {'bootstrap.servers': "localhost:9092"}
producer = Producer(**conf)

# 发送消息到 Kafka 的 topic
producer.produce('my_topic', key='key', value='Hello Kafka!')
producer.flush()
```

RabbitMQ 与 Kafka 在处理消息的可靠性、持久性以及性能上各有千秋，企业应根据自身的业务场景选择合适的消息队列。

2. 🚀 Celery 异步任务队列

2.1 使用 Celery 实现任务的异步处理与调度

Celery 是一个用于实现异步任务处理的分布式任务队列。它能帮助开发者轻松实现任务的异步执行，特别是在处理需要耗费大量时间的任务时（如视频处理、邮件发送等），Celery 的异步机制能有效减少用户的等待时间，提升系统的响应速度。

基本原理：
Celery 任务是通过消息队列进行分发的。当一个任务被提交时，Celery 会将该任务放入消息队列中，工作进程（Worker）从队列中取出任务并执行。Celery 支持多种消息代理，包括 RabbitMQ、Redis 和 Kafka 等。

from celery import Celery

# 创建 Celery 应用
app = Celery('tasks', broker='redis://localhost:6379/0')

# 定义一个异步任务
@app.task
def add(x, y):
    return x + y

在上述代码中，add 函数是一个异步任务。我们通过 Celery 的 @app.task 装饰器将它转换为异步任务。在程序运行时，任务会被添加到 Redis 消息队列中，等待 Celery 工作进程来处理。

2.2 Celery 与 Redis、RabbitMQ 的集成

Celery 支持多种消息代理，下面详细介绍 Celery 与 Redis 和 RabbitMQ 的集成。

1. 集成 Redis

Redis 是 Celery 最常见的消息代理之一，配置简单，且适用于较轻量的应用场景。可以通过配置 Celery 使用 Redis 作为消息代理。

# 配置 Celery 使用 Redis 作为 Broker
app = Celery('tasks', broker='redis://localhost:6379/0')

任务示例：

@app.task
def send_email(to_email):
    # 模拟发送邮件的过程
    print(f"Sending email to {to_email}")

2. 集成 RabbitMQ

与 Redis 相比，RabbitMQ 更加适合复杂的消息队列需求，特别是在消息需要多种消费者处理的情况下。Celery 也支持与 RabbitMQ 集成。

# 配置 Celery 使用 RabbitMQ 作为 Broker
app = Celery('tasks', broker='pyamqp://guest@localhost//')

任务示例：

@app.task
def process_order(order_id):
    # 模拟订单处理过程
    print(f"Processing order {order_id}")

Celery 与不同消息代理的集成方式各有不同，开发者可以根据业务需求选择合适的消息代理。

3. 🔐 消息的可靠性

3.1 确保消息的幂等性处理，避免重复消费

消息队列系统中的一个关键问题是如何确保消息的幂等性。幂等性是指无论操作执行多少次，结果都是一样的。在消息队列中，如果消费者重复消费同一条消息，可能会导致数据不一致或重复处理。因此，确保消息幂等性处理是系统设计中的关键部分。

幂等性实现策略：

唯一标识符：为每条消息分配一个唯一的 ID，消费者在处理消息时检查该 ID 是否已经处理过，如果已处理，则跳过该消息。
幂等操作：确保业务逻辑本身是幂等的，比如在数据库中插入数据时，使用 INSERT IGNORE 或 UPSERT 操作，避免重复插入。

示例代码：

import hashlib

processed_messages = set()

def process_message(message):
    # 计算消息的唯一哈希值
    message_id = hashlib.sha256(message.encode()).hexdigest()
    
    # 检查是否已处理
    if message_id in processed_messages:
        print("Message already processed")
    else:
        # 执行消息处理逻辑
        print(f"Processing message: {message}")
        processed_messages

.add(message_id)

3.2 死信队列的应用

死信队列（Dead Letter Queue, DLQ）是指消息在多次尝试处理失败后，未被成功处理的消息队列。它帮助系统处理异常或未能正常处理的消息，避免因为某些特定消息导致系统卡住。

当消息在主队列中无法被正确消费时，通常会被转移到死信队列中，开发人员可以定期查看死信队列，找到问题消息并进行处理。

4. 📊 任务监控与重试机制

4.1 Celery 的任务状态监控与失败重试

Celery 提供了丰富的任务状态监控功能，开发人员可以通过这些功能查看任务的执行状态、任务的完成情况、任务失败的次数等。

任务状态监控：

Celery 提供了一个后台工具 Flower，可以实时监控任务的执行情况。

# 安装 Flower
pip install flower

# 启动 Flower
celery -A tasks flower

通过 Flower 的 Web UI，可以查看每个任务的执行状态。

失败重试机制：

当任务执行失败时，Celery 提供了自动重试机制，开发者可以设置任务的最大重试次数和重试间隔时间。

@app.task(bind=True, max_retries=3)
def send_sms(self, phone_number, message):
    try:
        # 模拟发送短信
        print(f"Sending SMS to {phone_number}")
        raise Exception("Failed to send SMS")
    except Exception as exc:
        # 任务失败时，自动重试
        self.retry(exc=exc, countdown=5)

在上述代码中，send_sms 任务在执行失败时会自动重试，最多重试 3 次，每次重试间隔 5 秒。

通过 Celery 的重试机制和状态监控，系统可以更好地处理任务失败情况，确保任务的稳定性与可靠性。

原文地址：https://blog.csdn.net/weixin_52392194/article/details/142350555

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：第二百三十九节 JPA教程 - JPA一对一延迟加载示例
下一篇：js MessageChannel 两个独立的窗口双向实时通信

rabbitmq 短信验证码
新建一个springboot项目,引入web，redis，rabbitmq。
阅读更多2024-09-21
tomcat中间件漏洞CVE-2017-12615，后台弱口令部署war包，CVE-2020-1938
2. ⽂件上传成功后，默认会在⽹站根⽬录下⽣成和war包名称⼀致得⽬录，然后⽬录中得⽊⻢就是压缩前的⽂件名。漏洞复现 http://172.16.1.22:8080/manager/html。制作W
阅读更多2024-09-21
力扣 438找到字符串中所有字母异位词
通过滑动窗口进行遍历，通过"hash"将字符串子串映射到异位词表示空间每一个表示代表了一个异位词空间(一个字符串的所有元素的全排列广义上讲，以上方法都属于一种hash。
阅读更多2024-09-21
数据结构 - 树与二叉树
/数据域//左结点与右结点的指针。
阅读更多2024-09-21
【STL】map和set相关知识详细梳理
1. set是按照一定次序存储元素的容器。2. 在set中，元素的value就是key，并且每个value必须是唯一的。set中的元素不能在容器中修改(元素总是const)，但是可以从容器中插入或删除
阅读更多2024-09-21
Day4:杨辉三角
题目：给定一个非负整数numRows,生成杨辉三角的前numRows行。在杨辉三角中，每个数就是左上方和右上方数的和。
阅读更多2024-09-21
Android Retrofit源码分析（一）：Retrofit是什么？和OkHttp的区别是什么？为什么需要他？
为什么会出现Retrofit？它的作用是什么？
阅读更多2024-09-21
微服务远程调用（nacos及OpenFeign简单使用）
问题：在微服务中，每个项目是隔离开的，当有一个项目请求其他项目中的数据时，必须发起网络请求，本文即对此问题展开讨论。
阅读更多2024-09-21
数字签名和CA数字证书的核心原理
看了蛋老师的视频就很容易理解了，首先对服务器的公钥和信息进行哈希运算得到一个短字符串，然后用CA机构中的私钥对这一短字符串进行加密就得到了一个数字签名，然后就这个数字签名放到数字证书中，同时服务器的公
阅读更多2024-09-21
git 操作远程别名
【代码】git 操作远程别名。
阅读更多2024-09-21

Python Web开发：消息队列与异步任务