用途 本技能提供在 Django 项目中使用 Celery 处理后台任务的“生产级”模式，消息代理支持 Redis 或 RabbitMQ。适用场景包括：为 Django 应用添加后台作业或异步处理；实现周期性/计划任务；把慢操作（邮件、PDF 生成、API 调用）从请求周期中卸载；使用 Celery Beat 做类 cron 调度；调试任务失败、重试或队列积压；为 Celery 任务编写测试。
安装与项目设置 安装命令：pip install 'celery[redis]' django-celery-results django-celery-beat
应用入口配置在 config/celery.py：设置 DJANGO_SETTINGS_MODULE，创建 Celery 实例（示例项目名 myproject），通过 config_from_object('django.conf:settings', namespace='CELERY') 读取设置，并用 autodiscover_tasks() 自动发现各 INSTALLED_APP 中的 tasks.py；还定义了一个绑定任务 debug_task（ignore_result=True）。在 config/init.py 中导入并暴露 celery_app。
Django 设置要点（config/settings/base.py）：消息代理 CELERY_BROKER_URL 默认 redis://localhost:6379/0；结果后端默认 django-db；序列化仅接受 json；CELERY_TASK_TRACK_STARTED 开启；硬超时 30 分钟、软超时 25 分钟；CELERY_WORKER_PREFETCH_MULTIPLIER 设为 1 防止工作进程囤积长任务；CELERY_TASK_ACKS_LATE 开启以便工作进程崩溃后重新入队；结果保留 24 小时；Beat 调度器使用 django_celery_beat.schedulers:DatabaseScheduler；把 django_celery_results、django_celery_beat 加入 INSTALLED_APPS。
启动工作进程 开发环境：celery -A config worker --loglevel=info；启动 Beat：celery -A config beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler；开发时可 worker --beat 合并运行（文档明确生产环境绝不可这样）。生产环境示例为带并发和队列：celery -A config worker --loglevel=warning --concurrency=4 -Q default,high_priority。
任务设计模式 基本任务用 @shared_task 并指定 name，任务内延迟导入模型和服务，对 User.DoesNotExist 只记录警告并返回，保证幂等。可重试任务通过 bind=True、max_retries=5、default_retry_delay=60、autoretry_for=(ConnectionError, TimeoutError)、retry_backoff=True、retry_backoff_max=600、retry_jitter=True 配置，并在限流错误时用 self.retry(exc=exc, countdown=int(exc.retry_after)) 指定延迟。幂等模式用带状态筛选的 update，仅当状态为 PROCESSING 时才更新为 SHIPPED，未更新则记录日志。带软超时的任务设置 soft_time_limit=120、time_limit=150，捕获 SoftTimeLimitExceeded 后先清理部分文件再抛出。
调用方式：.delay() 即发即忘；apply_async 配合 countdown 或 eta 延迟执行；apply_async 传 queue 做路由；.apply() 仅用于测试或调试同步执行。
Beat 调度 代码定义用 CELERY_BEAT_SCHEDULE 配合 crontab 或秒数，示例含每日 2 点、每 60 秒、每周一 8 点。数据库定义使用 django_celery_beat 的 CrontabSchedule 与 PeriodicTask（update_or_create），可在 Django admin 或代码中管理。
Canvas 编排 chain 顺序执行并传递结果；group 并行执行；chord 并行后由回调聚合结果，均通过 .delay() 触发。
错误处理与死信队列 通过 @task_failure.connect 连接信号，把所有任务失败上报 Sentry（附任务名、task_id、args、kwargs）。任务内达到 max_retries 后写入 FailedCharge 死信表并返回（不抛异常），否则 raise self.retry(exc=exc)。
测试 单元测试不依赖 broker：用 pytest、patch 服务层，断言调用及缺失用户不抛异常。集成测试在 test 设置里开启 CELERY_TASK_ALWAYS_EAGER=True 和 CELERY_TASK_EAGER_PROPAGATES=True，使任务同步执行并重新抛出异常。重试测试用 apply(args=[1], throw=True) 捕获 ConnectionError，并断言首次尝试调用次数（eager 下为 1）。
监控 celery -A config inspect active / stats / reserved；Redis 用 redis-cli llen celery 查看队列长度；Flower 通过 pip install flower 后 celery -A config flower --port=5555 启动 Web 实时监控。
反模式 文档明确列出：不要传递模型实例，应传主键；不要在生产视图同步调用任务；非幂等任务必须加状态守卫，避免重试导致重复扣款或履约。
生产清单 工作进程崩溃重启用 supervisord 或 systemd；CELERY_TASK_ACKS_LATE=True 崩溃后重入队；CELERY_WORKER_PREFETCH_MULTIPLIER=1 公平分配长任务；按优先级分队列 -Q default,high_priority,low_priority；设置软超时；集成 Sentry 捕获 task_failure；使用 Flower 或类似监控查看队列深度；Beat 只在单节点运行以防重复调度。
相关技能 django-patterns、django-tdd、python-testing。