Python高级技术之：`Python`中的`Message Queue`：`Celery`和`RabbitMQ`在异步任务中的实践。 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

各位观众老爷，大家好！我是你们的老朋友，Bug终结者，今天咱们不聊风花雪月，就来聊聊Python世界里的“快递小哥”——消息队列。

没错，今天的主题就是：Python中的Message Queue，特别是Celery和RabbitMQ在异步任务中的实践。准备好了吗？让我们一起踏上这段降妖伏魔的旅程吧！

一、为什么要用“快递小哥”？

想象一下，你在网上购物，点了个“立即购买”，然后浏览器就卡死不动了，等了半天啥反应都没有，你会不会想把电脑砸了？这就是同步任务的弊端。用户请求直接触发耗时操作，用户必须等待，用户体验极差！

而异步任务呢？你点了“立即购买”，页面告诉你“订单已提交，正在处理…”，然后你就可以继续逛其他商品了。订单处理（扣款、生成订单等等）在后台默默进行。这就是异步任务的魅力！

那么，问题来了，怎么实现异步呢？这就需要我们的“快递小哥”——消息队列上场了。

二、 “快递小哥”的原理：消息队列

消息队列（Message Queue，简称MQ）就像一个中转站。应用程序A（生产者）把消息扔到MQ里，应用程序B（消费者）从MQ里取出消息进行处理。生产者和消费者之间解耦了，不再需要直接交互。

用个更接地气的例子：

生产者（Producer）： 你点了外卖，相当于“生产”了一个“点餐消息”。
消息队列（Message Queue）： 外卖平台的服务器，负责接收、存储和分发“点餐消息”。
消费者（Consumer）： 商家接到你的订单，开始做饭，相当于“消费”了“点餐消息”。

好处显而易见：

解耦： 你不用直接跟商家对话，平台帮你搞定。
异步： 你下了单就可以干别的，不用等着商家做饭。
削峰： 商家一下子接到100个订单也不会崩溃，平台会慢慢把订单分发给商家。
容错： 就算商家服务器挂了，订单也不会丢，平台会等商家恢复后再分发。

三、 Celery：Python异步任务的瑞士军刀

Celery是一个强大的Python分布式任务队列。它可以让你轻松地创建、调度和执行异步任务。就像一把瑞士军刀，功能齐全，使用方便。

1. 安装Celery

pip install celery

2. 选择一个“快递站”：Broker

Celery需要一个消息中间件作为“快递站”，也就是Broker。常用的Broker有：

RabbitMQ： 功能强大，性能稳定，生产环境首选。
Redis： 速度快，配置简单，适合小型项目。

这里我们选择RabbitMQ。

3. 安装RabbitMQ

不同操作系统安装RabbitMQ的方式不同，请参考官方文档： https://www.rabbitmq.com/download.html

4. 一个简单的Celery例子

# tasks.py
from celery import Celery
import time

# Celery配置
app = Celery('my_tasks', broker='amqp://guest:guest@localhost:5672//')  # 这里的broker地址要根据你的RabbitMQ配置修改

@app.task
def add(x, y):
    time.sleep(5)  # 模拟耗时操作
    return x + y

# main.py
from tasks import add

# 触发异步任务
result = add.delay(4, 4)
print("任务已提交，正在后台执行...")
print(f"任务ID: {result.id}")

# 获取任务结果（可选）
# print(f"任务结果: {result.get()}")  # 会阻塞直到任务完成

5. 运行Celery Worker

打开一个终端，运行Celery Worker：

celery -A tasks worker --loglevel=info

6. 运行主程序

打开另一个终端，运行主程序：

python main.py

你会看到输出：

任务已提交，正在后台执行...
任务ID: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx

然后，在Celery Worker的终端里，你会看到任务开始执行，并最终输出结果。

7. 代码解释

app = Celery('my_tasks', broker='amqp://guest:guest@localhost:5672//')：创建Celery应用，指定应用名称和Broker地址。
@app.task：将一个函数装饰成Celery任务。
add.delay(4, 4)：触发异步任务。 delay函数会将任务扔到RabbitMQ里，等待Celery Worker来处理。
result.id：任务的唯一ID。
result.get()：获取任务结果。注意，get()函数会阻塞，直到任务完成。

四、 Celery进阶：配置、任务状态、错误处理

1. Celery配置

Celery的配置非常灵活，可以通过多种方式进行配置：

配置文件： 可以将配置信息写在一个配置文件里，然后在Celery应用中加载。
环境变量： 可以通过环境变量来设置配置信息。
直接在代码中配置： 就像我们在上面的例子中那样。

常用的配置项：

配置项	描述
`broker_url`	Broker地址。
`result_backend`	存储任务结果的Backend。常用的有`redis`、`database`等。
`task_serializer`	任务序列化方式。常用的有`pickle`、`json`等。
`result_serializer`	结果序列化方式。
`accept_content`	接受的内容类型。
`timezone`	时区。
`enable_utc`	是否启用UTC时间。
`task_routes`	任务路由规则。可以将不同的任务路由到不同的队列。
`task_acks_late`	任务确认机制。`True`表示在任务完成后才确认，`False`表示在任务开始前就确认。
`worker_prefetch_multiplier`	Worker预取任务的数量。
`worker_concurrency`	Worker并发执行任务的数量。

