FastAPI多Worker环境下如何避免APScheduler任务重复执行？

当使用Uvicorn启动多Worker时，每个Worker都会独立加载FastAPI应用代码，导致APScheduler被多次初始化并启动，任务重复执行。以下是基于Python标准库的最简解决方案，无需额外依赖第三方模块：

实现思路 #

通过跨平台文件锁实现单实例调度器：只有第一个成功获取文件锁的Worker进程会启动APScheduler，其他进程因无法获取锁而跳过调度器启动步骤，从而保证任务仅在单个进程中执行。

修改后的代码 #

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import FastAPI
import os
import sys
from contextlib import contextmanager

@contextmanager
def single_instance_lock(lock_file="/tmp/apscheduler_single_instance.lock"):
    lock_fd = None
    try:
        # 创建/打开锁文件，仅写模式，不存在则创建
        lock_fd = os.open(lock_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o644)
        
        # 尝试获取排他锁（非阻塞）
        if sys.platform == 'win32':
            import msvcrt
            msvcrt.locking(lock_fd, msvcrt.LK_NBLCK, 1)
        else:
            import fcntl
            fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        
        yield
    except (BlockingIOError, PermissionError):
        # 无法获取锁，说明已有进程启动调度器，直接跳过
        pass
    finally:
        if lock_fd is not None:
            os.close(lock_fd)

def some_job():
    print(f"hello world from process {os.getpid()}")

app = FastAPI()
shed = AsyncIOScheduler()
shed.add_job(some_job, "interval", seconds=5)

@app.get("/test")
async def test():
    return {"Hello": "world"}

# 仅在获取到锁的进程中启动调度器
with single_instance_lock():
    shed.start()

代码说明 #

1. 文件锁上下文管理器：single_instance_lock函数实现了跨平台的非阻塞排他锁，保证同一时间只有一个进程能获取锁。
2. 调度器启动控制：用with语句包裹shed.start()，只有成功获取锁的进程会执行调度器启动逻辑，其他进程自动跳过。
3. 自动锁释放：当进程退出时，文件描述符会自动关闭，锁也随之释放，避免遗留无效锁。

验证效果 #

启动多Worker后执行：

uvicorn main:app --workers 4

此时任务仅会每隔5秒打印一次hello world from process [PID]，而非4次，解决了重复执行问题。

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