FastAPI 本身不支持定时任务,必须依赖外部调度器如 APScheduler;应使用 BackgroundScheduler 绑定 lifespan 生命周期管理,配合外置 JobStore(如 Redis)和分布式锁确保多 worker 下任务唯一执行。

FastAPI 本身不支持定时任务,所谓“自动化任务触发平台”必须靠外部调度器实现——直接用 BackgroundTasks 或裸写 while True + time.sleep() 是生产环境雷区,不是快,是埋得快。
为什么 BackgroundTasks 不能当定时器用
BackgroundTasks 的设计目标是“请求响应后顺手干点事”,它没有状态、不记时间、不持久、不跨请求。常见错误包括:
- 在
add_task()里塞while True循环 —— 卡死 Uvicorn worker 线程,无法回收 - 把
scheduler.start()放进路由函数 —— 每次调 API 都启一个新调度器,内存泄漏+定时任务爆炸式叠加 - 用全局变量存
scheduler但没做进程隔离 —— 多 worker(如--workers 4)下任务重复执行 4 次
APScheduler 必须用 BackgroundScheduler + lifespan
别碰 AsyncIOScheduler:Uvicorn 已独占 event loop,强行用它会冲突报错 RuntimeError: This event loop is already running。正确姿势是:
- 用
BackgroundScheduler(线程安全,不抢 loop) - 初始化和启停必须绑定
lifespan,不是@app.on_event("startup")—— 后者在多 worker 下每个进程都执行一遍,而lifespan是 Uvicorn 级别生命周期管理,更可靠 - job store 必须外置:默认内存存储(
MemoryJobStore)重启即丢任务;SQLite 或 Redis 才算真正“自动化”
示例初始化片段:
立即学习“Python免费学习笔记(深入)”;
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
<p>jobstores = {"default": SQLAlchemyJobStore(url="sqlite:///jobs.sqlite")}
scheduler = BackgroundScheduler(jobstores=jobstores)</p><p>@asynccontextmanager
async def lifespan(app: FastAPI):
scheduler.start()
yield
scheduler.shutdown(wait=False)动态增删任务必须带 id 且校验重复
API 接口增删任务时,不设 id 或忽略 replace_existing=True,会导致同一逻辑被反复注册,最终每分钟跑 5 次、10 次……直到 CPU 告急。
- 添加任务务必指定
id字符串,比如id="daily_report" - 用
replace_existing=True避免重复注册 - 删除前先
scheduler.get_job("daily_report")判空,防止JobLookupError - 任务函数推荐写成异步(
async def),但 APScheduler 线程池默认不 await,需包装:await asyncio.to_thread(job_func)(Python 3.9+)
多 worker 部署时唯一执行的关键是存储+锁
Uvicorn 启多个进程(--workers 3)时,光靠 SQLite 不足以防重复——它没分布式锁。真实场景要:
- 用 RedisJobStore 替代 SQLite,配合
redis-py和 Redis 实例 - 启用 APScheduler 的
coalesce=True(合并错过的执行)、max_instances=1(单任务最多 1 实例) - 更稳的做法:加一层 Redis 分布式锁,比如用
redis.set("task:report:lock", "1", ex=300, nx=True),失败则跳过
这一步最容易被跳过,结果就是凌晨三点收到 3 条一模一样的告警邮件。


















