Django-Celery

这篇讲什么

• Celery 解决两件事：把不在「请求—响应」关键路径上的慢活摘出去，以及在没有请求时也能按周期干活。这一篇是两篇合一：前半按架构讲清 Broker / Worker / Result Backend / Beat 各管什么、一条任务的生命周期长什么样、跨语言和高可用各自能做到什么程度；后半是操作手册，从安装一路跑到 beat 定时、admin 看任务、单元测试。
• 结论都来自本机实测（Django 5.2.19 + Celery 5.6.3 + Redis，Windows 与 WSL 两边都跑过），包括 solo 池与多 worker 的耗时对比、队列积压数、结果键的 TTL、跨语言投递的报错原文。
• 旧笔记用的 djcelery 早已废弃，实测报错一并贴出。

这篇的年代说明

• 旧笔记里的 djcelery（django-celery）在 Django 1.x 时代就废弃了，PyPI 上现在连包都没有（实测 ERROR: No matching distribution found for djcelery）。它带来的 python manage.py celery worker 也随之消失，实测在 Django 5.2 + Celery 5.6 里跑旧命令只会得到：

  Unknown command: 'celery'
  Type 'manage.py help' for usage.

• 定时任务与结果存储改用官方生态的 django-celery-beat + django-celery-results，命令行入口换成 celery -A 项目名 worker。

• 另外注意 django-celery-beat 2.9.0 的元数据写死 Django<6.1：在 5.2.19 基线上它装得上、导入也正常，但换到 6.1.x 时装它会让 pip 静默把 Django 降到 6.0.10（实测）。装插件前先看 Requires-Dist。

解决什么问题

• 慢：用户在等 response。如果视图里要跑一段 10 秒的动作（调第三方、生成报表、发一批短信），用户就得干等 10 秒。
• 没人触发：网站每小时要同步一次天气。HTTP 请求是「有人访问才有动作」，定时的事没有请求来推它。
• 两个问题同一个答案：把动作从「请求-响应」这条线上摘出去，交给一个常驻进程（worker）去执行；再给它配一个闹钟（beat）。
• 和两个常见替代方案的分工：
  • threading / 进程内 ThreadPoolExecutor：同一个 Web 进程里，重启就没了，也没有重试和结果存储。
  • 系统的 cron / Windows 计划任务：只能定时，不能「用户点一下就立刻异步跑」，也没有任务状态。
  • Celery：两者都覆盖，还带 broker 持久化、结果后端、重试、编排（group/chord）。
• 顺带一句和 Signals 的分工：信号解决「一处变化要多处响应」，但它是同步的；Celery 解决「现在不做，稍后在别的进程做」。常见组合是信号负责发现变化，Celery 负责干活（详见《Django-Signals》）。

Celery 架构之美

• 概念清晰：Task / Broker / Worker / Result Backend / Beat 五个角色各司其职，一个 @shared_task 装饰器就把普通函数变成任务。
• API 简洁：add.delay(4, 4) 投递，result.get() 取结果，两行搞定；进阶需求再上 apply_async 的参数。
• 关键设计可替换：broker、结果后端、序列化器、调度器都是配置项，换 RabbitMQ 只要改一行 URL。
• 协议开放：消息就是 broker（AMQP / Redis）里的普通消息，别的语言只要能连 broker、按格式投递，就能和 Celery worker 对话。
• 天生可扩展：worker 数量与队列一一对应，加进程就能加吞吐（下面有实测加速比）。

四个角色与核心概念

角色	是什么	本机用的
Task	一个 Python 函数，加上装饰器就成了任务	core/tasks.py 里的 add(x, y)
Periodic Task	按周期执行的任务，可用固定间隔或 crontab 表达式	django_celery_beat.PeriodicTask 表里的一行
Broker 消息代理	存「待办」的队列，Web 进程往里投，Worker 从里取；可以是 Redis、RabbitMQ、Amazon SQS，也可以多个做高可用	Redis 8.2.8（WSL 里），redis://:password@127.0.0.1:6379/0，队列名 celery
Worker 工作进程	独立进程，循环取任务并执行；可以多个，做高可用和横向扩展	celery -A proj worker，本机 prefork 池 20 并发
Result Backend 结果存储	存任务的返回值/状态	django-db（MySQL 的 django_celery_results_taskresult 表）或 Redis db1 的 celery-task-meta-*
Beat 调度器	闹钟进程，按周期把任务投进队列	celery -A proj beat + DatabaseScheduler

