任务队列解决什么

一个接口里要发邮件、生成报表、压缩图片，用户等得心焦，还容易被超时打断。任务队列把这类耗时操作从请求链路里拆出去，让接口先返回「已受理」，后台慢慢干。Celery 就是 Python 生态里最流行的任务队列框架。

Celery 的架构由三部分组成：生产者（业务代码把任务丢进队列）、Broker（消息中间件，负责暂存任务，常见用 Redis 或 RabbitMQ）、Worker（真正执行任务的进程）。任务在队列里排队，多个 Worker 并发消费，天然支持水平扩展。

什么时候该上任务队列？判断标准很简单：这个操作能不能异步？如果用户不需要等它的结果，而且它比较耗时，就值得拆成任务。比如发验证码邮件、导出大数据、视频转码、定时清理数据。反过来，用户要立刻看到结果的操作（比如登录校验），不应该进队列。

引入队列也带来代价：多了一个中间件要运维，任务执行的时序变复杂，失败重试要自己设计。小项目先别急着上，等真的出现「接口太慢」「任务超时」再引入不迟。

任务队列的本质是「把时间从请求里偷出来」。
本文核心知识点
  • BrokerRedis 或 RabbitMQ 做消息中间件。
  • 任务@app.task 定义可异步执行的函数。
  • Worker后台进程并发消费任务。
  • 定时beat 调度周期任务。
Celery 任务队列配图：蓝调渐变背景上任务流水线经传送带送入处理枢纽的扁平科技插画
任务进队列，Worker 慢慢消化，接口秒回。

安装与基础配置

Celery 本体加一个 Broker 的适配库就够起步。Redis 做 Broker 最简单，一台机器就能跑。

pip install celery redis

新建一个 celery_app.py，把应用对象建好。这里指定 Redis 做 Broker，同时也用它存任务结果。

from celery import Celery

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

app.conf.timezone = "Asia/Shanghai"
app.conf.task_serializer = "json"
app.conf.result_serializer = "json"

时区要显式配成 Asia/Shanghai，否则定时任务按 UTC 走，会比本地时间差 8 小时。序列化统一用 JSON，跨语言调试都方便。

定义与调用任务

在 tasks.py 里用 @app.task 装饰一个函数，它就变成了可异步执行的任务。任务里写真实业务：发邮件、生成报表、调第三方接口。

from celery_app import app
import time

@app.task
def send_welcome_email(user_id):
    # 模拟发邮件耗时
    time.sleep(2)
    print(f"欢迎邮件已发送给用户 {user_id}")
    return {"user_id": user_id, "status": "sent"}

业务代码里调用时不用直接执行，而是用 .delay 把任务丢进队列，立刻返回一个 AsyncResult 对象，接口马上就能响应。

# views.py
def register(request):
    # 业务处理
    user_id = create_user(request)
    # 异步发邮件，不阻塞响应
    send_welcome_email.delay(user_id)
    return {"status": "ok", "user_id": user_id}

启动 Worker 的命令：celery -A tasks worker --loglevel=info。Worker 起来后，队列里的任务会被逐个消费。开多个 Worker 进程或加多台机器，任务吞吐量就上来了。

结果存储与状态查询

任务执行完的结果存在 backend（Redis）里，业务方可以用任务 ID 轮询状态，拿到返回值。

result = send_welcome_email.delay(42)

# 查看任务是否完成
print(result.ready())       # False / True
print(result.state)         # PENDING / SUCCESS / FAILURE

# 阻塞等结果（一般不在 Web 请求里用）
value = result.get(timeout=10)
print(value)                # {"user_id": 42, "status": "sent"}

Web 请求里别用 get 阻塞等结果，那就失去异步的意义了。正确姿势是前端先拿到 task_id，再用轮询接口查询状态，状态好了再展示结果。result_expires 要配一个合理值，默认一天后结果自动过期，避免 Redis 里堆一堆没人要的结果。

定时任务与队列分流

定时任务（比如每天凌晨清理缓存、每周生成报表）用 Celery beat 调度。beat 是一个独立的调度进程，到点把任务推进队列。

from celery.schedules import crontab

app.conf.beat_schedule = {
    "cleanup-every-midnight": {
        "task": "tasks.cleanup_expired",
        "schedule": crontab(hour=0, minute=30),
    },
    "report-every-monday": {
        "task": "tasks.weekly_report",
        "schedule": crontab(day_of_week="mon", hour=9),
    },
}

启动命令是 celery -A tasks beat。注意 beat 和 worker 是两个进程，要分别启动，也可以用 beat 内嵌到 worker 的方式减少运维。

任务多了以后，可以把任务按类型分到不同队列，用不同优先级的 Worker 处理。比如发邮件的任务慢但不紧急，放 low 队列；图片转码任务吃 CPU，单独一个队列限并发。

app.conf.task_routes = {
    "tasks.send_welcome_email": {"queue": "email"},
    "tasks.transcode_video": {"queue": "heavy"},
}

# worker 按队列消费
# celery -A tasks worker -Q email --loglevel=info