在FastAPI中利用lifespan与依赖注入高效管理Redis连接池

云萱君_7716

发布时间：2025-12-01 12:36:33

717人浏览过

来源于php中文网

原创

在FastAPI中利用lifespan与依赖注入高效管理Redis连接池

本文详细介绍了如何在fastapi应用中，利用其`lifespan`生命周期管理功能，结合依赖注入（`depends`），优雅地创建和释放redis连接池。通过将连接池的初始化与清理绑定到应用启动和关闭事件，并设计一个从该池获取redis客户端的依赖函数，实现了资源的高效复用与统一管理，同时避免了在路由中直接访问`request.app.state`的复杂性，提升了代码的可维护性与专业性。

引言：FastAPI应用中的资源管理挑战

在构建高性能的异步Web服务时，高效地管理外部资源（如数据库连接池、缓存客户端、消息队列连接等）至关重要。这些资源通常需要在应用程序启动时进行初始化，并在应用程序关闭时进行清理，以确保资源的正确分配和释放，避免资源泄露。FastAPI作为一款现代、高性能的Web框架，提供了强大的依赖注入系统，但如何将全局资源的生命周期管理与依赖注入机制完美结合，是开发者需要面对的一个常见问题。

本文将聚焦于如何在FastAPI中，利用其lifespan功能来管理Redis连接池的创建与释放，并结合依赖注入，使得在路由函数中能够以简洁、一致的方式获取Redis客户端实例，同时避免直接在业务逻辑中暴露底层资源管理细节。

FastAPI的生命周期管理：lifespan

FastAPI提供了lifespan（或旧版中的on_event("startup")和on_event("shutdown")）机制，允许开发者在应用程序启动和关闭时执行自定义逻辑。这是一个理想的场所来初始化和清理应用程序所需的全局资源。

lifespan是一个异步上下文管理器，它在FastAPI应用程序启动时进入，在应用程序关闭时退出。其基本结构如下：

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 应用程序启动时执行的逻辑，例如初始化数据库连接池
    print("Application startup: Initializing resources...")
    app.state.my_resource = "Initialized Resource" # 将资源存储在 app.state
    yield
    # 应用程序关闭时执行的逻辑，例如关闭数据库连接池
    print("Application shutdown: Cleaning up resources...")
    # 清理 app.state.my_resource

通过将资源存储在app.state对象上，我们可以在应用程序的任何地方（包括依赖项）访问这些全局资源。

使用lifespan管理Redis连接池

我们将以redis.asyncio库为例，演示如何使用lifespan来管理Redis连接池，并结合FastAPI的依赖注入系统。

1. 配置Redis连接URL

首先，我们需要一个地方来存储Redis的连接URL。推荐使用Pydantic Settings进行配置管理。

app/core/config.py:

Sentinal Redis

监控Redis服务器健康状况、内存、性能及BullMQ队列。检查队列深度、失败任务，分析慢查询并诊断问题。

下载

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    REDIS_URL: str = "redis://localhost:6379/0" # 默认Redis连接URL

settings = Settings()

2. 定义lifespan上下文管理器

在FastAPI应用的主文件中，定义lifespan上下文管理器来初始化和关闭Redis连接池。

app/main.py:

from contextlib import asynccontextmanager
from typing import AsyncGenerator, Annotated

import redis.asyncio as aredis
from fastapi import FastAPI, Depends, APIRouter, Request

from app.core.config import settings # 导入配置