• 关键点：Web 进程和 Worker 是两个进程。所以 Worker 里不能假设「Django 已经启动好了」，它自己会 django.setup()；数据库连接、文件句柄都得自己管。

• 一次投递的完整生命周期（时间点都是本机实测的）：

  1. Web 进程调 add.delay(4, 4)
     -> 立刻拿到 task_id，state=PENDING（0.00 s）
  2. kombu 把消息序列化后 LPUSH 进 Redis 的 celery 队列
  3. worker 取到消息，按 name 在注册表里找函数，执行
     -> Task core.tasks.add[...] succeeded in 0.055s: 33
  4. 结果写进 Result Backend（django-db 就是写一行表）
  5. 调用方 AsyncResult(id).get() 从后端读到 8，state=SUCCESS
     -> 整条链路往返 0.52 s

  • 第 3 步之前 worker 不在？消息就留在队列里不丢，实测 redis-cli -n 0 LLEN celery 能看到积压数=1，get(timeout=3) 抛 TimeoutError。
  • 这也解释了 PENDING 的二义性：从没投过、还在排队、已被丢弃，三种情况都是它。要区分就开 CELERY_TASK_TRACK_STARTED 拿 STARTED。

Celery 的跨平台

• 先把话说准：Celery 本体只有 Python 实现。跨语言的是它底下的消息协议——AMQP（RabbitMQ）和 Redis 的列表/发布订阅，都是公开协议。

• 这个说法能不能落地？本机做了一个实验：完全不 import celery，只用 redis-py 手工拼一条 Celery 协议 2 的消息塞进队列，看 worker 认不认。

  import base64
  import json
  import uuid
  
  import redis
  
  task_id = str(uuid.uuid4())
  body = json.dumps([[11, 22], {}, None]).encode()      # [args, kwargs, errbacks]
  message = {
      "body": base64.b64encode(body).decode(),
      "content-encoding": "base64",
      "content-type": "application/json",
      "headers": {
          "task": "core.tasks.add",                     # 必须和 worker 注册的名字一致
          "id": task_id,
          "argsrepr": "(11, 22)",
          "kwargsrepr": "{}",
          "origin": "langclient@another-host",
          "protocol": 2,
      },
      "properties": {
          "correlation_id": task_id,
          "delivery_mode": 1,
          "body_encoding": "base64",
          "delivery_tag": task_id,                      # 少了这个字段 kombu 会报错
          "delivery_info": {"exchange": "", "routing_key": "celery"},
      },
  }
  
  redis.Redis(host="127.0.0.1", port=6379, password="password", db=0) \
       .lpush("celery", json.dumps(message))

  worker 日志：Task core.tasks.add[8d180146-...] received
              Task core.tasks.add[8d180146-...] succeeded in 0.05526239995379001s: 33
  结果表里：('SUCCESS', '33')

• 少写一个字段会怎样，也实测了：properties 里不带 delivery_tag 时，worker 消费到消息却在 kombu 的传输层就崩了，任务根本没进入执行：

  File ".../kombu/transport/virtual/base.py", line 340, in __init__
      delivery_tag=properties['delivery_tag'],
  KeyError: 'delivery_tag'

• 结论：

  1. 别的语言可以给 Celery 投任务，前提是照抄 broker 的消息格式；这条路维护成本高，字段是 kombu 的私有约定，版本会变。
  2. 生产上更稳的做法是：非 Python 侧只当生产者时，走一个薄的 HTTP/MQ 网关交给 Python 投，或者用各语言的 Celery 兼容库（Node 的 node-celery、Go 的 Machinery 等，维护状态要自己评估），而不是手写 JSON。
  3. 反过来（worker 执行完，结果给别的语言读）就很干净：结果后端是 Redis 时，别的语言直接读 celery-task-meta-<id> 那个键就行，实测值是标准 JSON。

Celery 的高可用架构

• 三个可复制的层次，逐个给出本机能验证到的部分：

  • Broker 层高可用：Redis 用哨兵/集群，RabbitMQ 用镜像队列或仲裁队列；配置上体现为 CELERY_BROKER_URL 可以给多个地址（; 分隔），连不上会自动切换。这一层本机只有一个 Redis 实例，没法实测故障切换，属于设计原则。

  • Worker 层横向扩展：加进程即可。实测同一批 10 个 sleep(1) 的任务，用 solo 池（一次只能跑一个）：

    1 个 worker：10.61 s
    2 个 worker：5.54 s
    比值=1.91

    接近线性，说明任务确实被两个进程分摊了。前提是它们消费同一个队列。

  • 队列隔离：不同业务用不同队列，各自配 worker，慢任务不会拖死快任务。实测：

    celery -A proj worker -Q urgent -n urgent_only@%h --pool=solo

    在线 worker：['urgent_only@Y9000P']
    add.apply_async(args=[1], queue="urgent").get() -> 1

    这个 worker 只监听 urgent，celery 默认队列里的消息它一条都不碰。反过来投给没人消费的队列，任务就永远 PENDING（实测 3 秒后依旧 PENDING，消息躺在 broker 的列表里）。

