async-jobs

异步任务

使用 Celery、ARQ、Redis 和 Temporal 处理后台任务。此技能是一个封装，而非手册：Celery 和 ARQ 已有完善的产品文档，因此这里仅介绍我们的增补内容，包括我们选定的阈值、可用配置、顺序约束和工具选择规则。产品机制通过链接引用，不再重复说明。

请先阅读 Read("references/ork-delta.md")。

快速参考

主题	我们的相关内容
配置	references/celery-config.md、rules/jobs-task-queue.md
任务路由	references/ork-delta.md（队列分类、预取层级、Redis 优先级）
Canvas 工作流	rules/celery-canvas.md
重试策略	references/ork-delta.md（退避上限、幂等层、锁 TTL）
调度	rules/jobs-scheduling.md、references/ork-delta.md（beat 进程模型）
监控	references/ork-delta.md（告警阈值、直方图桶）
结果后端	rules/jobs-monitoring.md、references/ork-delta.md（返回约定）
ARQ 模式	rules/jobs-task-queue.md、references/ork-delta.md（预算、池所有权）
Temporal 工作流	rules/temporal-workflows.md
Temporal 活动	rules/temporal-activities.md

10 个主题领域，rules/ 中有 6 个规则文件，references/ork-delta.md 中包含我们的增补内容。

快速入门

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def process_payment(self, order_id: str):
    try:
        return gateway.charge(order_id)
    except TransientError as exc:
        raise self.retry(exc=exc, countdown=2 ** self.request.retries * 60)

如需查看更多示例，请阅读 Read("references/quick-start-examples.md")，了解 Celery 重试任务以及 ARQ/FastAPI 集成模式。

上游覆盖范围（无需重复说明）

需要了解产品机制时，请查阅这些资料。我们保留右侧列的内容，因为这些内容属于上游无法了解的内部门槛、实际配置或排序约束。

主题	第一方资料	保留的内部子集位于
Celery 设置、序列化器、时间限制、worker 标志	https://docs.celeryq.dev/en/stable/userguide/configuration.html and .../optimizing.html	references/celery-config.md、rules/jobs-task-queue.md
队列声明、路由器类、Redis 优先级机制	https://docs.celeryq.dev/en/stable/userguide/routing.html	references/ork-delta.md
chain / group / chord / signature 语义	https://docs.celeryq.dev/en/stable/userguide/canvas.html	rules/celery-canvas.md 保留内部 canvas 子集。其关于在 chord 中使用 si() 的指导尚未验证且存在争议：依赖此指导前，请先对照上游 canvas 页面确认参数传递行为
autoretry_for、retry_backoff、Reject、任务基类	https://docs.celeryq.dev/en/stable/userguide/tasks.html	references/ork-delta.md、rules/jobs-task-queue.md
Beat 调度、crontab 语法、DatabaseScheduler	https://docs.celeryq.dev/en/stable/userguide/periodic-tasks.html and https://django-celery-beat.readthedocs.io/en/latest/	rules/jobs-scheduling.md 保留我们的 beat_schedule 结构；references/ork-delta.md 保留进程模型
Flower 标志、inspect、信号名称	https://docs.celeryq.dev/en/stable/userguide/monitoring.html, https://docs.celeryq.dev/en/stable/userguide/signals.html, https://flower.readthedocs.io/en/latest/config.html	references/ork-delta.md
结果后端、AsyncResult、自定义状态	https://docs.celeryq.dev/en/stable/userguide/configuration.html	rules/jobs-monitoring.md 保留我们的状态端点和 update_state() 用法
每个任务的 rate_limit、control.rate_limit、Redis Lua	https://docs.celeryq.dev/en/stable/userguide/workers.html, https://redis.io/docs/latest/develop/programmability/eval-intro/	references/ork-delta.md
ARQ WorkerSettings、enqueue_job、_defer_by / _defer_until、Job 状态	https://arq-docs.helpmanual.io/	rules/jobs-task-queue.md 保留 worker 骨架；references/ork-delta.md 保留预算
FastAPI lifespan 和依赖项接线	https://fastapi.tiangolo.com/advanced/events/	references/ork-delta.md
使用 SET NX EX 实现分布式锁	https://redis.io/docs/latest/commands/set/	references/ork-delta.md

配置

加载：Read("references/celery-config.md")。

