一、先讲个我踩过的坑去年接手一个 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(lifespan=lifespan)就这么简单。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("postgresql+asyncpg://user:pass@localhost/db") AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) @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(lifespan=lifespan)注意一个细节：我在启动时建表了。如果你用 Alembic 做迁移，建表这步可以省掉，别混着用，不然容易出问题。场景 2：Redis 连接import redis.asyncio as redis @asynccontextmanager async def lifespan(app: FastAPI): app.state.redis = await redis.from_url("redis://localhost", decode_responses=True) 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(lifespan=lifespan)路由里用的时候：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坑 2：yield后面的代码不执行有个新手同事问我：" 我在yield后面写了打印，怎么从来没看到过？"原因是他用uvicorn启动后，直接Ctrl+C强杀了进程。强杀的情况下 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，建议找个时间迁一下，半小时的事，能省掉后面很多麻烦。
一、先讲个我踩过的坑去年接手一个 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(lifespan=lifespan)就这么简单。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("postgresql+asyncpg://user:pass@localhost/db") AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) @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(lifespan=lifespan)注意一个细节：我在启动时建表了。如果你用 Alembic 做迁移，建表这步可以省掉，别混着用，不然容易出问题。场景 2：Redis 连接import redis.asyncio as redis @asynccontextmanager async def lifespan(app: FastAPI): app.state.redis = await redis.from_url("redis://localhost", decode_responses=True) 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(lifespan=lifespan)路由里用的时候：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坑 2：yield后面的代码不执行有个新手同事问我：" 我在yield后面写了打印，怎么从来没看到过？"原因是他用uvicorn启动后，直接Ctrl+C强杀了进程。强杀的情况下 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，建议找个时间迁一下，半小时的事，能省掉后面很多麻烦。