• 结果层：django-db 把结果写进业务库，天然跟着数据库做备份；换成 Redis 结果后端时结果是有 TTL 的临时数据（实测 TTL=86400），别把它当长期审计。
• 调度层高可用要特别注意：beat 只能有一个实例。同时跑两个 beat，同一个 PeriodicTask 会被投两次。要做 beat 的容灾，用「主备 + 分布式锁」或者交给外部调度（Kubernetes CronJob 投任务），不要靠多开 beat。

落到 Django 项目里

• 概念到代码的对应关系，完整可跑的配置见下面「接到 Django 上：要改的四个文件」一节，这里只留一张速查表：

  概念	在 Django 项目里的位置
  Celery 实例	proj/celery.py（app.config_from_object("django.conf:settings", namespace="CELERY")）
  让 Django 启动时加载它	proj/__init__.py 里 from proj.celery import app as celery_app
  broker / 后端 / 时区	settings.py 的 CELERY_BROKER_URL、CELERY_RESULT_BACKEND、CELERY_TIMEZONE
  任务本体	各应用的 tasks.py（autodiscover_tasks 只认这个文件名）
  定时任务	django_celery_beat 的表 + CELERY_BEAT_SCHEDULER
  结果表	django_celery_results 的 TaskResult + CELERY_RESULT_EXTENDED
  视图里怎么用	r = my_task.delay(...)，把 r.id 返回给前端轮询

• 和 Signals 的配合（这也是架构层面值得记住的一条链路）：

  模型保存 → post_save 接收器（同步、快）→ transaction.on_commit → task.delay()（异步、慢活）

  接收器里只做「投任务」这一件事，别把耗时动作写在里面，否则用户还是在等。

安装（官方推荐路线，本机实测）

pip install "celery[redis]"          # 官方推荐：带 redis extra
pip install django-celery-beat django-celery-results

• 实测装完的版本：

  celery                 5.6.3
  kombu                  5.7.0
  redis                  8.1.0
  django-celery-beat     2.9.0
  django-celery-results  2.6.0
  Django                 5.2.19

• 一定要写 celery[redis]，不要只写 celery。 本机就踩过：环境里已经装了 celery 和 redis，但第一次连 Redis broker 直接崩：

  File ".../kombu/transport/redis.py", line 119, in <module>
      from packaging.version import Version
  ModuleNotFoundError: No module named 'packaging'

  • 原因在打包元数据里看得很清楚：packaging 只挂在 kombu 的 redis extra 上（Requires-Dist: packaging==26.3; extra == "redis"），裸装 celery + 裸装 redis 都不会带上它。
  • 修法就是补上 extra：pip install "celery[redis]"（实测只会多装一个 packaging，不会动 Django 版本）。

• 命令行入口是 celery.exe（在虚拟环境的 Scripts 目录里）。激活环境后直接 celery -A proj worker ...；没激活就用 python -m celery ...。

脱离 Django 的最小例子

• 三行定义一个任务：

  tasks.py

  from celery import Celery
  
  app = Celery("hello", broker="redis://127.0.0.1:6379/0")
  
  
  @app.task
  def add(x, y):
      return x + y

  run_task.py

  from tasks import add
  
  result = add.delay(4, 4)
  print("task result: %s" % result.get(timeout=10))

  celery -A tasks worker --loglevel=info

• 实测输出：

  task_id=7bd5509f-d39b-4672-b1ca-749d4df3f8c7
  task result: 8
  state=SUCCESS

• 但本机这台 Redis 设了 requirepass，所以旧笔记那句 broker='redis://localhost:6379/0' 直接跑会卡在认证上：

  OperationalError: Authentication required.

  URL 里带上密码就行：redis://:password@127.0.0.1:6379/0（password 走环境变量，不要把明文写进代码）。

接到 Django 上：要改的四个文件