决策	建议
序列化器	JSON（绝不使用 pickle）
确认模式	延迟确认（task_acks_late=True）
预取	公平调度时设为 1，追求吞吐量时设为 4-8
时间限制	软限制 < 硬限制（540 / 600）
时区	始终使用 UTC

任务路由

决策	建议
队列数量	5 个：critical / high / default / low / bulk
优先级级别	0-9，四个 Redis 优先级开关必须一起设置
Worker 分配	每个队列分配专用 worker
预取	critical 为 1，high 为 2，default 为 4，low/bulk 为 8
路由	路由规则超过 5 条后使用 Router 类

Canvas 工作流

加载：Read("rules/celery-canvas.md")。

决策	建议
顺序执行	使用 s() 创建 Chain
并行执行	独立任务使用 Group
汇总	使用 Chord（所有 header 任务都必须成功，body 才会运行）
忽略输入	使用 si() 不可变签名
Chain 中出错	Reject 会停止 Chain，retry 会继续执行
部分失败	Chord header 任务返回错误字典

重试策略

决策	建议
重试延迟	指数退避，启用抖动，上限为 600s
最大重试次数	临时性错误重试 3-5 次，永久性错误重试 0 次
幂等性	使用 Redis 标记（TTL 为 86400s）和供应商幂等键
失败任务	使用 DLQ 供人工审核
单例	使用 TTL 长于硬性时限的 Redis 锁

调度

加载：Read("rules/jobs-scheduling.md")。

决策	建议
调度类型	基于时间的任务使用 Crontab，按频率执行的任务使用浮点间隔
动态调度	DatabaseScheduler（django-celery-beat）
重叠执行	使用 Redis 锁，默认 3600s，长任务使用 7200s
Beat 进程	使用独立进程；嵌入式 --beat 仅用于开发
时区	始终使用 UTC

监控

决策	建议
仪表板	使用持久化存储的 Flower
指标	将 Prometheus 接入 task_prerun / task_postrun / task_failure
健康状况	检查 Broker 是否可访问、是否至少有一个 Worker，以及队列深度
告警	critical > 100，default > 5000，Worker < 1
自动扩缩容	队列深度 > 500

结果后端

加载：Read("rules/jobs-monitoring.md")。

决策	建议
状态存储	使用 Redis 结果后端，仅存储状态和小型 JSON
大型结果	存储在 S3 或数据库中，任务返回引用字典
进度	使用 update_state() 设置自定义状态
结果查询	使用 AsyncResult 并检查状态

ARQ 模式

加载：Read("rules/jobs-task-queue.md")。

决策	建议
简单异步任务	使用 ARQ（原生异步），设置 max_jobs=10、job_timeout=300
Pool 所有权	由 FastAPI 生命周期管理，切勿按请求调用 create_pool
复杂工作流	使用 Celery（chains、chords、DLQ、每个任务的速率限制）
进程内快速任务	使用 FastAPI BackgroundTasks，耗时低于 30s，且仅用于非关键任务
大模型工作流	使用 LangGraph，而非 Celery

工具选择

如需查看 ARQ、Celery、RQ、Dramatiq 和 FastAPI BackgroundTasks 的完整工具对比表，请加载：Read("references/quick-start-examples.md")。

反模式（禁止）

加载详细内容：Read("references/anti-patterns.md")，查看完整列表。

关键规则：绝不在请求处理程序中运行长任务，绝不在任务内部阻塞等待结果，绝不在 Redis 中存储大型结果，重试任务始终使用幂等性。

Temporal 工作流

加载：Read("rules/temporal-workflows.md")。

决策	建议
工作流 ID	使用有业务含义且具备幂等性的 ID
确定性	使用 workflow.random()、workflow.now()
I/O	始终通过活动执行，绝不直接执行

Temporal 活动

加载：Read("rules/temporal-activities.md")。

决策	建议
活动超时	大多数情况下使用 start_to_close
错误处理	对业务错误设置为不可重试
测试	集成测试使用 WorkflowEnvironment.start_local()

相关技能

• ork:python-backend - FastAPI、asyncio、SQLAlchemy 模式
• ork:langgraph - LangGraph 工作流模式（用于大模型工作流，而非 Celery）
• ork:distributed-systems - 弹性模式、熔断器
• ork:monitoring-observability - 指标和告警

功能详情

加载详细内容：Read("references/capability-details.md")，查看关键词索引 和问题到功能的映射。