Python中Celery分布式任务队列的介绍与使用指南

pip install celery redis

# tasks.py
from celery import Celery

# 创建 Celery 实例，'my_app' 是当前项目的名称
app = Celery('my_app',
             broker='redis://localhost:6379/0',      # 使用 Redis 作为 Broker
             backend='redis://localhost:6379/0')     # 使用 Redis 作为 Backend

# 定义一个任务
@app.task
def add(x, y):
    print(f"正在计算 {x} + {y}...")
    # 模拟一个耗时操作
    import time
    time.sleep(5)
    result = x + y
    print(f"计算完成！结果是 {result}")
    return result

@app.task
def send_email(to, subject, body):
    # 模拟发送邮件的逻辑
    print(f"正在向 {to} 发送邮件，主题：{subject}")
    # ... 实际的发信代码 ...
    return f"邮件已发送至 {to}"

celery -A tasks worker --loglevel=info

# 在 Python Shell 中
from tasks import add, send_email

# 1. 异步调用 - 最常用的方式
# 这行代码会立即返回一个 AsyncResult 对象，而不会阻塞程序
result = add.delay(4, 6) # .delay() 是快捷的异步调用方法
print(f"任务ID：{result.id}") # 立即打印任务ID，此时任务正在后台运行

# 2. 获取任务结果（可选）
# 因为我们的任务有 5 秒休眠，所以需要等待一下再获取结果
print("正在等待结果...")
value = result.get(timeout=10) # .get() 会阻塞，直到任务完成并返回结果
print(f"任务结果是：{value}")

# 3. 检查任务状态（可选）
print(f"任务状态：{result.status}") # 可能是 PENDING, SUCCESS, FAILURE 等

# 4. 调用发送邮件的任务
email_result = send_email.delay('user@example.com', '欢迎光临', '感谢您注册！')
print(f"邮件任务ID：{email_result.id}")

# 安装 Flower
pip install flower

# 启动 Flower (在项目目录下)
celery -A tasks flower

# 然后在浏览器中访问 http://localhost:5555

# app.py
from flask import Flask, jsonify
from celery import Celery

# 配置 Flask
app = Flask(__name__)

# 配置 Celery（通常从配置文件读取）
def make_celery(app):
    celery = Celery(
        app.import_name,
        broker='redis://localhost:6379/0',
        backend='redis://localhost:6379/0'
    )
    # 可选：更新Celery配置，使其与Flask配置保持一致
    celery.conf.update(app.config)
    return celery

celery = make_celery(app)

@celery.task
def background_task():
    # 一个模拟的耗时任务
    import time
    time.sleep(10)
    return "后台任务完成！"

@app.route('/start_task', methods=['POST'])
def start_task():
    # 在视图函数中触发后台任务
    task = background_task.delay()
    return jsonify({'task_id': task.id}), 202 # 202 Accepted 表示请求已被接受处理

@app.route('/check_status/<task_id>')
def check_status(task_id):
    task = background_task.AsyncResult(task_id)
    if task.state == 'PENDING':
        response = {'state': task.state, 'status': '任务正在排队或执行中...'}
    elif task.state == 'SUCCESS':
        response = {'state': task.state, 'result': task.result}
    else: # FAILURE 等其他状态
        response = {'state': task.state, 'status': str(task.info)}
    return jsonify(response)

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