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() 等方法，但要注意参数序列化和权限控制：
立即学习“Python免费学习笔记（深入）”；
- 触发器类型必须明确传字符串，如
'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 也未自动创建。
