redis-py asyncio下ConnectionPool行为及FastAPI使用咨询

1. 并发超过max_connections时的行为 #

redis-py异步版的ConnectionPool在并发任务数超过max_connections限制时，默认会把超额的连接请求排队等待，直到有已占用的连接被释放回池。

如果初始化连接池时设置了wait_timeout参数（默认是None，即无限等待），当等待时间超过设定值时，会抛出redis.exceptions.ConnectionError异常。比如：

pool = ConnectionPool(max_connections=10, wait_timeout=5)  # 等待5秒仍无可用连接则抛出异常

2. 单个连接池创建多客户端vs复用单个客户端 #

两种方式本质上共享同一个连接池，性能几乎没有差异——因为Redis客户端实例只是连接池的一层包装，创建多个实例不会额外消耗资源，所有请求最终都由同一个池分配连接。

但复用单个客户端更高效合理：既省去了重复创建实例的微小开销，代码结构也更简洁清爽，维护起来更方便。

3. 你的示例代码优化建议 #

推荐直接复用已初始化的redis_shared客户端，修改后的代码如下：

from redis.asyncio import ConnectionPool, Redis
from fastapi import Depends, FastAPI

app = FastAPI()

# 初始化全局连接池和共享客户端
pool = ConnectionPool(max_connections=10)
redis_shared = Redis(connection_pool=pool)

async def get_redis():
    return redis_shared  # 复用单个客户端实例

@app.get("/")
async def my_route(redis: Redis = Depends(get_redis)):
    return await redis.get('mykey')

4. 额外最佳实践 #

• 合理设置max_connections：根据Redis服务器的最大连接数（默认通常是10000）和FastAPI的并发量调整，别设太小导致请求排队超时，也别设太大占用Redis过多资源。
• 用依赖注入统一管理：通过FastAPI的Depends管理Redis客户端，后续修改配置、替换测试模拟客户端都更方便。
• 捕获连接异常：在业务代码里捕获ConnectionError，返回友好的错误响应，避免服务崩溃。
• 应用 shutdown 时关闭连接池：释放Redis连接资源，避免泄漏：

@app.on_event("shutdown")
async def shutdown_event():
    await pool.disconnect()

内容的提问来源于stack exchange，提问作者DurandA