如何确保FastAPI lifespan代码仅在Uvicorn单个Worker中执行？

问题说明 #

在FastAPI里用lifespan管理APScheduler的启停，但开2个Uvicorn Worker时，每个Worker都会单独初始化调度器，导致test()每秒跑两次，不符合只执行一次的需求。

原代码：

from fastapi import FastAPI
from datetime import datetime
from contextlib import asynccontextmanager
from apscheduler.schedulers.asyncio import AsyncIOScheduler
import uvicorn

def test():
    print(f"Test scheduler {datetime.now()}")
    
@asynccontextmanager
async def lifespan(app: FastAPI):
    scheduler = AsyncIOScheduler() 
    
    scheduler.add_job(test, trigger="cron", second="0-30")
    scheduler.start()
        
    yield
    
    scheduler.shutdown()
    
app = FastAPI(lifespan=lifespan)

if __name__ == "__main__":
    uvicorn.run("main:app", workers=2)

实际输出（每秒两次执行）：

INFO:     Uvicorn running on http://127.0.0.1:5000 (Press CTRL+C to quit)
INFO:     Started parent process [34952]
INFO:     Started server process [34957]
INFO:     Waiting for application startup.
INFO:     Started server process [34958]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Application startup complete.
Test scheduler 2024-02-21 14:51:15.001191
Test scheduler 2024-02-21 14:51:15.001314
Test scheduler 2024-02-21 14:51:16.000480
Test scheduler 2024-02-21 14:51:16.001643
Test scheduler 2024-02-21 14:51:17.000901
Test scheduler 2024-02-21 14:51:17.002765
...

解决方法 #

1. 用文件锁确保仅一个进程启动调度器 #

通过文件锁机制，让多个Worker进程中只有一个能成功获取锁并启动调度器，其他进程跳过初始化步骤。

修改后的lifespan代码：

from fastapi import FastAPI
from datetime import datetime
from contextlib import asynccontextmanager
from apscheduler.schedulers.asyncio import AsyncIOScheduler
import uvicorn
import os
import fcntl

def test():
    print(f"Test scheduler {datetime.now()}")
    
@asynccontextmanager
async def lifespan(app: FastAPI):
    scheduler = None
    lock_file = open("scheduler.lock", "w")
    try:
        # 尝试获取排他锁，非阻塞模式
        fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
        scheduler = AsyncIOScheduler() 
        scheduler.add_job(test, trigger="cron", second="0-30")
        scheduler.start()
        yield
    except BlockingIOError:
        # 获取锁失败，说明已有进程启动了调度器
        yield
    finally:
        if scheduler:
            scheduler.shutdown()
        lock_file.close()
        # 可选：删除锁文件，避免下次启动残留
        try:
            os.remove("scheduler.lock")
        except OSError:
            pass
    
app = FastAPI(lifespan=lifespan)

if __name__ == "__main__":
    uvicorn.run("main:app", workers=2)

2. 使用APScheduler分布式存储后端 #

借助Redis、MySQL等外部存储作为APScheduler的任务存储，让多个Worker的调度器共享任务状态，确保任务只执行一次。

首先安装依赖：

pip install apscheduler[redis]

修改后的代码：

from fastapi import FastAPI
from datetime import datetime
from contextlib import asynccontextmanager
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.redis import RedisJobStore
from apscheduler.executors.asyncio import AsyncIOExecutor
import uvicorn

def test():
    print(f"Test scheduler {datetime.now()}")
    
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 配置Redis作为任务存储
    jobstores = {
        'default': RedisJobStore(host='localhost', port=6379)
    }
    executors = {
        'default': AsyncIOExecutor()
    }
    scheduler = AsyncIOScheduler(jobstores=jobstores, executors=executors)
    # 设置replace_existing=True，确保重复启动时不会添加重复任务
    scheduler.add_job(test, trigger="cron", second="0-30", id="test_task", replace_existing=True)
    scheduler.start()
    yield
    scheduler.shutdown()
    
app = FastAPI(lifespan=lifespan)

if __name__ == "__main__":
    uvicorn.run("main:app", workers=2)

3. 定时任务与FastAPI独立部署 #

把定时任务单独抽成一个进程运行，和FastAPI服务完全分离，彻底避免Worker进程的干扰。

创建单独的定时任务脚本scheduler.py：

from datetime import datetime
from apscheduler.schedulers.asyncio import AsyncIOScheduler

def test():
    print(f"Test scheduler {datetime.now()}")

if __name__ == "__main__":
    scheduler = AsyncIOScheduler()
    scheduler.add_job(test, trigger="cron", second="0-30")
    scheduler.start()
    # 保持进程运行
    import asyncio
    asyncio.get_event_loop().run_forever()

然后分别启动服务：

# 启动FastAPI多Worker服务
uvicorn main:app --workers=2
# 启动定时任务进程
python scheduler.py

内容的提问来源于stack exchange，提问作者kyktyj