如何在启动FastAPI应用前添加持续运行的后台任务？

你之前尝试的BackgroundTasks是为单个请求关联的后台任务设计的，必须触发请求才会执行，而且它依赖请求上下文，没法直接在lifespan事件里使用。要实现应用启动就持续运行、直到应用关闭才停止的后台任务，需要用异步任务的原生方式处理。

实现步骤 #

• 定义支持优雅停止的后台任务函数，用事件控制循环
• 在lifespan上下文管理器中启动任务，在应用关闭时完成清理

完整代码示例 #

import asyncio
from fastapi import FastAPI
from contextlib import asynccontextmanager

# 用于控制任务停止的信号事件
stop_task_event = asyncio.Event()

async def continuous_background_task():
    """持续运行的后台任务逻辑"""
    while not stop_task_event.is_set():
        # 替换成你的实际任务代码
        print("后台任务正在执行...")
        await asyncio.sleep(3)  # 用异步睡眠避免阻塞事件循环，模拟任务间隔
    print("后台任务已优雅停止")

@asynccontextmanager
async def app_lifespan(app: FastAPI):
    # 应用启动时：创建并启动后台任务
    background_task = asyncio.create_task(continuous_background_task())
    yield  # 应用运行阶段
    # 应用关闭时：发送停止信号，等待任务结束
    stop_task_event.set()
    await background_task

# 初始化FastAPI应用并绑定lifespan
app = FastAPI(lifespan=app_lifespan)

# 测试接口，验证应用状态
@app.get("/")
async def health_check():
    return {"status": "running", "message": "后台任务正在持续执行"}

关键说明 #

1. 为什么不用BackgroundTasks？
   BackgroundTasks是FastAPI为请求场景设计的工具，它会在当前请求处理完成后执行任务，生命周期和请求绑定，没有请求上下文的lifespan环境下无法注入和使用。

2. 优雅停止的必要性
   用asyncio.Event控制任务循环，能避免应用关闭时强制终止任务导致的资源泄漏（比如未关闭的数据库连接、未完成的IO操作）。

3. 异步任务注意事项
   任务内部的IO操作必须使用异步版本（比如asyncio.sleep、异步数据库驱动），否则会阻塞整个FastAPI的事件循环，影响接口响应。

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