aioredis 1.x 已弃用，必须迁移到 redis.asyncio；需修改导入、显式设置 decode_responses、使用 async with pipeline、复用 connectionpool 并管理生命周期。
aioredis 1.x 已弃用，redis.asyncio 是唯一官方推荐的异步客户端
如果你还在用 aioredis==1.x（比如 from aioredis import create_redis），现在就得停——它早在 2021 年就停止维护，且不兼容 Redis 7+ 的新命令（如 FT.SEARCH、JSON.GET）。官方明确把 aioredis 2.x 重命名为 redis 包，并把异步模块拆进 redis.asyncio 子模块。不是“三选一”，而是：旧项目必须迁移到 redis.asyncio，新项目直接用它。
常见错误现象：ImportError: cannot import name 'create_redis' from 'aioredis'，或连接后调用 hgetall() 返回 bytes 而非 dict（因默认响应解析逻辑变更）。
- 迁移不是改 import 就完事：
aioredis1.x 的Redis实例是协程工厂，而redis.asyncio.Redis是可直接 await 的实例 -
decode_responses=True必须显式传入，否则所有字符串值仍为bytes -
connection_pool初始化方式不同：不再用ConnectionPool.from_url()，改用redis.asyncio.ConnectionPool.from_url()
从 redis-py 同步代码切到 redis.asyncio 的最小改动点
同步客户端 redis.Redis 和异步客户端 redis.asyncio.Redis API 高度一致，但所有方法都变成 awaitable。这不是语法糖替换，而是 I/O 模型切换，直接影响调用链。
使用场景：Django/Flask 中已有 redis.Redis(host='...') 单例，想在 FastAPI 或 asyncio Web 框架中复用逻辑。
立即学习“Python免费学习笔记（深入）”；
- 所有方法名不变，但必须加
await：把r.get("key")改成await r.get("key") -
pipeline必须用async with r.pipeline() as pipe:，不能用pipe.execute()，得用await pipe.execute() - 不支持同步阻塞操作：比如
blpop在redis.asyncio中叫blpop，但行为是异步等待，不能混进同步线程里调用 - 连接池复用要注意：
redis.asyncio.ConnectionPool不能和redis.ConnectionPool混用，否则报RuntimeError: There is no current event loop
redis.asyncio 连接管理的坑：别在每次请求里 new client
很多人以为 “async” 就该每个请求建一个 redis.asyncio.Redis 实例，结果压测时连接数暴涨、TIME_WAIT 堆积、甚至触发 Redis 的 maxclients 限制。
根本原因：redis.asyncio.Redis 默认启用连接池，但如果你每次都 Redis(host=...)，就等于新建一个独占连接池（含多个空闲连接），而不是复用。
- 正确做法：全局单例 + 连接池复用 —— 用
redis.asyncio.ConnectionPool显式创建，再传给Redis(connection_pool=pool) - 连接池参数要调：默认
max_connections=2**31（几乎无限），生产环境建议设为max_connections=50左右，配合应用并发量 - 注意生命周期：FastAPI 的
lifespan或 ASGIstartup/shutdown中显式await pool.disconnect()，否则进程退出时连接不释放 - 别用
from_url简写绕过池管理：redis.asyncio.Redis.from_url("redis://")看似方便，但每次调用都新建池，比手写还危险
升级后命令行为变化最易忽略的三点
API 表面兼容，但底层序列化、错误类型、返回结构有静默差异。线上出问题往往卡在这几处。
-
scan_iter返回的是 async generator，不能直接 for 循环，得async for key in r.scan_iter("prefix:*"): -
hgetall返回Mapping[str, str]（当decode_responses=True），不再是dict；如果原逻辑用了.keys()再转 list，没问题；但若用了dict.keys() & set(...)，可能因类型检查失败 - 错误类型变了：
redis.exceptions.ConnectionError仍是同步模块的，异步环境下实际抛的是redis.exceptions.ConnectionError（同名但模块路径不同），用except redis.exceptions.ConnectionError:会捕不到，得用except redis.exceptions.ConnectionError:—— 注意导入来源必须是import redis，不是import redis.asyncio
最复杂的其实是 pipeline + transaction 混合场景，比如 watch 后的条件执行，异步下 retry 逻辑要重写，不是加个 await 就能跑通。这点文档很少提，但线上事务失败率会上升。
