如何在Python FastAPI应用中启动异步后台守护进程执行定时任务

之前方案失败原因 #

• 直接await守护函数：守护函数是死循环，await会阻塞启动流程，导致FastAPI无法完成端口监听，服务不可用
• 直接使用fastapi.BackgroundTasks：该类是请求级的后台任务，绑定到单次HTTP请求上下文，在启动事件中实例化的对象没有关联请求，启动流程结束后会被直接销毁，任务不会执行
• 多线程方案：每个Uvicorn worker是独立进程，启动N个worker就会生成N个独立线程执行任务，导致重复写入数据，且跨线程操作异步事件逻辑本身不符合异步编程规范

正确实现方案 #

单Worker场景（无重复执行问题） #

直接用asyncio.create_task将守护协程提交到主事件循环即可，不会阻塞启动流程，事件循环空闲时会自动执行任务：

import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup_event() -> None:
    """服务启动时运行的任务"""
    # 提交后台任务到事件循环，不阻塞启动流程
    asyncio.create_task(Gatherer().start_metering_daemon())

注意：建议在start_metering_daemon的循环内加全局异常捕获，避免任务抛出未处理异常后终止，无人感知

多Worker场景（避免重复写入） #

多Worker下每个进程是独立的，无法直接在应用层互斥，两种轻量方案可选，无需引入Celery：

1. 拆分部署：只启动1个Worker运行定时任务，其他Worker只处理API请求，或者单独启动一个无端口监听的异步进程专门跑定时任务
2. 分布式锁拦截：
   • 用Redis的SETNX实现轻量锁，每次执行采集任务前先抢锁，抢到再执行，没抢到直接跳过本次，锁的过期时间设置为大于任务最大执行时长即可
   • 无Redis的情况下也可以用数据库唯一键约束，每次执行前插入带执行时间戳/日期的唯一记录，插入成功则执行，失败则跳过，完全无需额外中间件
     示例锁逻辑：

async def gather_meterings(self) -> None:
    # 抢锁，过期时间300秒可根据实际任务执行时长调整
    lock_ok = await redis.set("metering_task:lock", "running", ex=300, nx=True)
    if not lock_ok:
        return
    try:
        # 原有数据采集、入库逻辑
        ...
    finally:
        # 执行完成释放锁
        await redis.delete("metering_task:lock")

内容的提问来源于stack exchange，提问作者Арсений Величко