必须使用 redis.asyncio.Redis；旧版 aioredis 已停更，安装只需 pip install redis，连接池须用 redis.asyncio.ConnectionPool，所有 I/O 操作需 await，避免新建实例而应全局复用，pubsub.listen() 需在独立协程中运行且不可重复调用。
asyncio 连 Redis 必须用 redis.asyncio.Redis，别碰 aioredis
旧版 aioredis 已归档停更，现在官方维护的异步 Redis 客户端就是 redis.asyncio（从 redis-py 4.2+ 开始内置）。用错包会导致 import 报错、连接不兼容或订阅收不到消息。
常见错误现象：ModuleNotFoundError: No module named 'aioredis' 或 AttributeError: 'Redis' object has no attribute 'publish'（其实是用了同步客户端却调了异步方法）。
- 安装只用
pip install redis，不用装aioredis -
from redis.asyncio import Redis是唯一正路；import aioredis是过时写法 - 连接池必须用
redis.asyncio.ConnectionPool，不能复用同步的redis.ConnectionPool
存取缓存：用 set() / get() 就行，但得 await
异步操作不是加个 async 就完事——所有 I/O 方法都得显式 await，否则返回的是 coroutine 对象，不是真实值。
使用场景：Web 请求中查用户 session、接口结果缓存、防重复提交的 token 校验。
立即学习“Python免费学习笔记（深入）”；
-
await redis.set("user:1001", "active", ex=3600)——ex是秒，不是毫秒 -
await redis.get("user:1001")返回bytes，不是 str，记得.decode()或用encoding="utf-8"初始化 client - 批量操作优先用
pipeline()：async with redis.pipeline() as pipe: await pipe.set(...).get(...).execute()
发布订阅：用 pubsub() + subscribe()，但别在协程里直接 listen()
pubsub().listen() 是个异步生成器，会一直阻塞等待消息。如果在 FastAPI 路由或普通协程里直接调它，整个事件循环就卡死。
正确做法是把它跑在独立任务里，或者用回调模式处理消息。
- 启动订阅要另起任务：
asyncio.create_task(pubsub.subscribe("channel_a"))不够，得配合pubsub.listen() - 推荐写法：
async for message in pubsub.listen(): ...，且这个for必须在单独的async def函数里运行 - 注意：一个
pubsub实例只能 listen 一次；重复调listen()会报RuntimeError: async generator already executing - 连接断开后不会自动重连，得自己捕获
ConnectionError并重建pubsub
连接管理：别每次请求都新建 Redis 实例
每个 Redis 实例背后默认带一个连接池，频繁创建销毁会耗尽 socket 资源，尤其在高并发 Web 场景下容易触发 ConnectionRefusedError 或超时。
性能影响明显：实测单机压测下，每请求 new client 比复用连接池慢 3–5 倍，且内存泄漏风险高。
- 全局复用一个
Redis实例（如 FastAPI 的app.state.redis） - 初始化时传
connection_pool更可控：pool = redis.asyncio.ConnectionPool.from_url("redis://localhost")，再Redis(connection_pool=pool) - 关闭服务前记得
await redis.close()和await pool.disconnect()，否则可能残留连接
最常被忽略的一点：Redis 异步客户端的 decode_responses=True 参数只对字符串命令生效（如 get、set），对 pubsub.listen() 返回的 message 字典里的 data 字段无效——它还是 bytes，得手动 decode。