• 结构长这样（proj 是项目包，core 是应用）：

  mysite/
  ├── manage.py
  ├── proj/
  │   ├── __init__.py      # 加一行：导出 celery_app
  │   ├── celery.py        # 新建：Celery 实例
  │   └── settings.py      # 加 CELERY_* 配置
  └── core/
      └── tasks.py         # 任务都写在这个文件名

  proj/celery.py

  import os
  
  from celery import Celery
  
  os.environ.setdefault("DJANGO_SETTINGS_MODULE", "proj.settings")
  
  app = Celery("proj")
  app.config_from_object("django.conf:settings", namespace="CELERY")
  app.autodiscover_tasks()
  
  
  @app.task(bind=True, ignore_result=True)
  def debug_task(self):
      print(f"Request: {self.request!r}")

  proj/__init__.py

  from proj.celery import app as celery_app
  
  __all__ = ("celery_app",)

• 三行配置逐个解释：

  • os.environ.setdefault(...)：worker 是独立进程，没人告诉它用哪个 settings，这行是兜底。
  • app.config_from_object("django.conf:settings", namespace="CELERY")：把 settings.py 里所有 CELERY_ 前缀的配置读进来，转成 Celery 的小写键。namespace 这个参数是关键，旧笔记漏了它，于是 BROKER_URL 那套顶层大写变量在新版里根本不生效。
  • app.autodiscover_tasks()：去每个 INSTALLED_APPS 里找名为 tasks 的模块并 import。

• settings.py 里要加的和要装的：

  settings.py

  INSTALLED_APPS = [
      "django.contrib.admin",
      "django.contrib.auth",
      "django.contrib.contenttypes",
      "django.contrib.sessions",
      "django.contrib.messages",
      "django.contrib.staticfiles",
      "django_celery_beat",       # 定时任务存数据库
      "django_celery_results",    # 结果存数据库
      "core",
  ]
  
  # ------------------------------------------------------------ #
  # Celery 5.x：全部走 CELERY_ 前缀
  # ------------------------------------------------------------ #
  import os
  
  _PW = os.environ["DJANGO_NOTE_DB_PW"]      # password 走环境变量，不要把明文写进代码
  
  CELERY_BROKER_URL = f"redis://:{_PW}@127.0.0.1:6379/0"
  CELERY_RESULT_BACKEND = "django-db"
  CELERY_CACHE_BACKEND = "django-cache"
  CELERY_TASK_TRACK_STARTED = True
  CELERY_TIMEZONE = "Asia/Shanghai"
  CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler"

  • CELERY_RESULT_BACKEND = "django-db" 是 django_celery_results 注册的后端别名，别写成 "database"（那是旧名）。
  • CELERY_BEAT_SCHEDULER 指向 DatabaseScheduler，定时任务就从数据库里读，可以在 admin 里改，不用重启 beat。

• 任务文件：

  core/tasks.py

  import time
  
  from celery import shared_task
  from django.db import connections
  
  from core.models import Order
  
  
  @shared_task
  def add(x, y):
      print("worker 里执行 add(%s, %s)" % (x, y))
      return x + y
  
  
  @shared_task
  def slow(seconds):
      time.sleep(seconds)
      return seconds
  
  
  @shared_task(name="core.orm_task")
  def orm_task(order_id):
      order = Order.objects.get(pk=order_id)
      order.mail_sent = True
      order.save()
      connections.close_all()      # worker 是长驻进程，用完的连接要关
      return order.no

  • 为什么用 @shared_task 而不是 @app.task：shared_task 不要求你的项目在 celery.py 里导出 app，Django 应用可以保持不依赖具体实例，测试时也能换 app。两种都能用，官方文档推荐 shared_task。

  • 旧笔记的 from celery import task + @task 在 Celery 5 已经删干净，实测：

    ImportError: cannot import name 'task' from 'celery'. Did you mean: 'Task'?

• 迁移，把 beat 和 results 的表建出来：

  python manage.py migrate

  Applying django_celery_beat.0019_alter_periodictasks_options... OK
  Applying django_celery_results.0014_alter_taskresult_status... OK

autodiscover 只认 tasks.py

• 实测：任务写在 core/other_tasks.py 时，worker 的 [tasks] 列表里没有它：

  . core.tasks.add
  . core.tasks.boom
  . core.tasks.slow

• 加上配置就有了：

  settings.py

  CELERY_IMPORTS = ("core.other_tasks",)

  . core.other_tasks.mul

