用途
Background Job Patterns 用于在 Python 后端中卸载长时间运行的任务，通过异步作业队列处理。适用场景包括：长时间运行的任务（报告生成、数据处理）、邮件/通知发送、定时/周期任务、Webhook 处理、数据导入/导出管道，以及非 LLM 的异步操作（LLM 工作流建议使用 LangGraph）。
工具选型
技能提供以下工具对比：
- ARQ：Python（异步），适合 FastAPI、简单任务，复杂度低。
- Celery：Python，适合复杂工作流、企业级场景，复杂度高。
- RQ：Python，适合简单 Redis 队列，复杂度低。
- Dramatiq：Python，适合可靠消息传递，复杂度中等。
ARQ（Async Redis Queue）
配置
在 backend/app/workers/arq_worker.py 中定义 worker：通过 create_pool 与 RedisSettings 创建连接；startup 初始化 worker 资源（数据库连接池、httpx 异步客户端），shutdown 清理资源。WorkerSettings 中设置 redis_settings、functions 列表（如 send_email、generate_report、process_webhook）、on_startup、on_shutdown、max_jobs（示例为 10）、job_timeout（示例为 300 秒）。
任务定义
使用 from arq import func。任务为异步函数，接收 ctx 字典与业务参数。示例 send_email 通过 httpx 向 SendGrid API 发送邮件并返回状态；generate_report 从数据库获取报告数据、渲染 PDF、保存文件并返回报告 ID 与大小。
从 FastAPI 入队
通过依赖 get_arq_pool 创建连接池。在路由中调用 arq.enqueue_job("generate_report", report.id, format=data.format) 入队，返回 report_id 与 job_id。查询作业状态时使用 Job(job_id, arq)，调用 job.status() 与 job.result()（仅在完成时获取结果）。
Celery（企业级）
配置
在 backend/app/workers/celery_app.py 中创建 Celery 实例，broker 与 backend 均使用 Redis。配置项包括：JSON 序列化、UTC 时区、task_track_started=True、task_time_limit=600（硬限制 10 分钟）、task_soft_time_limit=540（软限制 9 分钟）、worker_prefetch_multiplier=1（公平分发）、task_acks_late=True（完成后确认）、task_reject_on_worker_lost=True。
任务定义
使用 @shared_task 装饰器。示例 send_email 设置 bind=True、max_retries=3、default_retry_delay=60、autoretry_for=(ConnectionError, TimeoutError)，失败时调用 self.retry(exc=exc)。generate_report 通过 self.update_state(state="PROGRESS", meta={"step": ...}) 上报进度（fetching、rendering、saving）。
链与组
- 顺序执行：
chain(extract_data.s(source_id), transform_data.s(), load_data.s(destination_id))。 - 并行执行：
group(process_chunk.s(chunk) for chunk in chunks)。 - 并行加回调：
chord([process_chunk.s(chunk) for chunk in chunks], aggregate_results.s())。
周期任务（Celery Beat）
通过 celery_app.conf.beat_schedule 配置，使用 crontab 或秒数。示例：每 6 小时清理过期会话（crontab(minute=0, hour="*/6")）、每天凌晨 2 点生成日报（crontab(minute=0, hour=2)）、每 5 分钟同步外部数据（300.0）。
FastAPI 集成
简单进程内后台任务可使用 BackgroundTasks，通过 background_tasks.add_task(send_welcome_email, user.email) 添加。分布式任务应使用 ARQ/Celery，例如 arq.enqueue_job("export_data", data.dict())。
作业状态跟踪
定义 JobStatus 枚举：PENDING、STARTED、PROGRESS、SUCCESS、FAILURE、REVOKED。查询 Celery 作业时使用 AsyncResult(job_id, app=celery_app)，返回 status、result（仅在 ready 时）与 progress（状态为 PROGRESS 时）。
反模式（禁止）
- 禁止同步运行长任务（会阻塞数分钟）。
- 禁止在失败时丢失作业（无重试、无错误处理）。
- 禁止在 Redis 中存储大结果（应存 S3/DB）。
- 禁止用 BackgroundTasks 处理分布式工作（服务器重启会丢失）。
关键决策
- 简单异步：ARQ（原生异步）。
- 复杂工作流：Celery（chains、chords）。
- 进程内快速任务：FastAPI BackgroundTasks。
- LLM 工作流：LangGraph（非 Celery）。
- 结果存储：Redis 存状态，S3/DB 存数据。
- 重试策略：带抖动的指数退避。
相关技能
langgraph-checkpoints（LLM 工作流持久化）、resilience-patterns（重试与回退）、observability-monitoring（作业指标）。
能力细节
- arq-tasks：关键词 arq、async queue、redis queue、background task；解决 FastAPI 中如何运行异步后台任务、简单 Redis 作业队列。
- celery-tasks：关键词 celery、task queue、distributed tasks、worker；解决企业级任务队列、复杂作业工作流。
- celery-workflows：关键词 chain、group、chord、celery workflow；解决顺序任务执行、并行任务处理。
- periodic-tasks：关键词 periodic、scheduled、cron、celery beat；解决按计划运行任务、类 cron 作业调度。