一、Redis 队列的本质
Redis 队列通常使用 List 数据结构实现。
Redis 中所有数据结构都是以 key-value 存储的。
结构：
Redis
└── DB0
└── key = "queue"
type = list
value = [msg3, msg2, msg1]
Python 连接 Redis：
import redis
r = redis.Redis(host="localhost", port=6379, db=0)
说明：
- db=0 表示使用 Redis 的 0号数据库
- Redis 默认有 16个数据库 (0–15)
二、LPUSH 命令
1 含义
r.lpush("queue", msg)
等价 Redis 命令：
LPUSH queue msg
作用：向 List 左侧插入元素
示例：
LPUSH queue msg1
LPUSH queue msg2
LPUSH queue msg3
结果：
queue = [msg3, msg2, msg1]
特点：
|
特性 |
说明 |
|
时间复杂度 |
O(1) |
|
插入位置 |
左侧 |
|
是否线程安全 |
是（Redis 单线程） |
|
常见角色 |
Producer（生产者） |
三、BRPOP 命令
1 含义
msg = r.brpop("queue", 0)
等价 Redis 命令：
BRPOP queue 0
作用：从 List 右侧取出元素，如果没有元素就阻塞等待
示例：
当前队列：
[msg3, msg2, msg1]
执行：
BRPOP queue
返回：
msg1
队列变成：
[msg3, msg2]
2 返回值结构
Redis 返回：
(queue_name, value)
Python 返回：
(b'queue', b'msg1')
所以代码写：
msg[1].decode()
3 timeout 参数
BRPOP key timeout
示例：
BRPOP queue 0
含义：
|
timeout |
行为 |
|
0 |
永久阻塞 |
|
>0 |
等待指定秒 |
例如：
BRPOP queue 10
等待10秒。
四、为什么使用 BRPOP
如果使用：
RPOP
代码可能是：
while True:
msg = r.rpop("queue")
问题：
当队列为空时：
CPU疯狂轮询
浪费 CPU。
而 BRPOP：
如果队列为空
Redis会阻塞连接
当 Producer 写入消息：
LPUSH queue msg
Redis 立即返回。
优点：
- 不浪费 CPU
- 实时消费
五、Redis 队列模型
基本架构：
Producer -> Redis Queue -> Consumer
示意：
Producer
↓
LPUSH queue
↓
Redis List
↓
BRPOP queue
↓
Consumer
六、Redis 队列存储的数据类型
Redis 本质存储：
binary-safe string (bytes)
所以可以存：
|
数据 |
是否推荐 |
说明 |
|
字符串 |
✅ |
最常见 |
|
JSON |
✅ |
最推荐 |
|
数字 |
✅ |
自动转换 |
|
二进制 |
⚠️ |
可以但不推荐 |
|
pickle对象 |
⚠️ |
不推荐 |
不能存：
|
数据 |
原因 |
|
函数 |
内存对象 |
|
进程 |
OS对象 |
|
协程 |
Python运行时对象 |
|
线程 |
Python运行时对象 |
推荐方式：JSON任务
示例：
import json
task = {
"task_id": 1,
"type": "image_embedding",
"path": "/data/a.png"
}
r.lpush("queue", json.dumps(task))
消费：
msg = r.brpop("queue", 0)
task = json.loads(msg[1])
七、为什么不直接存图片
虽然可以：
with open("a.png","rb") as f:
r.lpush("queue", f.read())
但生产环境不推荐。
原因：Redis 是：内存数据库
图片：体积大
正确做法：只存 路径或URL
{
"task_type": "image_embedding",
"image_path": "/data/a.png"
}
八、BRPOP 删除消息制
BRPOP 的行为是：
取出元素 + 删除元素
示例：
当前队列：
queue = [task3, task2, task1]
执行：
BRPOP queue
返回：
task1
队列变为：
[task3, task2]
九、为什么会丢消息
假设：
queue = [task1]
Worker：
BRPOP queue
Redis：
queue = []
Worker 开始处理：
process(task1)
如果此时：
worker崩溃
服务器宕机
程序异常
结果：
task1 已经从 Redis 删除
但没有处理成功
所以：
消息丢失
十、生产环境常见用途
1 异步任务
例如：
用户上传图片：
API
↓
Redis Queue
↓
Worker
↓
图片处理
API 不需要等待处理完成。
2 日志处理
多个服务
↓
Redis Queue
↓
日志消费者
↓
ES / Kafka
3 削峰填谷
高并发请求：
10000 req
写入队列：
Redis Queue
worker 慢慢处理：
worker1
worker2
worker3
十一、Redis List 队列的问题
主要问题：
1 没有 ACK
无法确认消息是否处理成功。
2 没有重试机制
任务失败无法自动重试。
3 容易丢消息
因为： BRPOP = 删除消息
十二、改进方案
方案1：BRPOPLPUSH
命令：
BRPOPLPUSH queue processing
流程：
queue -> processing
任务完成：
LREM processing task
好处：任务不会丢失。
十三、简单完整示例
Producer
import redis
r = redis.Redis(host="localhost", port=6379, db=0)
for i in range(5):
r.lpush("queue", f"task-{i}")
Consumer
import redis
r = redis.Redis(host="localhost", port=6379, db=0)
while True:
msg = r.brpop("queue", 0)
print("处理任务:", msg[1].decode())