• 旧笔记里写的是 CELERY_IMPORES（多个 R），实测这个拼错不会报任何错，只是任务列表里悄悄少了那个模块 —— 然后就是经典的「任务投出去了但 worker 说没注册」：

  Received unregistered task of type 'core.other_tasks.mul'

• 另一个坑：app.autodiscover_tasks() 是惰性的。在主进程里立刻打印 app.tasks 是空的，因为真正 import 发生在 app finalize（worker 启动）时。想确认注册结果，看 worker 启动横幅或 celery -A proj inspect registered。

启动：三件事按顺序来

# 1) broker（本机 Redis 跑在 WSL 里，已经是服务自启，这里只是示意）
wsl -d Ubuntu-26.04 -- systemctl status redis

# 2) worker
celery -A proj worker --loglevel=info

# 3) 需要定时任务时再加一个 beat
celery -A proj beat --loglevel=info

# 4) Web
python manage.py runserver

• 启动横幅里能核对配置有没有生效（实测）：

  -------------- celery@Y9000P v5.6.3 (recovery)
  - ** ---------- .> transport:   redis://:**@127.0.0.1:6379/0
  - *** --- * --- .> concurrency: 20 (prefork)
   -------------- [queues]
                  .> celery           exchange=celery(direct) key=celery
  [tasks]
    . core.orm_task
    . core.tasks.add
    . proj.celery.debug_task

  • transport 那行的 ** 是 Celery 自己把密码打了码，日志里不会泄明文。
  • [tasks] 里 core.orm_task 是因为我给它写了 name="core.orm_task"，覆盖了默认的 core.tasks.orm_task。名字是路由和 send_task 的依据，改名要谨慎。

• Windows 上要不要 --pool=solo？ 网上教程几乎都说「Windows 必须 solo」。本机实测：Celery 5.6.3 + Python 3.14 下默认 prefork 能正常起（concurrency: 20 (prefork)，子进程是 SpawnPoolWorker-1..20），任务也照样执行。

  • 但 solo 池一次只跑一个任务（max-concurrency: 1），调试时反而更容易看清日志，所以「本地开发用 --pool=solo，压测/生产用默认」是合理选择。
  • 真正在 Windows 上失效的是超时，见后面「Windows 的两个限制」。

投递任务

from core.tasks import add

r = add.delay(4, 4)              # 便捷写法
r = add.apply_async(args=[4, 4], kwargs={}, countdown=3, queue="urgent")
print(r.id, r.state)             # 任务 id 立刻就有，状态先可能是 PENDING
print(r.get(timeout=30))         # 阻塞等结果

• 实测的往返：add.delay(4, 4) 之后 state=PENDING，get(timeout=30) 返回 8，整个往返 0.52 秒。

• apply_async 常用的几个参数：

  参数	作用	实测
  countdown=3	3 秒后才执行	实际等待 3.54 秒
  eta=aware_datetime	到点执行（必须是带时区的 datetime）	now+2s 实测 2.53 秒
  queue="urgent"	投到别的队列	worker 加了 -Q celery,urgent 才有人消费，0.51 秒完成
  task_id=uuid()	自定义 id，方便事后 revoke / 查	用它查到结果行
  expires=10	10 秒内没被消费就丢弃	——

• 直接调用 add(5, 6) 是同步执行，不走队列（实测那行 print 出现在主进程终端里）。这个写法只该出现在单元测试里。

• 不 import 任务也能投：app.send_task("core.orm_task", args=[2])。跨项目、跨语言时常用，实测能拿到结果。

取结果：AsyncResult 的状态机

from celery.result import AsyncResult

r = AsyncResult(task_id)        # 换个进程、只凭 id 也能查
r.state                         # PENDING / STARTED / SUCCESS / FAILURE / REVOKED
r.successful(), r.result, r.traceback

• 状态含义与实测：

  状态	什么时候	实测
  PENDING	还没结果。注意它同时表示「任务从没存在过」	查一个瞎编的 id 也是 PENDING
  STARTED	任务真的开始执行（要开 CELERY_TASK_TRACK_STARTED）	表里 date_started 有值
  SUCCESS	正常返回	result 就是返回值
  FAILURE	抛异常	result 是异常对象本身
  REVOKED	被撤销	见下

• 因为 PENDING 有二义性，「前端轮询任务进度」不能只判 PENDING，要么开 STARTED，要么用 r.ready() + 自己存 id。

• 没有 worker 在跑时投任务，实测：

  刚投完 state=PENDING
  get(timeout=3) -> TimeoutError: The operation timed out.
  队列 celery 里积压的消息数=1

  也就是说任务不会丢，它躺在 broker 的队列里等。排查时可以直接看队列长度（redis-cli -n 0 LLEN celery）。

