不能只用rpush+lpop，因其非阻塞导致轮询空转、任务崩溃即丢失、多消费者竞争引发重复或漏消费；可靠方案需brpop阻塞等待+唯一task_id标记+setex防重机制。
直接用 redis-py 的 lpush/rpop 就能跑起来，但不加阻塞和重试机制的任务队列，上线后大概率丢任务。
为什么不能只用 rpush + lpop
这是新手最常写的组合，看着像入队出队，实际在高并发或消费者短暂离线时会立刻暴露问题：
-
lpop是非阻塞的：没任务就立刻返回None，你得自己写轮询，CPU 白耗、延迟高 - 任务被
lpop取出但处理中途崩溃，这个任务就永远消失了（无确认、无重入） - 多个消费者同时
lpop同一个 key，Redis 不保证原子性下的竞争行为，可能重复消费或漏消费
真正可用的最小闭环，得用 brpop（阻塞右弹出）+ lpush（失败回退）+ 唯一任务 ID。
用 brpop 实现带超时的可靠消费
brpop 会让客户端在列表为空时挂起，直到有新元素或超时，避免空转。它天然适合“等待任务”的场景：
import redis
r = redis.Redis()
# 阻塞最多 5 秒，从 'task_queue' 右端取一个任务
result = r.brpop(['task_queue'], timeout=5)
if result:
queue_name, task_data = result
# 处理 task_data，成功则不管；失败则 lpush 回去
try:
process_task(task_data)
except Exception:
r.lpush('task_queue', task_data) # 放回队首，下次优先重试
注意：brpop 参数是列表（支持多队列优先级），且它操作的是**右端**，所以入队必须用 rpush 才能保持 FIFO；如果误用 lpush 入队 + brpop 出队，实际变成 LIFO。
如何避免任务重复执行或丢失
核心是把“取出”和“确认完成”拆成两步，并引入中间状态。简易但有效的做法：
- 入队时生成唯一
task_id（如uuid4().hex），整个任务数据用 JSON 存为字符串 - 消费者用
brpop取出后，立刻用setex写一个processing:{task_id}标记，过期时间设为预计处理时长的 2–3 倍 - 处理完再
delete这个标记；若处理失败或超时，另一个消费者发现该task_id已存在且过期，就认为上一个消费者已死，可安全重试 - 不依赖 Redis 事务也能防重：只要每次检查
exists processing:{task_id}成功才继续，就能跳过正在处理中的任务
这比单纯靠 lpush 回退更稳——回退只能解决崩溃，解决不了进程卡住、网络分区、K8s Pod 被杀等静默故障。
别忽略连接和序列化的实际坑
本地开发跑通不等于线上可用：
- 用
redis.Redis(decode_responses=True)，否则brpop返回的是bytes，JSON 解析直接报TypeError - 生产环境必须配
socket_connect_timeout和socket_timeout（建议都设 3 秒），不然 Redis 暂时不可达会导致整个 worker 卡死 - 任务数据别塞大文件或二进制——Redis 是内存数据库，单个 value 超 1MB 就该警惕；图片、PDF 等存对象存储，只传 URL
-
brpop的timeout别设 0（永久阻塞），K8s readiness probe 或 SIGTERM 无法优雅中断，worker 会 hang 住
真正难的不是搭出队列，而是让每条任务在断电、重启、扩容缩容下仍可追溯、可重放、不重复——这些细节藏在 setex 过期策略、decode_responses 开关、超时数字里，而不是某行代码里。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
