当前位置：首页 > news >正文

Fastapi教程：使用 aioredis 连接池执行Redis 的高效异步操作

news 2025/7/19 6:06:11

在构建高性能的 Web 应用时，缓存系统是一个至关重要的组成部分。Redis 是最常见的缓存系统之一，它提供了高效的存储与读取机制。然而，在与 Redis 进行频繁交互时，创建和销毁连接可能会成为瓶颈。为了优化这一问题，我们可以使用 Redis 连接池。

在本篇博客中，我们将介绍如何在 FastAPI 中使用 aioredis 连接池进行 Redis 的高效异步操作，并讲解如何设置连接池的大小来优化 Redis 操作。

为什么使用 Redis 连接池？

Redis 连接池提供了以下优势：

高效的连接复用：通过连接池，多个请求可以复用 Redis 连接，避免了频繁地创建和销毁连接，提升性能。
资源管理：连接池可以限制最大连接数，防止过多的连接导致 Redis 服务器过载。
简化代码：与每次手动创建 Redis 连接相比，连接池能够自动管理连接生命周期，简化代码。

aioredis 是一个 Python 异步 Redis 客户端，它支持 asyncio，能够在 FastAPI 这样的异步框架中高效工作。

环境准备

首先，确保安装了 fastapi 和 aioredis：

pip install fastapi
pip install uvicorn
pip install aioredis

创建 Redis 连接池依赖

我们将使用依赖注入来管理 Redis 连接池。通过定义一个依赖函数，FastAPI 可以在每个请求生命周期内为我们提供 Redis 连接池。

1. 创建 Redis 连接池

在创建 Redis 连接池时，我们可以设置连接池的最大连接数和最小连接数。aioredis.create_redis_pool 函数提供了这些参数。通过设置适当的连接池大小，我们可以优化 Redis 的使用，避免连接数过多导致 Redis 服务器负载过高。

minsize：连接池的最小连接数。即使没有请求，连接池也会保持至少这么多连接。
maxsize：连接池的最大连接数。当连接池中连接数达到最大值时，新的请求将被阻塞，直到有连接空闲。

创建带有连接池大小配置的 Redis 连接池

import aioredis
from fastapi import FastAPI, Depends
from typing import AsyncGenerator# Redis 配置
REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_DB = 0
REDIS_POOL_MIN_SIZE = 10  # 最小连接数
REDIS_POOL_MAX_SIZE = 50  # 最大连接数# FastAPI 应用实例
app = FastAPI()# Redis 连接池依赖函数
async def get_redis_pool() -> AsyncGenerator[aioredis.Redis, None]:"""创建并返回 Redis 连接池。通过依赖注入管理 Redis 连接池。"""redis = await aioredis.create_redis_pool((REDIS_HOST, REDIS_PORT),db=REDIS_DB,encoding="utf-8",minsize=REDIS_POOL_MIN_SIZE,  # 设置最小连接数maxsize=REDIS_POOL_MAX_SIZE   # 设置最大连接数)try:yield redisfinally:redis.close()await redis.wait_closed()

在上述代码中，我们设置了最小连接数 (REDIS_POOL_MIN_SIZE) 为 10，最大连接数 (REDIS_POOL_MAX_SIZE) 为 50。根据应用的并发需求，可以调整这些值来平衡性能和资源消耗。

2. 定义路由并使用依赖注入

接下来，我们将依赖注入 Redis 连接池到 FastAPI 的路由处理函数中。这样，Redis 连接池将作为一个参数传入每个需要与 Redis 交互的路由。

示例 1：设置缓存

@app.post("/cache/{key}")
async def cache_data(key: str, value: str, redis: aioredis.Redis = Depends(get_redis_pool)):"""设置缓存数据"""try:await redis.set(key, value)return {"message": "Data cached successfully", "key": key, "value": value}except Exception as e:return {"error": str(e)}

示例 2：获取缓存

@app.get("/cache/{key}")
async def get_cached_data(key: str, redis: aioredis.Redis = Depends(get_redis_pool)):"""获取缓存数据"""try:value = await redis.get(key)if value is None:return {"message": "Key not found"}return {"key": key, "value": value}except Exception as e:return {"error": str(e)}

示例 3：删除缓存

@app.delete("/cache/{key}")
async def delete_cached_data(key: str, redis: aioredis.Redis = Depends(get_redis_pool)):"""删除缓存数据"""try:result = await redis.delete(key)if result == 0:return {"message": "Key not found"}return {"message": f"Data with key {key} deleted successfully"}except Exception as e:return {"error": str(e)}

在这些路由中，redis: aioredis.Redis = Depends(get_redis_pool) 表示 FastAPI 会自动调用 get_redis_pool 来提供一个 Redis 连接池实例。通过这种方式，我们可以避免手动管理 Redis 连接池，并保持代码的简洁和模块化。

3. 启动和关闭事件

为了确保 Redis 连接池在 FastAPI 应用启动时创建并在关闭时清理，我们需要在 @app.on_event("startup") 和 @app.on_event("shutdown") 事件中进行相应的管理。这个管理已经在 get_redis_pool 函数中实现，因为 Redis 连接池会在 yield 语句之后关闭。

因此，我们不需要在 FastAPI 中显式管理 Redis 连接池的启动和关闭，FastAPI 会自动处理。

完整代码示例

以下是完整的代码示例，展示了如何在 FastAPI 中使用 Redis 连接池和依赖注入，并设置连接池大小：

import aioredis
from fastapi import FastAPI, HTTPException, Depends
from typing import AsyncGenerator# Redis 配置
REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_DB = 0
REDIS_POOL_MIN_SIZE = 10  # 最小连接数
REDIS_POOL_MAX_SIZE = 50  # 最大连接数# FastAPI 应用实例
app = FastAPI()# Redis 连接池依赖函数
async def get_redis_pool() -> AsyncGenerator[aioredis.Redis, None]:"""创建并返回 Redis 连接池。通过依赖注入管理 Redis 连接池。"""redis = await aioredis.create_redis_pool((REDIS_HOST, REDIS_PORT),db=REDIS_DB,encoding="utf-8",minsize=REDIS_POOL_MIN_SIZE,  # 设置最小连接数maxsize=REDIS_POOL_MAX_SIZE   # 设置最大连接数)try:yield redisfinally:redis.close()await redis.wait_closed()@app.post("/cache/{key}")
async def cache_data(key: str, value: str, redis: aioredis.Redis = Depends(get_redis_pool)):"""设置缓存数据"""try:await redis.set(key, value)return {"message": "Data cached successfully", "key": key, "value": value}except Exception as e:return {"error": str(e)}@app.get("/cache/{key}")
async def get_cached_data(key: str, redis: aioredis.Redis = Depends(get_redis_pool)):"""获取缓存数据"""try:value = await redis.get(key)if value is None:return {"message": "Key not found"}return {"key": key, "value": value}except Exception as e:return {"error": str(e)}@app.delete("/cache/{key}")
async def delete_cached_data(key: str, redis: aioredis.Redis = Depends(get_redis_pool)):"""删除缓存数据"""try:result = await redis.delete(key)if result == 0:return {"message": "Key not found"}return {"message": f"Data with key {key} deleted successfully"}except Exception as e:return {"error": str(e)}