在 FastAPI + Uvicorn 多 worker 部署场景中，若将定时任务（如 APScheduler）置于 lifespan 中，每个 worker 会独立启动一份调度器，导致任务重复执行；本文提供一种基于父进程初始化调度器的可靠方案。
在 fastapi + uvicorn 多 worker 部署场景中，若将定时任务（如 apscheduler）置于 `lifespan` 中，每个 worker 会独立启动一份调度器，导致任务重复执行；本文提供一种基于父进程初始化调度器的可靠方案。
当使用 uvicorn.run(..., workers=N) 启动 FastAPI 应用时，Uvicorn 采用 prefork 模式：主进程（parent）先启动，再 fork 出多个子进程（workers）来处理请求。每个子进程都会完整执行 lifespan 逻辑——这意味着若你在 @asynccontextmanager 中初始化并启动 AsyncIOScheduler，N 个 worker 就会运行 N 个完全独立的调度器实例，造成定时任务被重复触发（如示例中每秒打印两次），违背业务预期。
根本解法是：将调度器生命周期与 Uvicorn 的进程模型对齐——在父进程（main thread）中初始化并启动调度器，而非在每个 worker 的 lifespan 中。此时只需一个 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()
# 注意：此处不定义 lifespan —— 调度器不由 FastAPI 生命周期管理
if __name__ == "__main__":
# ✅ 在父进程（fork 前）初始化调度器
scheduler = BackgroundScheduler()
scheduler.add_job(
test,
trigger="cron",
second="0-30" # 每半秒执行一次（用于演示）
)
scheduler.start()
print("Background scheduler started in parent process.")
try:
# 启动 Uvicorn：workers 将作为子进程 fork，不重复创建 scheduler
uvicorn.run("main:app", workers=2, host="127.0.0.1", port=5000)
finally:
# ✅ 父进程退出前优雅关闭调度器
scheduler.shutdown(wait=True)
print("Background scheduler shut down.")? 关键要点说明：
- 调度器类型选择：必须使用 BackgroundScheduler（非 AsyncIOScheduler），因其基于 threading.Timer，可在任意线程/进程中运行，不依赖 asyncio event loop，适合父进程上下文；
- 启动时机：scheduler.start() 必须在 uvicorn.run() 调用之前执行，确保其运行于 fork 前的父进程；
- 信号安全：Uvicorn 默认捕获 SIGINT/SIGTERM 并协调子进程退出，但父进程中的 BackgroundScheduler 需手动 shutdown()；建议包裹在 try/finally 中保障资源释放；
- 避免共享状态风险：切勿在 BackgroundScheduler 的 job 函数中直接操作 FastAPI 应用状态（如 app.state），因该对象仅存在于 worker 进程中；如需共享数据，请改用 Redis、数据库或文件锁等跨进程机制；
- 进阶替代方案：对于高可用场景，可考虑将定时任务完全剥离至独立服务（如 Celery Beat + Redis），由外部调度器触发 FastAPI 接口，实现彻底解耦与水平扩展。
通过将调度逻辑移出 lifespan、交由 Uvicorn 父进程统一管控，你既能保留多 worker 的并发优势，又能严格保证定时任务的全局唯一性——这是生产环境部署 FastAPI 定时任务的推荐范式。