• revoke 实测：

  from celery import uuid
  
  rid = uuid()
  add.apply_async(args=[1, 2], task_id=rid, countdown=5)
  app.control.revoke(rid, timeout=5)

  revoke 后 AsyncResult.state=REVOKED
  表里这行=('REVOKED', '{"exc_type": "TaskRevokedError", "exc_message": ["revoked"], ...}')

  • 默认只是「记入黑名单，执行前丢弃」，正在跑的任务不会被杀。要真杀进程用 terminate=True（Linux 上有效，Windows 上不可靠）。

• 失败任务在客户端能拿到什么，实测：

  boom.get() 抛出：ValueError: 任务里故意抛的异常
  state=FAILURE  result=ValueError('任务里故意抛的异常')  traceback=''

  • get() 会把 Worker 里的异常在客户端重新抛出，异常类型和消息都保留了。

  • 但 traceback 是空的 —— 堆栈只在 Worker 的日志里：

    File "E:\...\core\tasks.py", line 31, in boom
        raise ValueError("任务里故意抛的异常")

  • 想让结果表里也存堆栈，要开 CELERY_RESULT_EXTENDED = True（见下一节）。

结果存储：django-db 还是 redis

• 两种后端的实测对比：

  配置	结果落在哪	实测证据	适合
  CELERY_RESULT_BACKEND = "django-db"	MySQL 表 django_celery_results_taskresult	表里 1 行，status='SUCCESS'、result='8'	想在 admin 里查、要长期留存
  "redis://:password@127.0.0.1:6379/1"	Redis db1 的 celery-task-meta-<id>	键类型 string、TTL=86400、值是 JSON；MySQL 里 0 行	结果量小、只要短期可查

  celery-task-meta-34edbdb2-a03d-4283-ace0-fa97f878ab9b
  {"status": "SUCCESS", "result": 5, "traceback": null, "children": [], "date_done": "2026-11-05T02:..."}

  • TTL 默认一天，由 CELERY_RESULT_EXPIRES（默认 24 小时）控制。

• django-db 模式下，表里一行长这样（实测）：

  task_id='f78aaa52-...' status='SUCCESS' result='8'
  task_name=None task_args=None worker=None traceback=''
  date_created=2026-11-04 18:37:10  date_started=2026-11-05 02:37:10  date_done=2026-11-04 18:37:10

• task_name / task_args / worker / traceback 全是空的，不是坏了，是默认不记。 打开开关才有：

  settings.py

  CELERY_RESULT_EXTENDED = True

  默认（未开 extended）：task_name=None task_args=None worker=None traceback=None
  开之后：task_name='core.tasks.add' task_args='"(20, 22)"' worker='celery@Y9000P'
          失败任务的 traceback 前 90 字='Traceback (most recent call last):\n  File "D:\\...\\site-p'

  • 注意 task_args 存的是 JSON 编码后的字符串，所以里面带引号（'"(20, 22)"'），按参数查的时候别忘了一层引号。
  • 定时任务的名字也在扩展列里：开了之后 periodic_task_name='probe-ext' 才填得进去。

• 时间戳有个 8 小时坑，值得单独说：date_started 比 date_created / date_done 晚 8 小时。

  • 原因（读源码 + 实测确认）：django_celery_results 写 date_started 用的是数据库的 Now()（即 MySQL 的 CURRENT_TIMESTAMP），而另外两列是 Python 侧的 UTC 值。本机 MySQL 的会话时区是 SYSTEM（=+08）：

    数据库时钟：NOW()=2026-11-05 02:37:06  UTC_TIMESTAMP()=2026-11-04 18:37:06  session_tz=SYSTEM

  • 实测改 TIME_ZONE = "Asia/Shanghai"、改 DATABASES["default"]["TIME_ZONE"] = "UTC" 都不会改变这个偏差（Django 4.1 起不再设置 MySQL 会话时区）。

  • 最省事的处理：不开 CELERY_TASK_TRACK_STARTED（默认就是关的），此时 date_started 是 NULL，不会有干扰；要开始就接受这一列仅作参考，业务时间一律用 date_created / date_done。

重试、编排

@shared_task(bind=True, max_retries=3)
def flaky(self, n):
    if self.request.retries == 0:
        raise self.retry(countdown=1, exc=RuntimeError("手滑了"))
    return n * 2

