一、Redis 队列的本质Redis 队列通常使用List 数据结构实现。Redis 中所有数据结构都是以key-value存储的。结构Redis └── DB0 └── key queue type list value [msg3, msg2, msg1]Python 连接 Redisimport redis r redis.Redis(hostlocalhost, port6379, db0)说明db0表示使用 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 返回(bqueue, bmsg1)所以代码写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 msgRedis 立即返回。优点不浪费 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]WorkerBRPOP queueRedisqueue []Worker 开始处理process(task1)如果此时worker崩溃 服务器宕机 程序异常结果task1 已经从 Redis 删除 但没有处理成功所以消息丢失十、生产环境常见用途1 异步任务例如用户上传图片API ↓ Redis Queue ↓ Worker ↓ 图片处理API 不需要等待处理完成。2 日志处理多个服务 ↓ Redis Queue ↓ 日志消费者 ↓ ES / Kafka3 削峰填谷高并发请求10000 req写入队列Redis Queueworker 慢慢处理worker1 worker2 worker3十一、Redis List 队列的问题主要问题1 没有 ACK无法确认消息是否处理成功。2 没有重试机制任务失败无法自动重试。3 容易丢消息因为 BRPOP 删除消息十二、改进方案方案1BRPOPLPUSH命令BRPOPLPUSH queue processing流程queue - processing任务完成LREM processing task好处任务不会丢失。十三、简单完整示例Producerimport redis r redis.Redis(hostlocalhost, port6379, db0) for i in range(5): r.lpush(queue, ftask-{i})Consumerimport redis r redis.Redis(hostlocalhost, port6379, db0) while True: msg r.brpop(queue, 0) print(处理任务:, msg[1].decode())
Redis 队列学习笔记（List Queue）
一、Redis 队列的本质Redis 队列通常使用List 数据结构实现。Redis 中所有数据结构都是以key-value存储的。结构Redis └── DB0 └── key queue type list value [msg3, msg2, msg1]Python 连接 Redisimport redis r redis.Redis(hostlocalhost, port6379, db0)说明db0表示使用 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 返回(bqueue, bmsg1)所以代码写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 msgRedis 立即返回。优点不浪费 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]WorkerBRPOP queueRedisqueue []Worker 开始处理process(task1)如果此时worker崩溃 服务器宕机 程序异常结果task1 已经从 Redis 删除 但没有处理成功所以消息丢失十、生产环境常见用途1 异步任务例如用户上传图片API ↓ Redis Queue ↓ Worker ↓ 图片处理API 不需要等待处理完成。2 日志处理多个服务 ↓ Redis Queue ↓ 日志消费者 ↓ ES / Kafka3 削峰填谷高并发请求10000 req写入队列Redis Queueworker 慢慢处理worker1 worker2 worker3十一、Redis List 队列的问题主要问题1 没有 ACK无法确认消息是否处理成功。2 没有重试机制任务失败无法自动重试。3 容易丢消息因为 BRPOP 删除消息十二、改进方案方案1BRPOPLPUSH命令BRPOPLPUSH queue processing流程queue - processing任务完成LREM processing task好处任务不会丢失。十三、简单完整示例Producerimport redis r redis.Redis(hostlocalhost, port6379, db0) for i in range(5): r.lpush(queue, ftask-{i})Consumerimport redis r redis.Redis(hostlocalhost, port6379, db0) while True: msg r.brpop(queue, 0) print(处理任务:, msg[1].decode())
