Python Web框架-FastAPI教程-生命周期asynccontextmanager + lifespan 作者:马育民 • 2026-05-09 23:25 • 阅读:10003 # 提出问题 在 FastAPI 开发后端项目时,数据库是必备资源。很多数新手会遇到以下 **3 个致命问题**: 1. **重复创建连接**:每请求一次创建一次数据库连接,数据库压力巨大,极易卡死。 2. **连接无法关闭**:服务关停后,数据库连接残留,产生大量死连接,占用数据库连接池。 3. **代码散乱难维护**:连接初始化、关闭代码分散在不同文件,后期无法统一管理。 ### 错误示范:新手最常写的代码 下面是典型错误写法:接口内部直接创建数据库连接。 ```python @app.get("/user") async def get_user(): # 每来一次请求,新建一次连接(极度错误) db = AsyncDB() await db.connect() return {"data": "用户数据"} # 没有手动关闭,连接泄漏 ``` ### 旧方案缺陷:startup/shutdown(淘汰) 有人会使用旧生命周期写法,但是依旧存在缺陷: ```python @app.on_event("startup") async def startup(): # 启动连接 global db db = AsyncDB() await db.connect() @app.on_event("shutdown") async def shutdown(): # 关闭连接 await db.close() ``` ### 旧方案存在的问题 - 初始化和关闭**代码分离**,可读性差,维护麻烦 - 依赖全局变量,多线程、多进程下不安全 - 没有上下文约束,极端情况下依旧资源泄漏 # 解决方案:Lifespan 生命周期 ### 什么是 lifespan? **lifespan** 是 FastAPI 官方推出的应用生命周期管理器,专门用于解决:**数据库、Redis、MQ 等全局资源统一初始化与销毁**。 它把「启动代码」和「关闭代码」写在同一个函数内,天然形成上下文,保证资源成对释放,彻底解决连接泄漏问题。 ### 什么是 asynccontextmanager? `asynccontextmanager` 是 Python 内置库 `contextlib` 提供的**异步上下文管理器装饰器**。 作用:固定格式拆分 **前置代码(启动)** 和 **后置代码(销毁)**,适配 lifespan。 ### 语法格式 ```python from contextlib import asynccontextmanager @asynccontextmanager async def demo(): # 前置逻辑:应用启动时执行(创建数据库连接) yield # 后置逻辑:应用关闭时执行(断开数据库连接) ``` ### 执行规则 - **yield 之前:启动代码** - **yield 本身:程序运行中** - **yield 之后:关闭代码** # 例子 这是官方唯一推荐写法,所有项目通用。 ```python from contextlib import asynccontextmanager from fastapi import FastAPI @asynccontextmanager async def lifespan(app: FastAPI): # ========== 启动阶段:创建数据库连接 ========== print("应用启动,准备连接数据库") yield # ========== 关闭阶段:断开数据库连接 ========== print("应用关闭,释放数据库连接") app = FastAPI(lifespan=lifespan) ``` ### 执行顺序 1. 运行 uvicorn,优先执行 yield 上方代码:连接数据库 2. 服务启动,监听端口,接收请求,复用同一个数据库连接 3. 按下 Ctrl\+C 关闭服务,执行 yield 下方代码:断开连接 # 案例 数据库全局资源管理 ### 工程规范:资源挂载到 app.state 禁止使用普通全局变量。FastAPI 官方规范:所有全局资源挂载至 `app\.state`,隔离性更强、多线程安全,专门用来存放数据库、Redis 等连接实例。 ### 代码 ```python from contextlib import asynccontextmanager from fastapi import FastAPI import asyncio # 模拟异步数据库连接(真实项目替换为 XX连接池XX) class AsyncDB: async def connect(self): await asyncio.sleep(0.2) print("✅ 数据库连接成功") async def close(self): await asyncio.sleep(0.2) print("❌ 数据库连接关闭") # 生命周期管理:统一管控数据库 @asynccontextmanager async def lifespan(app: FastAPI): # 【启动】初始化数据库连接(真实项目是连接池) db = AsyncDB() await db.connect() # 挂载到全局,所有接口共享 app.state.db = db yield # 【关闭】优雅释放数据库连接 await app.state.db.close() app = FastAPI(lifespan=lifespan) # 在接口中使用全局数据库连接 @app.get("/user") async def get_user(): # 取出全局数据库实例 db = app.state.db return {"code": 200, "msg": "复用全局数据库连接,无重复创建"} ``` # 新旧写法对比 | 对比项 | lifespan(新版) | startup/shutdown(旧版) | | --- | --- | --- | | 代码位置 | 数据库连接/关闭写在一起 | 分散两处,容易遗漏关闭 | | 连接安全性 | 上下文强制释放,无死连接 | 异常退出容易残留连接 | | 变量使用 | app\.state 安全挂载 | 依赖全局变量,线程不安全 | | 官方状态 | 持续维护、主推 | 0\.95\+ 标记废弃 | # 易错点总结 ### 语法易错 - **必须写 yield**:缺少 yield,程序无法切换运行状态,数据库卡死 - **必须加 async**:数据库异步连接,lifespan 强制异步函数 - **异步操作必须 await**:connect、close 必须加 await ### 工程规范易错 - **禁止** 在 lifespan 写 **同步阻塞代码**,会阻塞数据库初始化 - 不要用全局变量存储数据库实例,统一挂载 app.state - 所有初始化资源,必须在关闭阶段手动释放,防止连接堆积 # 阻塞同步代码 将 **阻塞函数名** 写到下面代码中 **注意:** **阻塞函数名** **后面不要加** **()**,否则表示直接调用 ``` await asyncio.get_event_loop().run_in_executor(None, 阻塞函数名) ``` ### 例子 ``` await asyncio.get_event_loop().run_in_executor(None, init) ``` **解释:** `init` 是阻塞同步函数名,后面 **必须不能加** `()`,否则表示直接调用 # 总结 - 数据库怕泄漏,连接要统一; - 装饰器修饰、函数必须异步; - yield 分割、前连后闭; - 资源挂载 state、严禁全局; - 新项目 lifespan、抛弃 startup。 原文出处:http://www.malaoshi.top/show_1GW3HX1SfD8w.html