• bind=True 把任务实例自己作为第一个参数传进来（self），于是能读 self.request（重试次数、id、来源）并调用 self.retry()。实测 flaky(21) 第一次抛、1 秒后重试、最终 get() 返回 42，状态 SUCCESS。

• 自动重试也可以只靠配置：CELERY_AUTORETRY_FOR=(Exception,) 之类，但显式 self.retry() 更好读。

• 并行分组 group：

  from celery import group
  
  g = group(add.s(i, i) for i in range(5))()
  print(g.get(timeout=60))

  [0, 2, 4, 6, 8]

• 分组完再汇总 chord（前面全部完成才触发回调）：

  from celery import chord
  
  cr = chord([add.s(i, i) for i in range(10)])(summarize.s())
  print(cr.get(timeout=60))

  合计 [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

  • 回调拿到的是所有子任务返回值的列表，所以 summarize(total) 里的 total 是个 list，不是数字。
  • chord 需要一个能记账的后端。django-db 用 ChordCounter 表实现，实测可用；换成 Redis 后端也一样能跑。

• 链式 chain / starmap 同理，都是 s()（signature）之间的组合：chain(add.s(2, 2), add.s(4))()。

定时任务：beat

• 两种配法。静态写在 settings 里（改了要重启 beat）：

  settings.py

  from celery.schedules import crontab
  
  CELERY_BEAT_SCHEDULE = {
      "sync-weather": {
          "task": "core.tasks.add",
          "schedule": 10.0,                    # 秒
          "args": (2, 2),
      },
      "every-daily-2am": {
          "task": "core.tasks.slow",
          "schedule": crontab(hour=2, minute=0),
          "args": (1,),
      },
  }

• 动态存数据库（CELERY_BEAT_SCHEDULER = DatabaseScheduler 之后，在 admin 里增删改，不用重启）：

  from django_celery_beat.models import IntervalSchedule, PeriodicTask
  
  every10, _ = IntervalSchedule.objects.get_or_create(every=10,
                                                      period=IntervalSchedule.SECONDS)
  PeriodicTask.objects.create(
      name="probe-every-10s",
      task="core.tasks.add",
      kwargs='{"x": 1, "y": 1}',
      interval=every10,
      description="每 10 秒加一次",
  )

  • IntervalSchedule.period 的可选值实测是 days / hours / minutes / seconds / microseconds。
  • args / kwargs 存的是 JSON 字符串，不是 Python 列表；写成 args=[1, 2] 会报序列化错误。
  • CrontabSchedule 有 timezone 字段，本机填 Asia/Shanghai，否则「每天 2 点」会按 UTC 触发。

• 跑起来之后 beat 的日志（实测 45 秒窗口）：

  beat: Starting...
  DatabaseScheduler: Schedule changed.
  Scheduler: Sending due task probe-every-10s (core.tasks.add)
  Scheduler: Sending due task probe-one-off (core.tasks.add)
  Scheduler: Sending due task probe-static (core.tasks.add)

  45 秒里各任务被投递的次数={'probe-every-10s': 4, 'probe-one-off': 1, 'probe-static': 5}
  probe-every-10s: total_run_count=2 last_run_at=2026-11-04 18:42:59+00:00 enabled=True
  probe-one-off  : total_run_count=0 enabled=False（one_off 跑完自动停用）

  • 静态 CELERY_BEAT_SCHEDULE 和数据库里的 PeriodicTask 会同时生效，实测两套都在跑。
  • one_off=True 的任务投递一次后 enabled 自动变 False（日志里只出现一次），但 total_run_count 不一定同步涨 —— 这个计数是 beat 定期回写的，会滞后（实测日志发了 4 次，表里只记到 2）。别把它当精确审计。
  • 把 enabled 改成 False 之后立刻不再调度（实测 12 秒里计数没涨）。

• beat 只投任务、不执行任务，所以 worker 必须在跑。只起 beat 不起 worker，你会看到日志一切正常但什么都没发生。

在 Django admin 里看任务

• 装了 django_celery_beat / django_celery_results 并 migrate 之后，admin 里会多出两组菜单，实测全部 200：

  /admin/django_celery_beat/periodictask/     -> 200
  /admin/django_celery_beat/crontabschedule/  -> 200
  /admin/django_celery_results/taskresult/    -> 200

• 结果列表页默认按 -date_done 排序，能直接看到 SUCCESS / FAILURE 的行。想在这一页看到任务名、参数、堆栈，记得开 CELERY_RESULT_EXTENDED = True，否则那几列永远是空的。

单元测试：让任务同步跑

tests.py

from django.test import override_settings


@override_settings(CELERY_TASK_ALWAYS_EAGER=True)
def test_add(self):
    self.assertEqual(add.delay(2, 22).get(), 44)

• 实测 task_always_eager=True 时 delay() 直接在当前进程同步执行，AsyncResult 里状态立刻是 SUCCESS，id 也照样有：

  [15] task_always_eager=True：delay 变同步 -> get=42 state=SUCCESS

  • 所以测试里不需要起 worker。

• 默认情况下 eager 执行不会把异常抛出来（异常被收进结果）。要让测试失败得干脆，再加一个开关：

  task_eager_propagates=False：boom.delay() 不抛，返回异常对象
  task_eager_propagates=True ：同样的调用直接抛 ValueError: 任务里故意抛的异常

• 完全不想碰 backend 时用 apply()（同步执行，不投队列）：

  add.apply(args=[2, 3]).get()      # 5

Windows 上的两个限制

• 超时不生效。 配了 CELERY_TASK_TIME_LIMIT = 30，实测跑一个 35 秒的任务：

  task_time_limit=30；跑一个 35 s 的 slow(35) 观察会不会被杀
  slow(35) -> 35（35.2 s）
  表里 slow(35) 的 status='SUCCESS' result='35'

  • 超时是靠给子进程发信号实现的，Windows 没有 SIGKILL/SIGALRM，所以 Celery 在 Windows 上不实现 hard/soft time limit（官方文档也写了这一点）。
  • 结论：需要「跑太久就杀」的能力时，别在 Windows 上跑 worker，交给 Linux（本机就是 WSL）或用外部监控。

• prefork 能用，但行为不同。 实测默认池在 Windows 上是 SpawnPoolWorker-*（靠 spawn 重新 import 模块），所以：

  • 任务模块的顶层代码会被反复执行，注册类操作要幂等（dispatch_uid、get_or_create）。
  • 子进程里 print 的内容出现在 worker 控制台，不在 Web 控制台 —— 调试时别找错地方。

新手卡点速查

现象	原因	处理
Unknown command: 'celery'	照抄了 manage.py celery worker（djcelery 的命令）	换成 celery -A proj worker
ImportError: cannot import name 'task'	Celery 5 删了 from celery import task	用 @shared_task / @app.task
ModuleNotFoundError: No module named 'packaging'	只装了 celery，没带 redis extra	pip install "celery[redis]"
OperationalError: Authentication required.	broker URL 没带 Redis 密码	redis://:password@127.0.0.1:6379/0
任务投出去但 worker 说没注册	任务文件不叫 tasks.py，或 CELERY_IMPORES 拼错	改名或写 CELERY_IMPORTS
state 永远是 PENDING	worker 没起 / 队列名不对 / 任务从没存在过	看 worker 横幅的 [queues]，redis-cli LLEN celery
get() 抛 TimeoutError	同上，任务还压在队列里	起 worker 或换正确的队列
结果表里 task_name、traceback 全空	默认不记扩展列	CELERY_RESULT_EXTENDED = True
date_started 比 date_done 晚 8 小时	那一列用的是数据库 NOW()，会话时区是 SYSTEM	关掉 CELERY_TASK_TRACK_STARTED 或忽略该列
任务跑不完也不被杀（Windows）	不支持 time limit	换 Linux/WSL 跑 worker
定时任务不执行	只起了 beat 没起 worker，或 args 不是 JSON 字符串	两个进程都要在；args='[1, 2]'
生产环境装 beat 时 Django 被降级	插件元数据写了 Django<6.1	装前看 Requires-Dist，或锁版本

参考

• Celery - Distributed Task Queue：官方文档首页，架构与用户指南的入口。
• Introducing Celery：Task / Broker / Worker / Backend 四个角色的官方定义。
• First steps with Celery：celery[redis] 的安装写法与最小示例。
• Configuring Celery — Using Django：celery.py + __init__.py + namespace="CELERY" 的官方模板。
• Calling tasks：countdown、eta、queue、expires 等参数。
• Combining work flows / Canvas：group、chord、chain 的编排语义。
• Periodic Tasks：beat 的定位、两种调度器与「只能跑一个 beat」的约束。
• django-celery-beat 与 django-celery-results：数据库版定时任务与结果后端。
• Concurrency — Celery user guide 与 Task time limits：官方对 Windows 不支持超时的说明。
• AMQP 0-9-1 规范：所谓「跨平台」的协议基础。