redis.asyncio 是 redis-py 官方异步模块,推荐新项目使用;需显式设置 decode_responses=True,用 async with 或 lifespan 管理连接池,避免手动 close、混用连接池方式及 ex=0 误用。

aioredis 已在 2023 年正式归并入 redis-py,当前主流做法是直接使用 redis.asyncio ——它不是替代品,而是官方维护的异步模块。强行用旧版 aioredis==1.3.1 会遇到连接池不兼容、无自动重连、文档缺失等问题,不建议新项目采用。
用 redis.asyncio.Redis 替代 aioredis.from_url
新版 redis.asyncio.Redis 接口完全兼容同步版,但所有方法返回协程,必须 await。它默认启用连接池,且支持 retry_on_timeout 和健康检查。
-
from_url依然可用,但应导入自redis.asyncio:from redis.asyncio import Redis;Redis.from_url("redis://localhost") -
decode_responses=True必须显式设置,否则get()返回bytes,容易在 JSON 解析或字符串比较时出错 - 不要手动调
close()后再wait_closed()——推荐用async with上下文管理器,它会自动释放连接池资源 - 若需复用连接池(如 FastAPI 依赖注入),应将
Redis实例设为全局单例或通过lifespan管理生命周期,避免每次请求新建实例
ConnectionPool 配置影响重连与并发表现
连接池不是“开了就稳”,关键参数直接影响故障恢复能力和吞吐上限:
-
max_connections=20是常见起点,但高并发场景下需结合redis-cli --latency和应用 QPS 调整;设太小会导致ConnectionError: No connection available -
retry_on_timeout=True+socket_connect_timeout=3+socket_timeout=5组合,能让一次GET在网络抖动时自动重试而非立即失败 -
health_check_interval=30开启后,空闲连接每 30 秒发一次PING,可提前发现断连,避免首次请求卡顿 - 别混用
from_url和手动构造ConnectionPool:前者内部已封装池逻辑,后者需自行管理aclose(),易漏掉
await redis.get() 卡住几秒才报错?这是连接失效的典型信号
新版 redis.asyncio.Redis 不会自动重建底层 TCP 连接。当 Redis 重启或网络中断后,下一次 await redis.get(...) 可能阻塞在 send_command 阶段,直到 socket_timeout 触发才抛 ConnectionError。
- 最轻量的兜底方式:在业务逻辑前加健康检查
await redis.ping(),捕获ConnectionError或TimeoutError后触发服务降级或告警 - 不要依赖
async with redis.client()——这个写法只对Pipeline有效,对普通Redis实例会报AttributeError: __aenter__ - 批量操作优先走
mget/mset,它们复用同一连接,比循环await get快 3–5 倍,也减少事件循环调度压力 -
ex=0不是“永不过期”,而是立刻过期;要设永久键,用ex=None
FastAPI 中正确注入 Redis 实例的姿势
在 lifespan 中初始化并关闭,比用 Depends 每次创建更安全:
立即学习“Python免费学习笔记(深入)”;
from fastapi import FastAPI, Depends
from redis.asyncio import Redis
redis_client: Redis = None
async def init_redis():
global redis_client
redis_client = Redis.from_url(
"redis://localhost:6379/0",
decode_responses=True,
health_check_interval=30,
retry_on_timeout=True,
)
async def close_redis():
if redis_client:
await redis_client.aclose()
app = FastAPI(lifespan=lambda app: (init_redis(), close_redis()))
- 避免在路由函数里写
redis = Redis.from_url(...)——连接池不会被复用,还可能引发文件描述符耗尽 - 如果用
Depends注入,必须确保依赖函数返回的是同一个实例,而非每次都 new 一个 - 测试环境可加
encoding="utf-8"显式声明,避免某些 Redis 版本返回 raw bytes 导致decode_responses=True失效
aclose() 必须被 await,且不能放在 finally 块里靠异常触发——它得在应用退出前主动调用。没关干净的连接池,会在进程退出时留下 ResourceWarning,长期运行的服务可能出现连接泄漏。


















