如何使用FastAPI的lifespan方法解决异步资源初始化问题

FastAPI生命周期管理的核心挑战

在基于FastAPI构建现代Web服务时，异步资源初始化是开发高频遇到的问题之一。官方文档虽然提供了lifespan的基本用法，但实际应用中开发者常会遇到异步上下文管理器的执行顺序问题。

典型问题场景还原

from fastapi import FastAPI
import asyncpg

app = FastAPI()

async def lifespan(app: FastAPI):
    # 数据库连接池初始化
    app.state.pool = await asyncpg.create_pool(...)
    yield
    await app.state.pool.close()

上述代码看似合理，但在实际运行时可能出现以下问题：

请求处理阶段访问未初始化的连接池
异步资源清理顺序异常
生命周期事件与中间件执行时序冲突

根本原因分析

深入研究发现，问题源于异步上下文管理器在FastAPI生命周期中的特殊行为：

启动阶段(startup)和关闭阶段(shutdown)的异步任务调度机制
Python生成器与异步生成器的执行差异
事件循环在不同阶段的可用性限制

解决方案实现

推荐使用双重验证模式确保资源可用性：

async def lifespan(app: FastAPI):
    try:
        pool = await asyncpg.create_pool(
            min_size=2,
            max_size=10,
            timeout=30
        )
        app.state.pool_ready = asyncio.Event()
        async with pool.acquire() as conn:
            await conn.execute("SELECT 1")
            app.state.pool = pool
            app.state.pool_ready.set()
        yield
    finally:
        if hasattr(app.state, 'pool'):
            await app.state.pool.close()

最佳实践建议

场景	解决方案	优势
数据库连接	预热测试连接+就绪事件	避免冷启动问题
外部服务	指数退避重试机制	提高容错能力
配置文件	预加载校验	早期发现问题

性能优化技巧

通过依赖注入优化资源访问模式：

async def get_db(request: Request):
    await request.app.state.pool_ready.wait()
    return request.app.state.pool

@app.get("/users")
async def list_users(db=Depends(get_db)):
    async with db.acquire() as conn:
        return await conn.fetch("SELECT * FROM users")

监控与调试

建议集成以下监控指标：

资源初始化耗时百分位图
就绪状态变更时间戳
关闭阶段的资源释放成功率

通过uvicorn的--lifespan on参数可以启用完整的生命周期事件日志，这对调试时序问题至关重要。