例如，使用配置文件进行配置：

# celeryconfig.py
broker_url = 'amqp://guest:guest@localhost:5672//'
result_backend = 'redis://localhost:6379/0'
task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = 'Asia/Shanghai'
enable_utc = True

# tasks.py
from celery import Celery
from celery.schedules import crontab

# 从配置文件加载配置
app = Celery('my_tasks')
app.config_from_object('celeryconfig')

@app.task
def add(x, y):
    time.sleep(5)
    return x + y

2. 任务状态

Celery可以跟踪任务的状态。常用的状态有：

PENDING：任务等待执行。
STARTED：任务开始执行。
SUCCESS：任务执行成功。
FAILURE：任务执行失败。
RETRY：任务重试。
REVOKED：任务被撤销。

可以通过result.state来获取任务状态。

# main.py
from tasks import add
import time

result = add.delay(4, 4)

while result.state != 'SUCCESS':
    print(f"任务状态: {result.state}")
    time.sleep(1)

print(f"任务结果: {result.get()}")

3. 错误处理

Celery提供了多种方式来处理任务执行过程中的错误：

重试： 可以设置任务自动重试。
忽略错误： 可以忽略任务执行过程中的错误。
自定义错误处理函数： 可以定义自己的错误处理函数，在任务执行失败时被调用。

例如，设置任务自动重试：

# tasks.py
from celery import Celery

app = Celery('my_tasks', broker='amqp://guest:guest@localhost:5672//')

@app.task(bind=True, max_retries=3)  # bind=True表示将task实例传递给函数，max_retries=3表示最大重试次数
def add(self, x, y):
    try:
        # 模拟可能出错的操作
        result = x / (y - 4)
        return result
    except Exception as exc:
        # 发生错误时重试
        raise self.retry(exc=exc, countdown=5)  # countdown=5表示5秒后重试

五、 RabbitMQ：可靠的消息传递专家

RabbitMQ是一个开源的消息队列中间件。它实现了AMQP（Advanced Message Queuing Protocol）协议，提供了可靠的消息传递机制。

1. RabbitMQ的核心概念

Producer： 消息生产者。
Exchange： 交换机。接收生产者发送的消息，并根据路由规则将消息路由到一个或多个队列。
Queue： 队列。存储消息，等待消费者来消费。
Binding： 绑定。将Exchange和Queue绑定在一起，定义消息的路由规则。
Consumer： 消息消费者。从队列中获取消息进行处理。

2. Exchange的类型

RabbitMQ支持多种类型的Exchange：

Direct Exchange： 直接交换机。根据Routing Key完全匹配的原则将消息路由到队列。
Fanout Exchange： 扇形交换机。将消息广播到所有绑定的队列。
Topic Exchange： 主题交换机。根据Routing Key的模式匹配原则将消息路由到队列。
Headers Exchange： 首部交换机。根据消息的Headers匹配原则将消息路由到队列。

3. RabbitMQ的持久化

为了保证消息的可靠性，RabbitMQ提供了持久化机制。可以将Exchange、Queue和消息都设置为持久化。这样，即使RabbitMQ服务器重启，消息也不会丢失。

4. RabbitMQ的确认机制

RabbitMQ提供了两种确认机制：

Publisher Confirms： 生产者确认。生产者发送消息后，RabbitMQ会返回一个确认消息，表示消息已经成功接收。
Consumer Acknowledgements： 消费者确认。消费者从队列中获取消息并处理后，需要向RabbitMQ发送一个确认消息，表示消息已经成功消费。

六、 Celery + RabbitMQ：黄金搭档

Celery和RabbitMQ是天生一对。 Celery负责任务的调度和执行，RabbitMQ负责消息的传递和存储。它们一起构建了一个强大的异步任务处理系统。

1. 如何选择Exchange类型？

Direct Exchange： 适用于需要精确路由的场景，例如，将不同类型的任务路由到不同的队列。
Fanout Exchange： 适用于需要广播消息的场景，例如，发送通知消息。
Topic Exchange： 适用于需要根据模式匹配进行路由的场景，例如，根据日志级别将日志消息路由到不同的队列。

2. 如何保证消息的可靠性？

将Exchange、Queue和消息都设置为持久化。
启用Publisher Confirms和Consumer Acknowledgements。

3. 如何优化Celery + RabbitMQ的性能？

合理设置Worker的并发数。
调整RabbitMQ的配置，例如，调整TCP Buffer Size。
使用连接池，减少连接的开销。

七、总结

今天我们一起学习了Python中的Message Queue，特别是Celery和RabbitMQ在异步任务中的实践。希望大家能够掌握这些知识，并在实际项目中灵活运用。

最后，记住，技术是为人类服务的，不要为了技术而技术。祝大家写出优雅、高效的代码！

感谢各位的观看，下次再见！

发表回复 取消回复

发表回复取消回复