SuperAPI AI 工具集

python-background-jobs

编程与开发

查看项目

更新于 10-15 21:38

技能介绍

用途:把长时间运行或不可靠的工作从请求/响应周期中解耦,让 API 立即返回,由后台 worker 异步处理。适用场景包括:耗时超过数秒的任务、发送邮件/通知/webhook、生成报表或导出数据、处理上传或媒体转换、集成不可靠外部服务、构建事件驱动架构。

核心概念:任务队列模式(API 接收请求、入队作业、立即返回作业 ID,worker 异步处理);幂等性(任务可能被重试,需设计为可安全重复执行);作业状态机(pending → running → succeeded/failed);至少一次投递(多数队列保证至少一次,代码需处理重复)。

快速开始示例使用 Celery(也提到 RQ、Dramatiq 以及 AWS SQS、GCP Tasks 等云方案同样可行): from celery import Celery app = Celery("tasks", broker="redis://localhost:6379") @app.task def send_email(to: str, subject: str, body: str) -> None: email_client.send(to, subject, body) send_email.delay("user@example.com", "Welcome!", "Thanks for signing up")

基本模式:一是立即返回作业 ID,对超过数秒的操作返回 job_id 并异步处理;二是 Celery 任务配置,设置 task_time_limit、task_soft_time_limit、task_acks_late、task_reject_on_worker_lost、worker_prefetch_multiplier,并用 max_retries、default_retry_delay、autoretry_for 处理瞬时错误;三是让任务幂等,先检查状态再执行、使用幂等键、upsert、去重窗口;四是作业状态管理,持久化状态转换便于轮询和调试。

最佳实践:立即返回、持久化作业状态、任务幂等、对外部调用使用幂等键、设置软硬超时、实现死信队列捕获永久失败任务、记录状态转换日志、对瞬时错误采用指数退避重试、不重试永久性失败(如校验错误、无效凭证)、监控队列深度并在积压增长时告警。更详细示例见 references/details.md。具体安装命令与版本信息暂未提供。