Uvicorn/FastAPI如何实现仅指定单个worker执行启动任务？

Uvicorn多worker部署模式下，每个worker是独立的子进程，默认startup事件会在所有worker进程启动时各触发一次，你可以通过以下两种方式控制仅单个worker执行目标任务：

方案1：通过UVICORN_WORKER_ID环境变量判断（推荐，实现最简单） #

Uvicorn 0.12.0及以上版本启动多worker时，会自动给每个worker进程注入UVICORN_WORKER_ID环境变量，取值从0开始按启动顺序递增，直接判断该值是否为0即可筛选出0号worker：

import os
import asyncio

@app.on_event('startup')
async def app_startup():
    worker_id = int(os.getenv("UVICORN_WORKER_ID", "0"))
    # 仅0号worker执行定时任务
    if worker_id == 0:
        asyncio.create_task(check_idle())
        asyncio.create_task(calculate_eta())

方案2：加锁机制（兼容所有Uvicorn版本，支持多机部署） #

如果你的Uvicorn版本低于0.12.0，或是需要多机部署避免单点问题，可以用锁机制保证只有一个worker抢到执行权限：

• 单机部署可以用文件锁：

import fcntl
import os
import asyncio

LOCK_FILE = "/tmp/worker_task.lock"

@app.on_event('startup')
async def app_startup():
    f = open(LOCK_FILE, "w")
    try:
        # 非阻塞抢锁，抢到的worker执行任务
        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
        asyncio.create_task(check_idle())
        asyncio.create_task(calculate_eta())
    except BlockingIOError:
        # 没抢到锁直接跳过
        pass

• 多机部署时把文件锁替换为Redis等分布式锁即可，逻辑完全一致。

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