必须手动配置asyncio+redis-py连接池，否则同步调用会阻塞事件循环；应使用redis.asyncio.Redis配合显式创建的ConnectionPool，避免混用同步/异步客户端，高频任务需用pipeline批量处理，消费者推荐brpoplpush实现可靠消费，并为每个consumer task添加异常捕获与重连逻辑。
asyncio + redis-py 连接池必须手动配置，否则会阻塞事件循环
默认的 redis.Redis() 实例是同步阻塞的，哪怕你在 async def 函数里调用 get() 或 lpush()，也会让整个 event loop 卡住。这不是 asyncio 不支持 Redis，而是 redis-py 默认没走异步路径。
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 用
aioredis（v2.x）或redis-pyv4.2+ 的Redis.from_url(..., decode_responses=True)配合connection_class=redis.asyncio.Connection - 连接池要显式创建：
pool = redis.asyncio.ConnectionPool.from_url("redis://localhost")，再传给Redis(connection_pool=pool) - 别在协程里混用
redis.Redis()和redis.asyncio.Redis()，类型不兼容，await会直接报TypeError: object XX can't be used in 'await' expression
任务入队不能直接 await lpush，得用 pipeline 或 batch 提升吞吐
高频发任务时，逐条 await redis.lpush("queue:tasks", payload) 会产生大量网络往返，延迟飙升，实际 QPS 可能不到 200。这不是 Redis 慢，是单命令 round-trip 拖垮了 asyncio 的并发优势。
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 用
pipeline批量入队：async with redis.pipeline() as pipe: pipe.lpush("queue:tasks", *payloads); await pipe.execute() - 如果任务来自 HTTP 请求体，优先在业务层做简单合并（比如每 10 条攒一次），而不是无脑单条 push
- 注意
pipeline.execute()是原子的，但失败时整个 batch 回滚，需按需加重试逻辑
消费者用 brpoplpush 实现“带确认”的可靠消费
纯 lpop 拿到任务后如果 worker 崩溃，任务就丢了；用 rpoplpush 中转到 processing 列表再处理完删，又得手动维护状态。更稳的做法是用 brpoplpush —— 它能阻塞等待、自动转移、还能设超时。
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 启动消费者时执行：
await redis.brpoplpush("queue:tasks", "queue:processing", timeout=30) - 拿到任务后立刻处理，成功则
await redis.lrem("queue:processing", 1, task_id)；失败则await redis.rpush("queue:failed", task_data) - 注意
brpoplpush的 timeout 参数必须设（比如 30），否则消费者协程会永久挂起，无法响应 cancel 或 shutdown
asyncio.create_task 启动消费者时，务必加异常捕获和日志
一个没包 try/except 的消费者协程崩溃后，create_task 不会抛错到主 loop，也不会终止进程，只会静默消失——你看着服务在跑，其实早没在消费了。
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 每个消费者任务都套一层 wrapper：
async def run_consumer(): try: await consumer_loop() except Exception as e: logger.error("consumer died: %s", e, exc_info=True) - 别用
asyncio.gather(*tasks, return_exceptions=True)掩盖问题，它会让错误“吞掉”，不利于定位 - 检查 Redis 连接断开时的报错：常见
ConnectionClosedError或TimeoutError，需要在 wrapper 里重连并重启消费循环
真正难的不是写几个 await，是让每个环节在出错、超时、重连、积压时都有明确行为。比如 brpoplpush 超时后要不要补 log，pipeline 失败后要不要降级为单条重试——这些细节不写进代码，队列就只是个好看的玩具。
