本文介绍在 fastapi + uvicorn 多工作进程（workers > 1）部署场景下，避免生命周期（lifespan）中启动的定时任务被重复执行的核心方案：将调度器移至主进程初始化，并选用 backgroundscheduler 替代 asyncioscheduler。
本文介绍在 fastapi + uvicorn 多工作进程（workers > 1）部署场景下，避免生命周期（lifespan）中启动的定时任务被重复执行的核心方案：将调度器移至主进程初始化，并选用 backgroundscheduler 替代 asyncioscheduler。
在使用 Uvicorn 以多进程模式（如 workers=2）运行 FastAPI 应用时，每个工作进程都会独立执行 lifespan 事件——这意味着若在 lifespan 中初始化并启动 AsyncIOScheduler，所有工作进程都会各自运行一套调度器，导致定时任务（如每秒执行一次的 test()）被多次并发触发，违背单例预期。
根本原因在于：Uvicorn 的多进程模型采用 fork 派生（Unix）或 spawn 启动（Windows），每个 worker 是完全独立的 Python 进程，拥有各自的事件循环、内存空间和调度器实例。lifespan 在每个 worker 内部被单独调用，因此无法天然实现“仅一个进程执行”的语义。
✅ 正确解法是：将调度器脱离 FastAPI 生命周期管理，提升至主进程（parent process）中初始化和启停。这确保全局仅存在一个调度器实例，且其生命周期与 Uvicorn 主进程对齐。
推荐使用 APScheduler 的 BackgroundScheduler（非异步版），因其基于线程而非事件循环，更适合作为主进程中的后台守护服务：
from fastapi import FastAPI
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
import uvicorn
def test():
print(f"Test scheduler {datetime.now()}")
app = FastAPI()
if __name__ == "__main__":
# ✅ 在主进程（parent）中创建并启动调度器
scheduler = BackgroundScheduler()
scheduler.add_job(
test,
trigger="cron",
second="0-30", # 每分钟第 0–30 秒各触发一次（即每秒一次）
coalesce=True, # 防止因延迟导致的任务堆积
max_instances=1 # 确保同一任务不会并发执行
)
scheduler.start()
# ⚠️ 注意：必须在 uvicorn.run() 之后、程序退出前显式 shutdown
try:
uvicorn.run("main:app", workers=2, host="127.0.0.1", port=5000)
finally:
scheduler.shutdown(wait=True) # 安全等待正在运行的任务完成? 关键要点说明：
- 不使用 lifespan：避免在每个 worker 中重复初始化调度器；
- BackgroundScheduler vs AsyncIOScheduler：前者基于 threading.Timer，可安全运行于主线程；后者依赖 asyncio 事件循环，而主进程无活跃 event loop，且多 worker 下仍会重复；
- coalesce=True 和 max_instances=1：防止因系统负载导致任务堆积或并发执行，强化单次语义；
- scheduler.shutdown(wait=True)：确保应用退出前完成清理，避免僵尸线程或资源泄漏；
- 信号处理补充（进阶）：如需优雅响应 SIGTERM（例如 Docker 停止），可结合 signal 模块注册钩子，但上述 try/finally 已覆盖绝大多数场景。
? 补充建议：若业务逻辑涉及数据库写入、文件操作或外部 API 调用等有状态行为，还需考虑分布式锁（如 Redis Lock）作为兜底方案，但对纯定时触发、无共享状态的轻量任务，主进程单例调度器已足够可靠。
综上，通过将调度器上提至主进程并选用线程安全的 BackgroundScheduler，即可在保持 Uvicorn 多进程高并发能力的同时，严格保障定时任务的全局唯一性与执行确定性。
