本文详细阐述如何在FastAPI应用中使用生命周期事件（lifespan）高效管理数据库连接池、Redis客户端等全局资源。通过在应用启动时初始化资源并存储于app.state，结合FastAPI强大的依赖注入系统，实现资源在整个应用生命周期内的共享与复用，同时保持代码的整洁与可维护性，避免在每个请求中重复创建和关闭资源。
在构建高性能的Web服务时，管理外部资源（如数据库连接池、Redis客户端、消息队列连接等）的生命周期至关重要。不恰当的资源管理可能导致性能瓶颈、资源泄露甚至应用崩溃。FastAPI提供了强大的lifespan机制（或旧版on_event）来优雅地处理这些问题，结合其依赖注入系统，可以实现高效且可维护的资源管理。
1. FastAPI中的资源管理挑战
在FastAPI应用中，常见的资源管理挑战包括：
- 资源初始化与销毁时机: 数据库连接池等资源应在应用启动时创建，并在应用关闭时安全地销毁，而不是在每个请求中重复创建和关闭。
- 全局资源访问: 如何在FastAPI的各个组件（如路由处理器、其他依赖项）中便捷地访问这些全局资源。
- 与依赖注入系统集成: 如何在享受FastAPI依赖注入便利性的同时，使用生命周期管理的全局资源，避免直接从request.app.state手动获取。
传统的做法可能是在模块级别直接初始化连接池，但这会导致连接池的生命周期与应用不匹配，且无法在应用关闭时优雅地释放资源。
2. lifespan机制与app.state
FastAPI的lifespan是一个异步上下文管理器，它允许您定义在应用启动前和关闭后执行的代码。这是管理全局资源生命周期的理想场所。
- lifespan上下文管理器: 在FastAPI应用初始化时传入，其内部的yield语句将应用生命周期分为启动前和启动后两个阶段。
- app.state: FastAPI应用实例提供了一个state属性，这是一个可用于存储任何应用级别状态的字典。在lifespan中初始化的资源可以存储在此处，供整个应用生命周期内的其他部分访问。
3. 集成Redis连接池：实践指南
本节将以redis.asyncio为例，详细演示如何使用lifespan管理Redis连接池，并通过FastAPI的依赖注入系统在路由中便捷地使用。
步骤一：定义应用配置
首先，我们需要一个配置来存储Redis连接URL。这通常通过pydantic-settings库实现。
# app/core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""
应用配置类，从环境变量或.env文件加载。
"""
REDIS_URL: str = "redis://localhost:6379/0" # 默认Redis连接URL
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
settings = Settings()步骤二：定义lifespan事件处理器
在应用启动时，我们将创建redis.asyncio.ConnectionPool和redis.asyncio.Redis客户端实例，并将其存储在app.state中。在应用关闭时，我们将优雅地关闭Redis客户端，从而关闭其底层连接池。
# app/main.py (部分代码)
import redis.asyncio as aredis
from fastapi import FastAPI
from contextlib import asynccontextmanager
from app.core.config import settings
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
FastAPI应用生命周期管理器。
在应用启动时创建Redis连接池和客户端实例，并存储在app.state中；
在应用关闭时关闭客户端，从而关闭底层连接池。
"""
print("应用启动：初始化Redis连接池...")
# 1. 创建Redis连接池
# decode_responses=True 确保从Redis获取的响应自动解码为字符串
redis_pool = aredis.ConnectionPool.from_url(settings.REDIS_URL, decode_responses=True)
# 2. 创建Redis客户端实例
# 这个客户端实例是线程安全的，它会从连接池中获取和释放连接。
# 它是应用中与Redis交互的主要接口，应保持长生命周期。
app.state.redis_client = aredis.Redis(connection_pool=redis_pool)
print("Redis连接池和客户端初始化完成。")
yield # 应用在此处运行，处理所有传入的请求
print("应用关闭：关闭Redis客户端和连接池...")
# 3. 在应用关闭时，关闭Redis客户端。
# 调用aclose()会关闭客户端持有的所有连接，并最终关闭其关联的连接池。
await app.state.redis_client.aclose()
print("Redis客户端和连接池已关闭。")
# 初始化FastAPI应用，并传入lifespan处理器
app = FastAPI(lifespan=lifespan)步骤三：创建依赖注入函数
为了在路由处理器中方便地使用Redis客户端，我们需要创建一个依赖注入函数。这个函数将直接从app.state中获取已初始化的redis.asyncio.Redis客户端实例并提供给请求。
# app/main.py (部分代码)
from typing import Annotated
from fastapi import Depends
# ... (上面的导入和lifespan定义) ...
async def get_redis_client() -> aredis.Redis:
"""
依赖注入函数，为每个请求提供一个Redis客户端实例。
该实例在应用生命周期中创建，并从app.state获取。
由于redis.asyncio.Redis客户端是长生命周期且内部管理连接池，
这里直接返回该实例即可，无需在请求结束后手动关闭。
"""
# 从app.state获取已经初始化的Redis客户端实例
return app.state.redis_client
# 定义一个类型别名，方便在路由中使用
RedisDependency = Annotated[aredis.Redis, Depends(get_redis_client)]重要说明: redis.asyncio.Redis客户端实例本身是设计为长生命周期的，它内部管理着从ConnectionPool获取和释放连接的逻辑。因此，在get_redis_client依赖函数中，我们直接返回这个长生命周期的客户端实例，而不需要使用yield或在finally块中对其进行aclose()操作。客户端的关闭由lifespan函数在应用关闭时统一处理。
步骤四：在路由中使用依赖
现在，您可以在任何FastAPI路由处理器中通过Depends(get_redis_client)（或其类型别名RedisDependency）来注入Redis客户端实例。
# app/main.py (部分代码)
# ... (上面的导入、lifespan定义、get_redis_client和RedisDependency定义) ...
@app.get("/redis-test")
async def redis_test(redis: RedisDependency):
"""
一个示例路由，使用注入的Redis客户端执行GET 
