简介面向需要借助异步任务队列处理耗时操作的开发者这是一份基于缓存服务与轻量级任务队列库实现的任务队列最小示例。资源演示了将任务写入队列、由独立工作进程监听并消费的完整流程适用于快速理解分布式任务处理或作为项目脚手架。压缩包内共5个文件包括2个脚本源文件、1个编译后字节码文件、1个说明文档和1个版本控制忽略配置整体仅3KB结构简洁便于阅读。目前已有385人学习下载。核心代码涵盖任务生成与处理函数两大模块说明文档中补充了缓存服务启动、后台队列进程运行等环境准备要点针对依赖安装、本地队列调试等常见环节该示例也提供了可复用的参考命令适合初学者在本地快速验证也适合有经验的开发者快速集成。1. 一个 python redis 任务队列 demo「自备」这两个字才是关键python redis 任务队列 demo 类项目网上能搜出一堆但多数都把 Redis 当成不用关心的黑匣子。这个标题里最值得琢磨的是「自备」你得自己启动 Redis、自己调参数、自己面对积压和丢任务。恰恰是这份自备逼着你把生产、存储、消费三条链路彻底摸清楚。适合两类人想搞懂队列原理再决定要不要上重型框架的开发者以及需要不依赖重框架、能快速改造成内部工具的团队。这个 demo 给你的不是一个一键启动的黑盒而是一套你能逐行看懂、按需改造的队列底座。下面从队列模型讲起给出最小可运行代码再逐步加上确认机制、延迟任务和优先级最后交代我踩过的坑和收尾习惯。2. 队列模型先立住为什么 Redis 的 list 天生适合做任务队列任务队列要解决的核心问题只有一个让耗时的工作离开请求链路放到后台慢慢跑。Redis 提供字符串、列表、有序集合、流好几种数据结构偏偏是 list 成了任务队列的事实标准原因是它的左进右出天然是 FIFO还支持阻塞读取和原子转移三个特性组合起来正好覆盖任务队列的全部需求。2.1 生产、存储、消费任务队列的三个角色任何任务队列都能拆成三个角色生产者把任务描述序列化成字符串push 进队列Redis 的 list 作为存储层让任务在内存里排队消费者从另一头把任务取出来反序列化后执行。我在写 demo 时统一用 JSON 作为任务序列化格式字段固定三个id 是全局唯一标识type 是任务类型payload 放业务参数。不要用 pickle跨语言场景会让你后悔也不要把大字段直接塞进任务体文件路径、对象 ID 这类引用才是更合理的选择。任务在队列里流动的形态是字符串任何能被 json.dumps 处理的对象都可以入队。接口约定要在一开始定好否则后面加字段会牵扯所有生产者和消费者。一个经验是维护一张任务字段表写明每个字段的类型、必选还是可选、谁负责写入、谁负责读取避免两个服务各自往同一个任务里塞不兼容的数据。2.2 为什么不是 pub/sub也不是 Streams先把最容易走歪的路堵上。Redis 的 pub/sub 看起来也是「一边发一边收」但有个致命缺陷消息不带积压消费者不在线消息直接丢发布方毫无感知。任务队列要求至少处理一次pub/sub 给的是至多一次方向完全相反。部署时消费者重启几秒钟这一窗口里发的任务就全部蒸发。Streams 是功能更完整的方案有消费组、有 ACK、有消息回溯但代价是命令复杂度和代码量明显上升。对自备 Redis 的 demo 和内部工具来说list 是性价比最高的起点。三者的取舍我整理成一张表方案消息积压消费者离线确认机制命令复杂度pub/sub无直接丢弃收不到无低list有内存排队回来后能补需要自己实现低Streams有可追溯消费组可回溯内置 ACK较高判断标准很简单你需要的是一批任务被可靠地消费而不是广播被实时接收那就选 list。等到真正出现「谁消费了这条消息」的审计需求再考虑 Streams第 4 章会具体讲切换时机。2.3 最小可运行 demo生产者入队与消费者取任务先给一个能直接跑通的最小版本。前提是你已经自备 Redis 并启动了服务连接参数用默认的 127.0.0.1:6379。生产者和消费者分别是两个独立脚本。# producer.py生产者把任务序列化后入队 import json import redis # decode_responsesTrue 让读出来的 bytes 自动变 str省去处处 .decode() r redis.Redis(host127.0.0.1, port6379, db0, decode_responsesTrue) def enqueue(queue: str, task: dict) - None: message json.dumps(task, ensure_asciiFalse) r.lpush(queue, message) # 头插 if __name__ __main__: for i in range(20): enqueue(task_queue, {id: i, type: demo, content: fjob-{i}}) print(队列长度:, r.llen(task_queue))逻辑说明lpush 把消息插到 list 头部消费者从尾部取先入先出。ensure_asciiFalse 让中文以明文存进 Redis排查时可以直接用命令行查看原始内容不用猜 Unicode 转义。llen 用来确认消息确实进去了。# worker.py消费者取任务并执行 import json import redis r redis.Redis(host127.0.0.1, port6379, db0, decode_responsesTrue) def handle(task: dict) - None: # 替换成你要执行的真实业务 print(task[id], task[content]) while True: raw r.lpop(task_queue) # 尾部取出取不到返回 None if raw is None: break # demo 里队列空了就退出 task json.loads(raw) handle(task)逻辑说明lpop 取出即删队列为空时返回 None。注意方向lpush 配 rpop 才是先进先出写成 lpush 配 lpop 就成了后进先出任务顺序会反转。提示这版只是演示数据流不是能上生产的版本。lpop 没有确认机制worker 处理中途崩溃任务就没了第 3 章解决这两个问题。3. 从「能跑」到「敢用」阻塞读取、确认机制与幂等设计最小 demo 能跑但离敢用还差三步让消费者在队列为空时休眠而不是空转、让任务被认领而不是被删除、让重复执行不产生副作用。这三步做完队列才有最基本的可靠性。我见过太多直接把第一版 worker 丢上线的案例血泪经验写在后面第 5 章。3.1 BRPOP 与 BRPOPLPUSH阻塞读和「取出即转移」先看阻塞读取。BRPOP 是 lpop 的阻塞版本接受一个 key 和一个超时时间timeout 单位是秒传 0 表示永久阻塞任务不来就不返回。这个参数是队列空转问题的解药。# 演示 BRPOP 的返回值结构 item r.brpop(task_queue, timeout0) # 返回 (queue_name, value)例如 (btask_queue, b{id: 1}) queue_name, raw item task json.loads(raw)BRPOPLPUSH 更进一步把「从源队列尾部取出」和「插入目标队列头部」合成一个原子操作。在任务队列里的典型用法是源队列放待处理任务目标队列放处理中的任务。任务一旦被取出立刻进入 processing 列表相当于被某个 worker 认领。import json import redis import time r redis.Redis(host127.0.0.1, port6379, db0, decode_responsesTrue) while True: # 原子操作从 task_queue 尾部取出同时放入 task_processing 头部 raw r.brpoplpush(task_queue, task_processing, timeout0) if raw is None: continue task json.loads(raw) try: time.sleep(0.05) # 模拟业务耗时 print(完成:, task[id]) # 处理成功才从 processing 移除任务才算真正消失 r.lrem(task_processing, 1, raw) except Exception as exc: print(失败:, exc) # 失败放回主队列重试同时从 processing 清掉 r.lpush(task_queue, raw) r.lrem(task_processing, 1, raw)逻辑说明这段代码的核心是 lrem 这行。任务从 processing 队列被移除的唯一条件是处理成功异常路径是放回主队列。如果你漏掉这一行任务会一直堆在 processing 里表面上队列空了实际上全卡在处理中。参数说明timeout0 是生产环境标准设置lrem 第二个参数 1 表示最多删除一条匹配项raw 是完整字符串正常情况下唯一但用 1 比用 0删除全部匹配项更安全避免重复任务被连带误删。3.2 确认机制任务被「认领」而不是被「删除」为什么要多引入一个 processing 队列因为原始 BRPOP 是取出即删任务离开 Redis 的那一刻Redis 就认为它已交付。如果 worker 在处理到一半时崩溃这个任务没有备份直接消失。BRPOPLPUSH 把取出和标记认领合成一步任务离开主队列时出现在 processing 里。worker 崩溃也没关系任务还躺在 processing只是没有归属。你需要一个恢复程序定期扫描 processing把停留过久的任务重新放回主队列。# worker 认领任务后刷新 processing 里该任务的 picked_at 时间戳 def mark_picked(r, raw: str) - None: task json.loads(raw) task[picked_at] time.time() new_raw json.dumps(task, ensure_asciiFalse) r.lrem(task_processing, 1, raw) # 只删一条旧记录 r.lpush(task_processing, new_raw) # 放回带时间戳的版本# recover_orphan.py找回卡在 processing 里的孤儿任务 import json import time import redis r redis.Redis(host127.0.0.1, port6379, db0, decode_responsesTrue) def recover_orphans(r, processing_queue: str, task_queue: str, timeout: int 600): for raw in r.lrange(processing_queue, 0, -1): task json.loads(raw) picked_at task.get(picked_at, 0) if time.time() - picked_at timeout: task[picked_at] time.time() r.lpush(task_queue, json.dumps(task, ensure_asciiFalse)) r.lrem(processing_queue, 1, raw) if __name__ __main__: while True: recover_orphans(r, task_processing, task_queue, timeout600) time.sleep(60)逻辑说明恢复逻辑依赖任务体里的 picked_at 时间戳worker 认领任务后通过 mark_picked 刷新。恢复程序每 60 秒跑一轮把超时未完成的任务重新投递。这里给的是定时任务基础版真正的生产实现会用 Lua 脚本把刷新合并成原子操作但原理一致。参数说明timeout 的含义是「任务在 processing 停留超过 N 秒就视为孤儿」取值必须大于单任务最长处理时间否则运行中的慢任务会被误伤。常见做法是按单任务耗时上限的两倍设比如任务最长跑 5 分钟timeout 给 600 秒。3.3 幂等处理重复执行不可怕可怕的是没有去重有了确认机制和重试任务就可能被执行两次一次正常处理一次失败重试。对打印日志这种任务无所谓但涉及扣款、发消息、写库存的操作重复执行就是事故。常见做法是任务体带全局唯一 id消费端在处理前做一次去重检查。Redis 的 SETNX 刚好能把检查加写入做成原子操作。def already_done(r, task_id: str, ttl: int 86400) - bool: # set nx 语义只有 key 不存在时才写入成功 result r.set(fdone:{task_id}, 1, nxTrue, exttl) return result is None逻辑说明第一次处理时 key 不存在set 成功返回 Truealready_done 返回 False放行业务第二次处理时 key 已存在set 被 nx 拦下返回 Nonealready_done 返回 True跳过。ex 控制去重窗口必须覆盖最大重试周期。参数说明ttl 不建议设永久占内存一般设成任务最长重试周期的 1.5 到 2 倍。一个任务重试 3 次、每次间隔 5 分钟ttl 给 1 小时绰绰有余。注意判重必须在业务副作用之前执行写在 try 外面否则去重形同虚设。4. 进阶形态延迟任务、优先级与 Streams 的选型边界基础版能用之后很快会碰到两个新需求任务要延迟执行比如 10 分钟后再发通知以及部分任务优先级更高比如人工审核优先于批量生成。这两个需求用 Redis 原生数据结构都能解决关键是方案选对。4.1 延迟任务用有序集合做时间轮有序集合 zset 的 score 天然适合存时间戳。入队时把 score 设为期望执行时间点一个独立的后台循环负责扫描到期任务到期后转投主队列本质是一个最小的时间轮。import json import time def enqueue_delayed(r, delay_queue: str, task: dict, delay_seconds: int) - None: # score 是期望执行时间的时间戳 score time.time() delay_seconds r.zadd(delay_queue, {json.dumps(task, ensure_asciiFalse): score})# scheduler.py把到期的延迟任务转投到主队列 def schedule_due(r, delay_queue: str, task_queue: str, batch: int 100) - int: now time.time() # 取 score 在 [0, now] 范围内的任务最多 batch 条 due r.zrangebyscore(delay_queue, 0, now, start0, numbatch) for raw in due: r.lpush(task_queue, raw) # 先投递 r.zrem(delay_queue, raw) # 后删除 return len(due) while True: moved schedule_due(r, task_delay, task_queue) if moved 0: time.sleep(1)逻辑说明先 lpush 再 zrem 的顺序是刻意的。先删再投递投递时崩溃任务就彻底丢失先投再删最多重复投递一次配合第 3 章的幂等机制可以兜住。参数说明batch 控制在单次扫描中转移的数量避免一次取出几十万条把网络打满扫描间隔 1 秒意味着延迟精度是秒级业务要求毫秒级就要压到 0.1 秒以下但空转成本会明显上升一般不建议。4.2 优先级多 list 轮询还是单 list 内排序优先级最直观的做法是给任务体加 priority 字段消费者取到后再排序。但 list 没有按字段查询和排序的能力你得扫完整个队列才能找到最高优先级任务任务量一大就是灾难复杂度直接变成 O(N)。我一般用多 list 方案按优先级建三个队列BRPOP 支持一次监听多个 key从左到右取第一个非空队列的元素。高优先级队列只要有货worker 永远先消费它。raw r.brpop([task_queue:high, task_queue:normal, task_queue:low], timeout0) if raw is None: continue queue_name, value raw # queue_name 用于区分任务来源与优先级 task json.loads(value)逻辑说明BRPOP 监听多个 key 时返回值第一个元素是实际命中的队列名第二个才是任务内容。这个队列名还可以用来统计各优先级的处理量方便观测高优任务是否堆积。这个方案的问题在于低优先级可能被饿死高优先级任务源源不断时low 队列永远没机会被取到。解决思路两条一是给 low 队列的任务也设最大等待时间入队时记录时间戳到期人工提升优先级二是接受「高优插队」的业务语义多数内部工具其实能接受后者。4.3 换 Streams 的时机两个方案放到一张表里看list 方案不是银弹。当任务需要多个消费组各自独立消费、需要追溯每条消息的处理流水、需要内置确认而不是自己维护 processing 队列时就该考虑 Redis Streams。两者关键差异能力list 方案Streams 方案入队出队命令LPUSH / BRPOPLPUSHXADD / XREADGROUP消费确认自己用 processing 队列模拟XACK 原生支持消费组无worker 竞争式消费多组隔离消费消息回溯不可查已处理消息Pending Entries List 可查代码量demo 级几十行一个消费组就要多组参数我的建议是内部工具、任务积压在百万条以内、不需要回溯list 完全够用少即是多。一旦出现「谁消费了这条消息」的审计需求换成 Streams 的代价主要在学习成本和命令复杂度功能上它确实省掉了自己维护确认机制的功夫但要重写一遍 worker migration 前最好把 4.1 的延迟任务和 4.2 的优先级方案一并评估进去。5. Redis 任务队列常见问题排查五个翻车现场与修复这一章是我最想让你跳过、又不希望你跳过的部分。下面五条全部来自真实运行时的排查过程每一条我都踩过按「现象、原因、解决」写清楚每条附一个排查命令。5.1 现象任务莫名丢失现象夜间批量任务第二天看少了一批日志里没有任何异常Redis 队列也空了。原因worker 用的 BRPOP 取出即删处理到一半进程被重启发布、内存溢出被杀任务不在地球上也没有备份。解决换成 BRPOPLPUSH processing 队列让任务被认领而不是被删除再把 3.2 的孤儿恢复程序挂上。同时检查自备 Redis 的持久化配置默认 RDB 快照在极端场景会丢数据任务重要就开 AOF everysec。排查时先看 Redis 日志里有没有启动失败或快照失败记录再确认 worker 进程的退出时间是否与丢任务时间吻合。5.2 现象worker 全部空转连接数飙高现象info clients 显示大量空闲连接Redis CPU 不高但连接数持续增长新任务进来时响应变慢。原因BRPOP 的 timeout 设成了 1 到 3 秒。队列一空所有 worker 每秒醒来重新发一次请求worker 越多空转请求越密集。解决生产环境 timeout 设 0永久阻塞。如果担心长连接被网络设备闲置回收设 30 到 60 秒并配合心跳保活而不是设成秒级。排查用 redis-cli 执行 info clients 看连接分布以及用 MONITOR 短暂观察是否有大量重复的 BRPOP 请求。5.3 现象同一任务被处理了两次现象数据库出现重复记录某个操作明明只在入队时调了一次却产生了两次副作用。原因失败重试直接 lpush 回主队列没有任何去重手段。at-least-once 语义下重复是必然不重复才是意外。解决任务体带唯一 id消费端先 setnx 判重再执行业务。这里有个细节判重要在业务副作用之前完成并且 ttl 要覆盖最长重试周期。排查时用 SCAN 匹配 done:* 前缀的 key看数量是否明显少于任务总数能快速判断是不是漏了幂等。5.4 现象Redis 内存被打满现象used_memory 曲线陡增写入开始报 OOM 错误队列 key 里全是过期的历史任务。原因两件事叠加一是入队没有上限生产者不关心消费速度无脑推二是任务体里塞了大字段比如图片 base64、整份报表 JSON。解决入队前检查 llen超过阈值比如 5 万个就拒收并告警任务体只放引用不放数据体大对象给 ID 让消费者自己去取按业务给队列 key 加日期后缀历史队列整体设置过期时间。排查命令是 info memory 看 used_memory 和 rdb_last_cow_size再用 MEMORY USAGE task_queue 看单个 key 占了多少。5.5 现象队列积压但 Redis CPU 很低现象llen 一直涨Redis 的命令处理量很低worker 机器的 CPU 也不高任务就是消费不动。原因消费者卡在后端 IO 上比如数据库慢查询、外部接口超时。worker 串行处理一个任务卡 10 秒后面所有任务跟着排队Redis 本身完全不忙。解决把 worker 拆成多进程每个进程独立阻塞读取给业务处理加超时熔断监控 p99 处理耗时而不是只盯 llen。排查时在 worker 里临时打日志记录每任务的开始和结束时间拉出来看耗时分布确认瓶颈在下游而不是队列本身。6. 收尾技巧优雅退出、队列监控和死信兜底三件事加起来不到三十行代码但能让这个 demo 变成真正可以托管的东西。第一件是优雅退出。worker 收到 SIGTERM 时应该把手头的任务处理完再退出而不是被强制终止导致任务卡在 processing。import signal import json import redis r redis.Redis(host127.0.0.1, port6379, db0, decode_responsesTrue) running True def stop(signum, frame): global running running False signal.signal(signal.SIGTERM, stop) signal.signal(signal.SIGINT, stop) while running: raw r.brpoplpush(task_queue, task_processing, timeout1) if raw is None: continue task json.loads(raw) try: # 业务处理代码 r.lrem(task_processing, 1, raw) except Exception as exc: print(处理失败:, exc)这里的 timeout1 是让阻塞调用每秒醒来一次检查 running 标志代价是每秒多一次空请求换来的却是可控的停机流程。第二件是队列监控。一个每 30 秒跑一次的巡检循环输出三个关键数字待处理、处理中、死信。积压和异常都能第一时间发现。import time while True: pending r.llen(task_queue) processing r.llen(task_processing) dead r.llen(task_dead) print(f待处理{pending} 处理中{processing} 死信{dead}, flushTrue) time.sleep(30)第三件是死信兜底也是我最想强调的一点给任务体加一个 retry 计数器重试超过上限的任务不再放回主队列而是进死信队列避免一个永远失败的任务堵死整条业务。MAX_RETRY 3 def consume(r): raw r.brpoplpush(task_queue, task_processing, timeout0) task json.loads(raw) try: handle_task(task) # 业务处理 r.lrem(task_processing, 1, raw) except Exception as exc: task[retry] task.get(retry, 0) 1 task[last_error] str(exc)[:200] new_raw json.dumps(task, ensure_asciiFalse) if task[retry] MAX_RETRY: r.lpush(task_dead, new_raw) # 死信留给人或定时任务处理 else: r.lpush(task_queue, new_raw) # 有上限的重试 r.lrem(task_processing, 1, raw)我第一次写队列 demo 时图省事失败任务直接放回队首结果一个永远失败的任务把整条队列堵死正常任务全部延迟。后来加上 retry 计数和死信队列才消停。现在哪怕是内部小工具我也会先把优雅退出和死信兜底做进去这两样加起来不到二十行但能救你很多次。希望帮到你。本文还有配套的精品资源点击获取
