redis.asyncio.Redis（>=4.2.0）是Python 3.7+下最稳妥选择，需卸载旧版aioredis/aredis，用BRPOPLPUSH实现阻塞队列，配大连接池并保障业务幂等性。
asyncio + aioredis 是当前最稳妥的选择
Python 3.7+ 环境下，aioredis（v2.x）已整合进 redis-py 本身，直接用 redis.asyncio.Redis 即可，无需额外安装旧版包。别再用已弃用的 aioredis==1.x 或 aredis，它们不兼容 Python 3.9+ 且无维护。
常见错误是 pip install aioredis 后仍报 AttributeError: module 'aioredis' has no attribute 'from_url'——本质是版本错配。正确做法是：
- 卸载旧包：
pip uninstall aioredis aredis - 安装新版：
pip install redis>=4.2.0rc1（正式版 4.2.0+ 已稳定支持 async） - 连接方式必须用
redis.asyncio.Redis.from_url(...)，不是aioredis.from_url
队列 push/pop 必须用 LIST + BRPOPLPUSH 避免轮询
别在 asyncio 里写 while True + sleep(0.1) 去轮询 lpop，这会吃光 event loop 资源。Redis 原生的阻塞式命令才是异步队列核心。
推荐组合：LPUSH 入队，BRPOPLPUSH 出队并暂存到 processing list（防失败丢失），处理完再 LREM 或 LTRIM 清理。示例关键逻辑：
import asyncio
import redis.asyncio as redis
<p>r = redis.Redis.from_url("redis://localhost")</p><p>async def enqueue(task_data: str):
await r.lpush("queue:main", task_data)</p><p>async def worker():
while True:</p><h1>阻塞等待最多 5 秒，返回 (src, dst, item) 或 None</h1><pre class="brush:python;toolbar:false;"> res = await r.brpoplpush("queue:main", "queue:processing", timeout=5)
if res is None:
continue
try:
await process_task(res)
await r.lrem("queue:processing", 1, res) # 成功后移除
except Exception:
# 失败时可重放或打日志，不删 processing 中的项
pass
突发流量下连接池配置不当会导致 TimeoutError
redis.asyncio.Redis 默认连接池最大连接数是 10，高并发时大量协程卡在 acquire 上，抛出 TimeoutError: Unable to acquire connection，不是 Redis 挂了，是客户端池满了。
必须显式调大连接池参数：
-
max_connections=100（根据 uvicorn workers 数 × 平均并发协程数预估） -
min_idle_connections=10（避免频繁建连开销） -
retry_on_timeout=True（网络抖动时自动重试，配合socket_connect_timeout=2）
完整初始化示例：redis.Redis.from_url("redis://localhost", max_connections=100, min_idle_connections=10, retry_on_timeout=True)
任务幂等性必须由业务层保证，Redis 不负责去重
BRPOPLPUSH 本身不保证 exactly-once：worker 进程崩溃、超时未 ack、网络断开都可能导致同一条消息被多次消费。Redis 只提供原子移动能力，不提供 ACK 机制。
应对方案只能靠业务编码：
- 每个 task 带唯一
task_id（如 UUID v4），入库前先SETNX task_id:done 1 EX 3600 - 使用 Lua 脚本封装“检查 + 写入 + 消费”三步为原子操作
- 避免依赖
DEL+SET两步，中间可能 crash 导致状态不一致
别指望 redis-streams 自动解决——它需要额外维护 consumer group 和 pending entries，复杂度远高于 LIST 场景，且在突发流量下 memory usage 更难控。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
