一、先讲个我踩过的坑去年接手一个 FastAPI 项目线上偶尔报数据库连接泄漏。查了半天发现是这么写的app.on_event(startup) async def startup(): await db.connect() app.on_event(shutdown) async def shutdown(): await db.disconnect()看起来没毛病对吧启动时连数据库关闭时断开。但问题出在如果startup执行到一半挂了shutdown根本不会跑。比如数据库连到一半超时了前面已经建立的几个连接就成了孤儿没人关。次数多了连接池就满了然后整个服务假死。更坑的是FastAPI 从 0.93 版本开始把on_event标记为 deprecated 了官方明确推荐用lifespan代替。我当时改完之后那个泄漏问题再也没出现过。所以今天这篇就是想把lifespan给你讲明白别再走我走过的弯路。二、lifespan到底是个啥简单说lifespan就是一个异步上下文管理器它包住了整个应用的生命周期。应用启动前进入yield之前的代码应用关闭后执行yield之后的代码。from contextlib import asynccontextmanager from fastapi import FastAPI asynccontextmanager async def lifespan(app: FastAPI): # 启动时执行 print(应用启动了初始化资源...) await db.connect() yield # 应用在这里运行 # 关闭时执行 print(应用关闭清理资源...) await db.disconnect() app FastAPI(lifespanlifespan)就这么简单。yield把代码分成两半上面是启动逻辑下面是清理逻辑。你可能会问这跟try/finally有啥区别本质上就是try/finally的语法糖。但关键在于不管启动过程中有没有异常yield后面的清理代码都会执行。这就解决了我前面说的那个泄漏问题。三、实际项目中怎么用场景 1数据库连接池这是最常见的用法。以 SQLAlchemy 异步引擎为例from contextlib import asynccontextmanager from fastapi import FastAPI from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker engine create_async_engine(postgresqlasyncpg://user:passlocalhost/db) AsyncSessionLocal sessionmaker(engine, class_AsyncSession, expire_on_commitFalse) asynccontextmanager async def lifespan(app: FastAPI): # 启动时创建连接池 async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) yield # 关闭时释放连接池 await engine.dispose() app FastAPI(lifespanlifespan)注意一个细节我在启动时建表了。如果你用 Alembic 做迁移建表这步可以省掉别混着用不然容易出问题。场景 2Redis 连接import redis.asyncio as redis asynccontextmanager async def lifespan(app: FastAPI): app.state.redis await redis.from_url(redis://localhost, decode_responsesTrue) yield await app.state.redis.close()这里我把 redis 客户端挂到了app.state上这样在路由里就能通过request.app.state.redis拿到。比全局变量干净测试的时候也好 mock。场景 3加载机器学习模型这个场景太典型了。模型加载慢不能每次请求都加载一次必须在启动时预加载import joblib asynccontextmanager async def lifespan(app: FastAPI): # 启动时加载模型这步可能要好几秒 app.state.model joblib.load(model.pkl) app.state.vectorizer joblib.load(vectorizer.pkl) print(模型加载完成) yield # 模型一般不需要特别清理但可以在这里释放显存 del app.state.model app FastAPI(lifespanlifespan)路由里用的时候from fastapi import Request app.post(/predict) async def predict(request: Request, text: str): model request.app.state.model vectorizer request.app.state.vectorizer features vectorizer.transform([text]) result model.predict(features) return {result: result.tolist()}场景 4多个资源一起管理真实项目不可能只初始化一个东西。数据库、Redis、MQ、模型…… 全塞在一个lifespan里也行但写多了就乱。我的做法是拆成多个上下文管理器然后用AsyncExitStack组合from contextlib import asynccontextmanager, AsyncExitStack asynccontextmanager async def init_db(app): await db.connect() yield await db.disconnect() asynccontextmanager async def init_redis(app): app.state.redis await redis.from_url(redis://localhost) yield await app.state.redis.close() asynccontextmanager async def lifespan(app: FastAPI): async with AsyncExitStack() as stack: await stack.enter_async_context(init_db(app)) await stack.enter_async_context(init_redis(app)) yieldAsyncExitStack的好处是前面的资源初始化成功了后面某个失败了已经成功的那些会自动按逆序清理。不会出现半初始化的僵尸状态。四、新手容易踩的几个坑坑 1在lifespan里做耗时的同步操作# 错误写法 asynccontextmanager async def lifespan(app): model load_big_model() # 这个函数是同步的要跑 10 秒 yield如果load_big_model()是个同步阻塞函数它会卡住整个事件循环。这期间应用没法处理任何请求甚至健康检查都超时K8s 可能直接把你 pod 杀了。解决办法是扔到线程池里跑import asyncio asynccontextmanager async def lifespan(app): model await asyncio.to_thread(load_big_model) app.state.model model yield坑 2yield后面的代码不执行有个新手同事问我 我在yield后面写了打印怎么从来没看到过原因是他用uvicorn启动后直接CtrlC强杀了进程。强杀的情况下 Python 来不及执行清理逻辑就退出了。正常的SIGTERM信号比如docker stop、k8s 滚动更新是会触发lifespan关闭逻辑的但kill -9不会。所以别把lifespan的清理当成 100% 可靠的兜底该做的幂等和超时还是要做。坑 3把lifespan当 middleware 用有人在lifespan里写请求级别的逻辑比如统计请求数。这是不对的。lifespan只在应用启动和关闭时各执行一次不是每个请求都跑。请求级别的逻辑用 middleware 或依赖注入。简单区分应用级资源连接池、模型、配置→lifespan请求级逻辑鉴权、日志、计时→ middleware / Depends坑 4测试的时候忘了lifespan用TestClient测试的时候默认是不会触发lifespan的。你需要这样from fastapi.testclient import TestClient with TestClient(app) as client: # 进入 with 时触发 startup退出时触发 shutdown response client.get(/) assert response.status_code 200用with包住lifespan才会正常执行。不然你测试里访问request.app.state.redis会报AttributeError。五、跟旧方式的对比最后做个总结为啥要从on_event迁到lifespan表格对比项app.on_eventlifespan状态已废弃deprecated官方推荐异常安全启动失败不触发 shutdown保证清理代码执行资源共享靠全局变量可挂app.state更规范多资源管理多个事件函数顺序不可控AsyncExitStack组合顺序明确测试支持需手动触发TestClient原生支持迁移也很简单把startup里的代码搬到yield前面shutdown里的搬到yield后面完事。写在最后lifespan不是什么高深的东西本质就是个上下文管理器。但它解决了 FastAPI 项目里一个很实际的问题资源的初始化和清理怎么写才安全、才优雅。我见过太多项目图省事把数据库连接写成全局变量启动时连一下就不管了。平时跑着没事一到发布、重启、异常的时候就出幺蛾子。写代码嘛还是得把这些 边角料 处理好线上才能少熬夜。如果你项目里还在用on_event建议找个时间迁一下半小时的事能省掉后面很多麻烦。
