fastapi本身不提供定时调度能力,backgroundtasks仅支持单次、请求绑定的异步执行,无法持久化、重启即失效,且多进程下易重复执行;正确方案是集成apscheduler,通过lifespan事件管理生命周期,并使用外部存储(如sqlite/redis)确保任务一致性。

FastAPI 本身不提供脚本调度能力,它只负责 HTTP 请求响应;真正的调度必须交由外部任务系统承担。直接在 BackgroundTasks 里塞循环或 time.sleep() 是错的——它会阻塞事件循环、泄漏线程、无法持久化、重启即失效。
为什么不能用 BackgroundTasks 做定时调度
BackgroundTasks 是单次、请求绑定、无状态的异步执行机制,设计初衷是“响应后顺手干点事”,不是“每隔 5 分钟跑一次”。常见误用包括:
- 在
BackgroundTasks.add_task()中调用while True: do_something(); time.sleep(300)—— 这会卡死一个 worker 线程,Uvicorn 不会回收它 - 把 APScheduler 的
scheduler.start()放进路由函数里 —— 每次请求都启一个新调度器,进程内堆叠多个 scheduler 实例,内存和 timer 句柄爆炸 - 依赖全局变量保存 scheduler 实例但没做单例保护 —— 多 worker(
--workers 4)下每个进程都起一份,任务重复执行 4 次
APScheduler + FastAPI 的安全启动方式
必须在应用生命周期外初始化调度器,并确保它只启动一次、与 Uvicorn 生命周期对齐。推荐做法:
- 用
lifespan事件管理:在startup中初始化并启动BackgroundScheduler,在shutdown中显式.shutdown() - 避免使用
AsyncIOScheduler:FastAPI 的 event loop 已被 Uvicorn 占用,APScheduler 的 async 版本会冲突,坚持用BackgroundScheduler - 存储 job 用
SQLAlchemyJobStore或RedisJobStore,别用默认内存存储 —— 否则重启就丢所有定时任务 - 示例片段:
from apscheduler.schedulers.background import BackgroundScheduler<br>from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore<br><br>jobstores = {<br> "default": SQLAlchemyJobStore(url="sqlite:///jobs.sqlite")<br>}<br>scheduler = BackgroundScheduler(jobstores=jobstores)<br><br>@app.on_event("startup")<br>async def start_scheduler():<br> scheduler.start()<br><br>@app.on_event("shutdown")<br>async def stop_scheduler():<br> scheduler.shutdown(wait=False)
如何通过 API 动态增删定时任务
APScheduler 提供 .add_job()、.remove_job()、.get_job() 等方法,但要注意参数序列化和权限控制:
- 触发器类型必须明确传字符串,如
'interval'、'cron'、'date',不能传实例(无法跨进程 pickle) - 函数名必须是模块内可导入路径,例如
'myapp.tasks.cleanup_db',不能传 lambda 或闭包 - 任务参数只支持 JSON 序列化类型(
str、int、dict、list),datetime要转 ISO 格式字符串 - 务必校验 job_id 是否已存在,重复添加会报
ConflictingIdError;删除前先try/except JobLookupError - 示例添加接口:
@app.post("/jobs")<br>def create_job(<br> job_id: str,<br> func: str, # 如 "myapp.tasks.send_report"<br> trigger: str = "interval",<br> minutes: int | None = None,<br> args: list | None = None,<br>):<br> try:<br> scheduler.add_job(<br> func,<br> trigger,<br> id=job_id,<br> minutes=minutes,<br> args=args or [],<br> )<br> return {"status": "ok"}<br> except Exception as e:<br> raise HTTPException(400, str(e))
多进程部署时的调度一致性问题
Uvicorn 默认用 --workers N 启多个进程,每个进程都有自己的 scheduler 实例,会导致同一任务被多次触发。解决路径只有两条:
- 禁用多 worker,改用单进程 + 多线程(
--workers 1 --threads 4),靠BackgroundScheduler内部线程池调度 —— 适合中小负载 - 用分布式调度器替代 APScheduler,例如
celery beat+ Redis/RabbitMQ,让 beat 进程统一发任务,worker 进程只消费 —— 生产环境唯一可靠方案 - 若坚持用 APScheduler,必须启用外部 job store(如 Redis),并配合
coalesce=True和max_instances=1防重入,但这不能 100% 避免竞态,仅降低概率
最易被忽略的一点:所有调度函数必须是纯函数或模块级函数,不能依赖 FastAPI 的 request state、Depends 注入对象或未初始化的全局连接池——因为它们运行在独立线程中,没有 request scope,DB session 也未自动创建。
大量免费API接口:立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游,为数据API应用能力赋能!











