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

在构建高性能的 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)}

总结

在本篇博客中，我们介绍了如何在 FastAPI 中使用 aioredis 连接池来实现异步 Redis 操作，并且讲解了如何配置连接池的最小连接数和最大连接数。通过设置合适的连接池大小，可以有效管理 Redis 连接，提升性能并避免 Redis 服务器过载。

使用 Redis 连接池的主要优势是提升性能、简化代码并合理管理资源。在构建高并发的 Web 应用时，连接池是优化 Redis 使用的重要工具。