《Python实战进阶》No 9：使用 Celery 实现异步任务队列

news2025/4/20 14:51:03

第9集：使用 Celery 实现异步任务队列

引言

在现代 Web 应用中，许多操作（如发送邮件、处理文件上传、执行复杂计算等）可能需要耗费较长时间。如果这些操作直接在主线程中执行，会导致用户请求阻塞，降低用户体验。为了解决这一问题，我们可以使用 Celery 来实现异步任务队列。

Celery 是一个强大的分布式任务队列框架，支持 Python 编写的异步任务调度和后台任务处理。本篇将详细介绍如何使用 Celery 构建异步任务队列，并结合实际案例展示其在 Flask 和 Django 项目中的应用。
在这里插入图片描述

1. 什么是 Celery？

Celery 是一个基于消息队列的任务调度工具，它允许你将耗时的任务从主线程中分离出来，交由后台进程异步执行。Celery 的核心组件包括：

任务生产者（Producer）：负责创建任务并将其发送到消息队列。
消息代理（Broker）：用于存储任务队列，常见的代理有 RabbitMQ 和 Redis。
任务消费者（Worker）：从消息队列中获取任务并执行。

2. 安装与配置

2.1 安装 Celery

首先，确保你的环境中已安装 Celery 和消息代理（这里以 Redis 为例）：

pip install celery redis

2.2 配置 Redis

Redis 是 Celery 常用的消息代理之一。你可以通过以下命令安装 Redis：

sudo apt install redis-server

启动 Redis 服务：

sudo systemctl start redis

验证 Redis 是否正常运行：

redis-cli ping
# 如果返回 "PONG"，说明 Redis 已成功启动。

3. 使用 Celery 的基本流程

以下是使用 Celery 的基本步骤：

创建 Celery 应用实例。
定义异步任务。
启动 Celery Worker。
在主程序中调用任务。

接下来我们将通过一个简单的示例演示这些步骤。

4. 示例：异步发送邮件

假设我们需要实现一个功能：当用户注册时，系统会向用户发送一封欢迎邮件。由于发送邮件是一个耗时操作，我们将其封装为 Celery 异步任务。

4.1 创建 Celery 应用

在项目根目录下创建 celery_app.py 文件：

from celery import Celery

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

# 定义异步任务
@app.task
def send_welcome_email(user_email):
    print(f"Sending welcome email to {user_email}...")
    # 模拟发送邮件的耗时操作
    import time
    time.sleep(5)
    print(f"Welcome email sent to {user_email}")

4.2 启动 Celery Worker

在终端中启动 Celery Worker：

celery -A celery_app worker --loglevel=info

这将启动一个 Celery Worker，监听任务队列并执行任务。

4.3 调用异步任务

在主程序中调用 send_welcome_email 任务。以下是一个 Flask 示例：

from flask import Flask, request, jsonify
from celery_app import send_welcome_email

app = Flask(__name__)

@app.route('/register', methods=['POST'])
def register():
    user_email = request.json.get('email')
    if not user_email:
        return jsonify({"error": "Email is required"}), 400
    
    # 异步调用任务
    send_welcome_email.delay(user_email)
    
    return jsonify({"message": "Registration successful. Welcome email will be sent shortly."}), 200

if __name__ == '__main__':
    app.run(debug=True)

4.4 测试

启动 Flask 应用：

python app.py

使用 Postman 或 curl 发送 POST 请求：

curl -X POST http://127.0.0.1:5000/register -H "Content-Type: application/json" -d '{"email": "test@example.com"}'

你会看到 Celery Worker 输出日志，表明任务正在异步执行。

5. Celery 的高级特性

5.1 定时任务

Celery 支持定时任务，可以用来定期执行某些操作。例如，每天凌晨清理过期数据：

安装 Celery 的扩展包 celery[redis] 和 celery[schedule]。
配置定时任务：

from celery.schedules import crontab

app.conf.beat_schedule = {
    'cleanup-every-midnight': {
        'task': 'tasks.cleanup_expired_data',
        'schedule': crontab(hour=0, minute=0),
    },
}

启动 Celery Beat：

celery -A celery_app beat --loglevel=info

5.2 任务结果存储

默认情况下，Celery 不会存储任务的执行结果。如果需要查看任务状态或结果，可以配置结果后端（如 Redis 或数据库）：

app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

然后可以通过 AsyncResult 获取任务结果：

from celery.result import AsyncResult

result = AsyncResult(task_id, app=app)
print(result.status)  # 查看任务状态
print(result.result)  # 查看任务结果

6. 在 Django 中使用 Celery

如果你使用的是 Django，可以通过 django-celery-results 和 django-celery-beat 扩展来简化 Celery 的集成。

安装依赖：

pip install django-celery-results django-celery-beat

配置 settings.py：

INSTALLED_APPS += [
    'django_celery_results',
    'django_celery_beat',
]

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'