当前位置：首页 > news >正文

FastAPI + APScheduler + Uvicorn 多进程下避免重复加载任务的解决方案

news 2025/9/22 6:58:31

1. 问题背景

2. 原因分析

3. 常见解决方案

3.1 方案一：单 worker 模式（简单粗暴）

3.2 方案二：文件锁（单机可用，跨平台推荐）

示例代码（使用 filelock，兼容 Linux 和 Windows）

3.3 方案三：使用 Redis/数据库 JobStore（推荐）

示例代码（Redis JobStore）

3.4 方案四：独立 Scheduler 进程（解耦推荐）

示例

4. 方案对比

5. 总结与推荐

1. 问题背景

在 FastAPI 中集成 APScheduler，可以方便地执行定时任务。但当我们用 Uvicorn 启动应用时，如果指定了 --workers > 1，就会出现 每个 worker 进程都会初始化一份 APScheduler 调度器 的情况。

例如：

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

结果是同一个任务被 重复执行 4 次（每个 worker 都跑一遍），这在生产环境是无法接受的。

2. 原因分析

Uvicorn 的多进程模式通过 multiprocessing fork 出多个完全独立的 Python 进程。
每个进程都会加载一份 FastAPI 应用实例，执行 @app.on_event("startup") 钩子。
如果在 startup 里初始化 APScheduler，就会导致 每个进程都启动了自己的调度器。

换句话说：Uvicorn 没有 master/worker 角色区分，每个 worker 都是“主进程”。

3. 常见解决方案

3.1 方案一：单 worker 模式（简单粗暴）

只启动一个进程，避免重复加载：

uvicorn main:app --workers 1

适合任务量不大、并发不高的应用。但缺点是失去了多进程并发能力。

3.2 方案二：文件锁（单机可用，跨平台推荐）

在应用启动时尝试获取文件锁，只有第一个获取锁的进程会启动 APScheduler，其他进程跳过。

示例代码（使用 `filelock`，兼容 Linux 和 Windows）

import os
import logging
from fastapi import FastAPI
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from filelock import FileLock, Timeout

app = FastAPI()
LOCK_FILE = "scheduler.lock"

def my_job():print(f"[PID={os.getpid()}] running scheduled task...")

@app.on_event("startup")
async def startup_event():lock = FileLock(LOCK_FILE)try:lock.acquire(timeout=0.1)scheduler = AsyncIOScheduler()scheduler.add_job(my_job, "interval", seconds=10, id="unique_job", replace_existing=True)scheduler.start()
app.state.scheduler = schedulerapp.state.lock = locklogging.info(f"[PID={os.getpid()}] Scheduler started (lock acquired).")
except Timeout:logging.info(f"[PID={os.getpid()}] Another process holds the scheduler lock. Skipping.")

@app.on_event("shutdown")
async def shutdown_event():if hasattr(app.state, "scheduler"):app.state.scheduler.shutdown()if hasattr(app.state, "lock"):app.state.lock.release()logging.info(f"[PID={os.getpid()}] Scheduler stopped and lock released.")

🔎 特点：

适合 单机部署；
锁文件路径必须可写；
跨平台可用（Linux/Windows 都支持）。

3.3 方案三：使用 Redis/数据库 JobStore（推荐）

APScheduler 支持 Redis、MongoDB、SQLAlchemy 等作为 JobStore。即使多个进程启动调度器，它们共享相同的任务存储，不会重复执行。

示例代码（Redis JobStore）

from fastapi import FastAPI
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.redis import RedisJobStore

app = FastAPI()

def my_job():print("Running job...")

@app.on_event("startup")
async def startup_event():jobstores = {"default": RedisJobStore(host="localhost", port=6379, db=0)}scheduler = AsyncIOScheduler(jobstores=jobstores)scheduler.add_job(my_job, "interval", seconds=10, id="unique_job", replace_existing=True)scheduler.start()app.state.scheduler = scheduler

@app.on_event("shutdown")
async def shutdown_event():if hasattr(app.state, "scheduler"):app.state.scheduler.shutdown()

🔎 特点：

适合 多进程、多机集群 部署；
保证任务全局唯一；
需要 Redis 或数据库支持。

3.4 方案四：独立 Scheduler 进程（解耦推荐）

将 APScheduler 独立成单独的进程或服务，只负责任务调度；FastAPI 应用只处理 API 请求。

示例

# scheduler.py
import asyncio
from apscheduler.schedulers.asyncio import AsyncIOScheduler

def my_job():print("Task executed")

async def main():scheduler = AsyncIOScheduler()scheduler.add_job(my_job, "interval", seconds=10)scheduler.start()await asyncio.Event().wait()

if __name__ == "__main__":asyncio.run(main())

运行：

python scheduler.py
uvicorn main:app --workers 4

🔎 特点：

彻底避免重复加载；
解耦，适合中大型项目。

4. 方案对比

方案	优点	缺点
单 worker	简单稳定	无法利用多进程并发
文件锁	跨平台，代码简单，单机有效	多机部署无效
Redis/DB JobStore	多进程/多机可用，全局唯一	依赖额外存储
独立 scheduler 进程	彻底解耦，灵活扩展	需要额外部署进程