# 定义 lifespan 上下文管理器
@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    FastAPI 应用程序生命周期管理器。
    在应用程序启动时初始化 Redis 连接池，并在关闭时将其断开。
    """
    print("Initializing Redis Connection Pool...")
    # 创建 Redis 连接池并存储在 app.state 中
    # decode_responses=True 使得从 Redis 获取的数据自动解码为字符串
    app.state.redis_pool = aredis.ConnectionPool.from_url(
        settings.REDIS_URL, decode_responses=True
    )
    print("Redis Connection Pool Initialized.")
    yield # 应用程序在此处开始处理请求
    print("Closing Redis Connection Pool...")
    # 应用程序关闭时，断开 Redis 连接池
    await app.state.redis_pool.disconnect()
    print("Redis Connection Pool Closed.")

# 创建 FastAPI 应用实例，并传入 lifespan
app = FastAPI(lifespan=lifespan)
router = APIRouter()

在此步骤中，我们：

在lifespan函数的yield之前，使用aredis.ConnectionPool.from_url创建了一个全局的Redis连接池。
将这个连接池实例存储在app.state.redis_pool中，使其在整个应用生命周期内可访问。
在yield之后，即应用程序关闭时，调用app.state.redis_pool.disconnect()来优雅地关闭连接池，释放所有连接。

3. 创建依赖注入函数

为了在路由函数中方便地使用Redis客户端，我们需要创建一个依赖注入函数。这个函数将从app.state中获取已初始化的连接池，并为每个请求提供一个Redis客户端实例。

# app/main.py (接着上面的代码)

async def get_redis_client(request: Request) -> AsyncGenerator[aredis.Redis, None]:
    """
    依赖注入函数，为每个请求提供一个 Redis 客户端实例。
    该客户端从应用程序全局连接池中获取连接。
    """
    # 从 app.state 中获取已初始化的 Redis 连接池
    pool = request.app.state.redis_pool
    # 从连接池创建一个 Redis 客户端实例
    # 注意：aredis.Redis 客户端是异步安全的，并且内部会管理从连接池获取和释放连接
    r = aredis.Redis(connection_pool=pool, decode_responses=True)
    try:
        yield r # 将 Redis 客户端提供给路由函数
    finally:
        # 在请求处理完毕后，关闭当前客户端实例，
        # 这会将底层连接释放回连接池。
        await r.aclose()

# 定义一个类型别名，方便在路由函数中使用
RedisDependency = Annotated[aredis.Redis, Depends(get_redis_client)]

这个get_redis_client依赖函数是关键：

它接收Request对象，以便访问app.state中存储的全局连接池。
它利用aredis.Redis(connection_pool=pool, ...)从全局连接池创建一个新的Redis客户端实例。这个客户端实例会在其生命周期内从连接池获取和释放连接。
使用try...finally块确保在请求处理完毕后，调用r.aclose()来关闭当前的Redis客户端实例，这会将它所使用的连接返回到连接池中，实现连接的复用。

4. 在FastAPI应用中使用依赖

现在，我们可以在FastAPI的路由函数中，通过依赖注入的方式，简洁地获取和使用Redis客户端了。

# app/main.py (接着上面的代码)

@router.post("/items/")
async def create_item(item_data: dict, r: RedisDependency):
    """
    示例路由：创建一个新的 item 并将其存储到 Redis。
    """
    item_id = item_data.get("id") or "new_item"
    key = f"item:{item_id}"
    await r.set(key, str(item_data))
    value = await r.get(key)
    return {"message": "Item created and stored in Redis", "item_id": item_id, "stored_value": value}

@router.get("/items/{item_id}")
async def read_item(item_id: str, r: RedisDependency):
    """
    示例路由：从 Redis 中读取一个 item。
    """
    key = f"item:{item_id}"
    value = await r.get(key)
    if value:
        return {"item_id": item_id, "data": value}
    return {"message": "Item not found", "item_id": item_id}

app.include_router(router)

通过r: RedisDependency，路由函数可以直接获得一个可用的Redis客户端，无需关心其初始化和清理过程。

优点与最佳实践

统一的资源管理：所有全局资源的生命周期（创建、初始化、清理）都由FastAPI的lifespan统一控制，确保了资源的正确管理。
依赖注入的便利性：通过Depends机制，路由函数无需关心资源获取的复杂细节，只需声明所需资源类型，FastAPI会自动注入，提高了代码的可读性和可维护性。
避免全局变量污染：将资源实例存储在app.state中，避免了直接使用全局变量，使得资源管理更加清晰和可控。
优雅的资源清理：lifespan确保在应用程序关闭时，连接池能够被正确断开，防止资源泄露。
解耦与测试性：这种模式将资源管理逻辑与业务逻辑解耦，使得单元测试和集成测试更加容易。你可以轻松地为get_redis_client依赖提供模拟实现。
避免直接在路由中使用request.app.state：虽然request.app.state可以访问全局资源，但将其封装在依赖函数中，使得路由函数更加专注于业务逻辑，避免了在每个路由中重复编写资源获取代码。