简介一套面向 Python 开发者的 Redis 任务队列入门示例聚焦用 RQ 将耗时任务从主流程剥离适合想理解生产者、队列与 worker 协作方式的初中级读者。压缩包共 5 个文件以 2 个 Python 脚本为核心分别演示任务提交与处理函数组织另含 README 说明、编译缓存及 gitignore 配置整体约 3KB体量轻便便于快速导入本地环境验证。内容围绕 Redis 服务、rqworker 启动及任务入队流程展开读者可据此观察任务创建、投递、消费的完整链路并参考示例调整队列名称、连接地址与处理逻辑README 还提供依赖安装和运行入口适合作为后端异步任务、爬虫调度或后台作业处理的原型参考。已有 385 人学习可作为短小但结构完整的练手材料。1. 从一份 Redis 队列 demo 说起它到底能解决什么排队问题接口里塞一个发邮件的动作响应时间从 90ms 涨到 2.4s这种写法在早期项目里很常见流量一上来就是雪崩。python redis 任务队列 demo 解决的正是这个问题Web 进程只负责把任务塞进 Redis另一个 Python 进程慢慢消费两边通过一个 list 解耦。这份 python-redis-queue-sample 没有引入 Celery、RQ 这类框架直接用 redis-py 操作 list 和 zset把入队、出队、确认、重试、延迟、死信几个环节拆成能单独读的小文件改起来不用翻框架源码。它适合两类人想搞清楚任务队列底层到底发生了什么的中级开发者以及要在小项目里快速搭一套能跑的异步任务、又不想背一整套框架依赖的团队。环境自备一个 Redis 实例加 Python 3.9 以上就够跑。2. 拆开这份 demo依赖清单、键空间设计与启动顺序拿到一个队列工程我通常先不看代码先看它往 Redis 里写了哪些 key。key 的设计决定了后面所有问题的形状——能不能水平扩消费者、能不能做延迟、能不能安全回收全在键空间里定死了。这份 demo 的键空间不复杂一共五类 key但每一类对应一个明确的生命周期阶段读懂了基本就理解了整套流程。2.1 依赖清单与最小运行环境工程对环境的胃口很小。Python 侧只有 redis-py 一个第三方包其余全是标准库json、time、signal、uuid、logging。Redis 侧没有用 Stream、没有用模块普通 5.0 以上的版本就行因为核心命令只用到 LPUSH、RPOP、LMOVE、ZADD、ZRANGEBYSCORE、HSET 这几个老命令。# 拉起一个本地 Redis,关掉持久化,方便反复压测 redis-server --port 6379 --save --appendonly no # 建虚拟环境并装唯一依赖 python -m venv .venv source .venv/bin/activate pip install redis4.5--save --appendonly no是压测期的用法别带到生产。生产上 RDB 加 AOF 至少开一个否则 Redis 重启那一刻ready 列表里的任务就是真的没了。redis4.5这个下限的由来是 LMOVE 相关方法签名在 4.x 才稳定下来再往下的版本调lmove()会碰到参数顺序不一致的问题属于白白浪费半小时的坑。环境变量集中在config.py里读常见做法是这几项变量名默认值说明REDIS_URLredis://127.0.0.1:6379/0连接串,含 db 编号QUEUE_PREFIXqueue:task所有 key 的前缀,便于多环境隔离VISIBILITY_TIMEOUT60任务从进入 processing 到被判定超时的秒数MAX_RETRY3超过后进死信队列BRPOP_TIMEOUT5阻塞读超时,不用 0VISIBILITY_TIMEOUT是最需要跟业务对齐的参数不是拍脑袋填的。它应当大于任务 P99 执行时间的 2 到 3 倍短了会把正常执行中的任务误判为超时并重新入队长了会让真正卡死的任务在队列里躺很久。我一般的做法是先按 P99 乘 3 设一个初值上线后看回收次数这个指标再调。2.2 键空间设计五类 key 各管一段生命周期这套 demo 的关键设计是列表里只放任务 ID任务正文单独存一个 hash。这个选择看起来多了一次读但它换来了三件事——任务体可以很大而不撑爆列表、同一个任务可以被重复入队而不重复存正文、排查时可以直接按 ID 查状态。key类型作用有无 TTLqueue:task:readylist待消费任务 ID,LPUSH 进、RPOP 出无queue:task:processingzset处理中任务,score 为超时时间戳无queue:task:delayedzset延迟任务,score 为执行时间戳无queue:task:deadlist超过重试上限的死信无queue:task:meta:{id}hash任务正文、状态、重试次数、开始时间有,7 天processing 用 zset 而不是 list是这份代码里最值得抄的一处。list 只能告诉你有哪些任务在处理中zset 的 score 可以直接回答哪些任务已经超时了回收逻辑从遍历全部在途任务变成一次ZRANGEBYSCORE在途任务上万时差别很明显。提示meta 的 TTL 一定要设。漏设的话任务量一大hash 的数量只增不减Redis 内存曲线会是一条只涨不跌的斜线很难看。2.3 目录结构与启动顺序工程的文件划分按职责切不按技术层切一个文件基本对应一个可独立运行的进程python-redis-queue-sample/ ├── config.py # 环境变量与常量 ├── connection.py # 连接池单例 ├── producer.py # 入队接口,含延迟任务入口 ├── worker.py # 消费者主循环 ├── requeue.py # 超时任务回收,独立进程跑 ├── promote.py # 延迟任务到期迁移,独立进程跑 └── scripts/ ├── reserve.lua # 原子取任务 └── promote.lua # 原子迁移延迟任务启动顺序有讲究先起requeue.py和promote.py再起worker.py最后才是往队列里灌任务的生产者。反过来先灌任务、后起 worker不会出错但你会在压测时看到一个假象——队列长度瞬间冲到几万然后匀速下降误以为是消费能力不行其实是启动顺序造成的堆积。验证这套东西通没通最省事的办法是开三个终端一个跑 worker 看日志一个跑redis-cli monitor看命令流一个用redis-cli llen queue:task:ready盯长度。往队列里塞一条任务monitor 里应该依次出现LPUSH ready、LMOVE ready processing、处理完的ZREM processing三条命令的顺序和数量对得上链路就是通的。3. 队列核心实现入队、出队与阻塞读的三种写法队列代码最容易写错的地方不是业务逻辑是取任务这一步。取任务这个动作在 Redis 里天然不是原子的而它恰好又是唯一一个任务可能凭空消失的位置。这份 demo 在这一步上有三处递进从最朴素的 RPOP 到 BRPOP再到 LMOVE 加 Lua 脚本每一层都在补上一层的漏洞。3.1 从 LPUSH BRPOP 到 LMOVE为什么不能裸用 BRPOP最直观的写法是生产者LPUSH、消费者BRPOP。代码短跑起来也没问题但它有一个致命窗口BRPOP把任务从列表里弹出来、返回到 Python 进程再到你的业务代码开始执行中间这段时间任务只存在于进程内存里。进程被 kill、机器重启、docker restart这条任务就彻底没了队列和数据库都对不上账。import json import redis r redis.Redis(decode_responsesTrue) # 生产者:任务正文进 hash,ID 进 ready 列表 def enqueue(task: dict) - str: task_id task[id] pipe r.pipeline(transactionTrue) pipe.hset( fqueue:task:meta:{task_id}, mapping{ body: json.dumps(task, ensure_asciiFalse), state: ready, retry: 0, }, ) pipe.expire(fqueue:task:meta:{task_id}, 7 * 24 * 3600) # 兜底 TTL pipe.lpush(queue:task:ready, task_id) pipe.execute() # 两条命令打包,避免只写了一半 return task_id # 消费者:LMOVE 在 Redis 内部完成弹出 转移到 processing def reserve(timeout: int 5) - str | None: return r.lmove( queue:task:ready, queue:task:processing, timeouttimeout, srcRIGHT, # ready 是 LPUSH 进的,从右取才是 FIFO destLEFT, )pipeline(transactionTrue)把 hset 和 lpush 包成一个 MULTI/EXEC避免出现正文写进去了但 ID 没进队列的半截状态。srcRIGHT这个方向不能反生产者用 LPUSH 从左压入消费者必须从右弹出两边同侧就变成 LIFO任务顺序会整个倒过来做订单类业务时是灾难。LMOVE解决的是弹出和落到 processing的原子性任务永远至少存在于一个 key 里。但它还没解决落到 processing 之后怎么被回收——这个放到第 4 章。3.2 用 Lua 把取任务 记账压成一次调用LMOVE之后还需要更新任务状态、写开始时间。如果这几步在 Python 里串行做就又出现了窗口LMOVE 成功但 HSET 没执行回收进程看到 processing 里有个任务却没有开始时间判断逻辑就会退化。Lua 脚本在 Redis 单线程里执行整段要么全做要么全不做正好用来消除这个窗口-- scripts/reserve.lua -- KEYS[1] ready, KEYS[2] processing -- ARGV[1] 当前时间戳, ARGV[2] 超时秒数, ARGV[3] meta 前缀 local task_id redis.call(RPOP, KEYS[1]) if not task_id then return nil -- 队列空,直接返回,由调用方决定是否重试 end local deadline tonumber(ARGV[1]) tonumber(ARGV[2]) redis.call(ZADD, KEYS[2], deadline, task_id) -- score 超时时刻 redis.call(HSET, ARGV[3] .. task_id, state, processing, started_at, ARGV[1]) return task_id配套的 Python 侧调用import time import redis r redis.Redis(decode_responsesTrue) RESERVE r.register_script(open(scripts/reserve.lua).read()) def reserve_atomic(prefix: str queue:task:, timeout: int 60) - str | None: return RESERVE( keys[f{prefix}ready, f{prefix}processing], args[int(time.time()), timeout, f{prefix}meta:], )register_script只上传一次脚本之后走 EVALSHA网络开销比每次传脚本体小。参数上要留神两个点deadline用的是 Redis 侧传进去的时间戳而不是脚本里调TIME这样多个 worker 机器时钟不一致时至少是显式的ARGV[3]传前缀而不是在脚本里拼死字符串是为了让同一份脚本能在不同环境测试、预发复用。注意如果 Redis 是 Cluster 模式脚本里跨 key 操作会报 CROSSSLOT。常见做法是给 key 加 hash tag比如queue:{task}:ready和queue:{task}:processing让它们落同一个槽。3.3 消费者主循环与优雅退出主循环要处理四件事空转、异常、信号、连接断线。这份 demo 的循环写得比较克制没有引入线程池消费并发靠多起进程解决——这个选择在小项目里是对的Python 有 GIL队列消费这种 IO 密集场景多进程比多线程省心得多。import logging import signal import time import redis log logging.getLogger(worker) STOP False def _handle(signum, frame): global STOP STOP True # 只置标志,不在信号处理函数里做 IO signal.signal(signal.SIGTERM, _handle) signal.signal(signal.SIGINT, _handle) def run(): r redis.Redis( decode_responsesTrue, socket_timeout10, socket_connect_timeout3, health_check_interval30, # 主动探活,防空闲连接被中间设备掐断 ) while not STOP: try: task_id reserve_atomic() if not task_id: continue # 没任务就继续,阻塞在 Lua 的 RPOP 上不超过 1 次 handle(task_id, r) except redis.exceptions.ConnectionError: log.warning(redis 连接断开,2s 后重连) time.sleep(2) # 退避,避免重连风暴 except Exception: log.exception(任务处理异常,转入重试) retry_or_dead(task_id, r) log.info(收到退出信号,处理完当前任务后退出)health_check_interval这一项很多示例代码里没有但在容器环境里几乎是必需的。空闲超过几分钟的连接被 NAT 或负载均衡静默丢弃后下一次读取会抛一个看起来毫无道理的TimeoutError加了这个参数 redis-py 会在发命令前先 PING 一下把死连接提前换掉。信号处理函数里只置STOP标志所有清理动作放到循环里做。这是血泪经验在信号处理函数里调 Redis 命令有时会卡死因为主线程可能正持有连接池的锁信号在同一个线程里重入直接自锁。4. 可靠性补强ACK、超时回收、延迟队列与死信前面三章跑通的是正常路径这一章处理的全是异常路径。线上队列出问题九成不是入队出队写错了而是任务处理到一半进程没了这种边界没兜住。这份 demo 在这块给了三条线超时回收、延迟迁移、重试到死信三条线各跑一个进程互不阻塞。4.1 超时回收给 in-flight 任务一个后悔药processing是 zsetscore 是超时时刻。回收进程只需要定期扫一遍score 小于现在的成员把它们重新塞回 ready。这里的关键还是抢占——多个回收进程同时扫到同一条不能都往 ready 里塞。-- scripts/requeue.lua -- KEYS[1] processing, KEYS[2] ready, ARGV[1] 当前时间戳 local ids redis.call(ZRANGEBYSCORE, KEYS[1], -inf, ARGV[1], LIMIT, 0, 50) local moved 0 for _, id in ipairs(ids) do -- ZREM 返回 1 表示抢到了,返回 0 说明别的进程已经处理 if redis.call(ZREM, KEYS[1], id) 1 then redis.call(LPUSH, KEYS[2], id) moved moved 1 end end return movedZREM的返回值在这里就是一把轻量锁不需要额外加 SETNX。一次最多搬 50 条LIMIT 0 50是为了控制单次脚本的执行时间Redis 是单线程的脚本跑太久会阻塞所有客户端。回收进程的轮询间隔一般设 5 到 10 秒比VISIBILITY_TIMEOUT小一个数量级就行。真正麻烦的不是回收本身是回收带来的重复执行。一个任务执行了 70 秒、超时阈值是 60 秒回收进程就会把它重新入队于是同一个任务被两个 worker 各跑一遍。这是 at-least-once 语义的固有代价不是 bug。做法只有两条要么业务侧保证幂等用任务 ID 做去重表、用数据库唯一键要么在任务体里带一个版本号写回时做 CAS。指望队列层做到 exactly-once在 Redis 上是不现实的。4.2 延迟队列用 zset 的 score 当定时器延迟任务的实现思路很朴素ZADD时把 score 设成应该执行的时间戳迁移进程每秒扫一次到期的搬进 ready。麻烦的还是抢占这里同样用ZREM的返回值判断import time import redis r redis.Redis(decode_responsesTrue) PROMOTE r.register_script(open(scripts/promote.lua).read()) def poll_delayed(prefix: str queue:task:, batch: int 50) - int: now int(time.time()) ids r.zrangebyscore(f{prefix}delayed, 0, now, start0, numbatch) moved 0 for tid in ids: # 用 ZREM 抢占,抢到才 LPUSH,避免重复入队 if r.zrem(f{prefix}delayed, tid) 1: r.lpush(f{prefix}ready, tid) r.hset(f{prefix}meta:{tid}, state, ready) moved 1 return moved一个容易忽略的点是批量大小和扫描频率的匹配。如果每秒到期 200 个任务而迁移进程每秒只搬 50 个延迟队列就会越堆越多表现出来是任务延迟越来越不准而不是报错。压测时一定要单独盯ZCARD queue:task:delayed这个值它应该是锯齿状波动而不是单调上涨。延迟队列还有一个隐形成本meta 的 TTL 必须大于最大延迟时长。任务延迟 3 天执行meta 只留 7 天问题不大延迟 30 天就危险了任务到期时正文已经过期被删worker 取到 ID 去HGETALL得到空字典只能当失败处理。4.3 重试上限与死信队列的落地参数重试不做指数退避队列会被瞬间打爆。这份 demo 的做法是把重试也走延迟队列失败后retry 1按2^retry秒算出下次执行时间ZADD进 delayed超过MAX_RETRY就LPUSH进 dead。import json import time import redis r redis.Redis(decode_responsesTrue) MAX_RETRY 3 def retry_or_dead(task_id: str, prefix: str queue:task:) - None: meta_key f{prefix}meta:{task_id} retry int(r.hget(meta_key, retry) or 0) 1 if retry MAX_RETRY: r.hset(meta_key, mapping{state: dead, retry: retry}) r.lpush(f{prefix}dead, task_id) return delay 2 ** retry # 2s / 4s / 8s r.hset(meta_key, mapping{state: delayed, retry: retry}) r.zadd(f{prefix}delayed, {task_id: int(time.time()) delay}) r.zrem(f{prefix}processing, task_id) # 从在途集合里摘掉2 ** retry的上限要夹一下2 ** 20是 12 天明显不合理。常见做法是min(2 ** retry, 3600)封顶一小时。死信队列本身不是终点它是个暂存区必须有人看——工程里至少要有个接口能LRANGE queue:task:dead 0 50最好再配一个告警dead 长度大于 0 就通知。死信没人看等于没有死信这个坑我见过不止一次。5. 避坑与排查这份 demo 上线前必须过的五道关demo 在本机跑通和在生产跑稳是两件事差别全在边界条件上。下面五条是我在类似工程上真实遇到过的每条按现象、原因、解决写照着排查基本能覆盖大部分线上故障。5.1 连接层与阻塞读三条最容易翻车的地方第一条空闲几分钟后第一次取任务抛 TimeoutError。现象是 worker 夜里安静了几个小时早高峰第一条任务就报连接超时重试一次又好了。原因不在 Redis 本身是连接在中间网络设备上被静默回收了客户端还以为它是活的。解决办法是建连时带上health_check_interval30和socket_keepaliveTrue让 redis-py 在做命令前先探活同时socket_timeout不能设太大10 秒足够设 60 秒只会让故障发现得更晚。第二条worker 扩到 8 个connected_clients涨到四五百。现象是 Redis 的INFO clients里连接数远超 worker 进程数乘预期。原因通常是两处一是在multiprocessingfork 之前就建好了连接池子进程继承了父进程的 socket多个进程共用一个 fd命令串包二是每个任务都redis.Redis()新建一个客户端。解决办法是把连接池的创建推迟到子进程入口函数里并且用ConnectionPool(max_connections32)限制单进程上限全局只留一个客户端实例。第三条CtrlC杀不掉 worker要等很久。现象是发 SIGTERM 后进程挂着不动日志也不输出。原因是阻塞读用了timeout0永久阻塞在 socket 上Python 的信号处理要等主线程从系统调用返回才能执行。解决办法是把阻塞超时设成 3 到 5 秒循环每轮检查一次退出标志最坏情况下 5 秒内一定能退出来。这个数字不要贪大5 秒已经能挡住绝大部分空轮询。5.2 数据一致性两条会让任务消失或重复的坑第四条任务偶尔被执行两次。现象是下游日志里同一个业务 ID 出现了两条成功记录间隔大约等于VISIBILITY_TIMEOUT。原因就是 4.1 说的任务执行时间超过可见性超时回收进程把它重新入队两个 worker 各跑了一遍。解决分两层队列层把VISIBILITY_TIMEOUT调到 P99 的三倍业务层必须做幂等——用任务里的业务唯一键写一张去重表或者写库时用INSERT ... ON CONFLICT DO NOTHING。只调参数不做幂等迟早还会撞上。第五条延迟任务堆积在 zset 里不动。现象是ZCARD queue:task:delayed一直涨但 ready 长度不动任务就是不执行。原因可能是迁移进程压根没起也可能是多个迁移进程用ZRANGEBYSCORE加ZREM的非原子组合互相踩谁都以为别人搬了。解决办法是把迁移写成 Lua 脚本或者严格用ZREM返回值抢占4.2 的写法同时给迁移进程加一个心跳 keySET promote:heartbeat ts EX 30监控发现 key 不存在就告警。进程静默死掉是队列系统里最隐蔽的故障没有心跳你根本不知道它停了。还有一个内存层面的坑值得单独提meta hash 没设 TTL。任务量日均几十万的时候一个月就是上千万个 key内存曲线一路上扬最后触发maxmemory淘汰策略把 ready 列表里的任务挤掉——这时候丢的不是缓存是真实任务。meta 的EXPIRE必须在入队时就设别指望事后补。6. 压测与观测怎么验证队列真的没丢任务队列这东西最怕的不是慢是看起来在跑其实丢了。判断有没有丢靠肉眼看日志不靠谱得靠总量守恒入队数 成功数 死信数 在途数 待消费数。写个压测脚本把这条等式跑一遍比读十遍代码都管用。6.1 用守恒等式造压import time import uuid import redis r redis.Redis(decode_responsesTrue) PREFIX queue:task: def produce(n: int 10000, delay_ratio: float 0.1) - None: pipe r.pipeline(transactionFalse) for i in range(n): tid uuid.uuid4().hex pipe.hset(f{PREFIX}meta:{tid}, mapping{ body: {type:demo,seq:%d} % i, state: ready, retry: 0, }) pipe.expire(f{PREFIX}meta:{tid}, 7 * 24 * 3600) if i % int(1 / delay_ratio) 0: pipe.zadd(f{PREFIX}delayed, {tid: int(time.time()) 30}) else: pipe.lpush(f{PREFIX}ready, tid) if i % 500 0: # 分批提交,单次管道别超过千级 pipe.execute() pipe.execute() def audit(expected: int) - dict: ready r.llen(f{PREFIX}ready) processing r.zcard(f{PREFIX}processing) delayed r.zcard(f{PREFIX}delayed) dead r.llen(f{PREFIX}dead) done int(r.get(f{PREFIX}stat:done) or 0) total ready processing delayed dead done return {expected: expected, actual: total, gap: expected - total} if __name__ __main__: produce(10000) time.sleep(120) print(audit(10000))gap恒等于 0 才算过关。如果 gap 是负数说明有任务被重复计数多半是回收逻辑重复入队如果 gap 是正数说明任务真的丢了去查LMOVE之后到ZADD processing之间有没有非原子操作。管道批量提交这一点也别省单条execute()一万次和五百条一批压测出来的吞吐能差三倍测出来的数就没参考价值了。6.2 三个必须盯的指标指标采集方式告警阈值ready 长度LLEN queue:task:ready连续 5 分钟单调上涨在途超时数ZCOUNT processing -inf now大于在途总量 5%死信长度LLEN queue:task:dead大于 0 立即告警ready 单调上涨基本等于消费能力不足或者 worker 全挂了在途超时数高说明VISIBILITY_TIMEOUT设小了或者有任务卡死死信大于 0 就是有业务在持续失败必须有人看。三个指标都能用redis-cli --latency之外的普通命令采接进现成的监控体系即可不需要额外组件。我后来养成的习惯是任何队列工程接手第一件事不是读业务代码而是先把这条守恒等式在测试环境跑一遍跑不平就先别谈优化。这套动作帮我省下过好几次上线后半夜爬起来捞任务的时间。希望帮到你。本文还有配套的精品资源点击获取