企业官网建设流程全解析

简介：一套面向 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 "redis>=4.5"

--save "" --appendonly no是压测期的用法，别带到生产。生产上 RDB 加 AOF 至少开一个，否则 Redis 重启那一刻，ready 列表里的任务就是真的没了。redis>=4.5这个下限的由来是 LMOVE 相关方法签名在 4.x 才稳定下来，再往下的版本调lmove()会碰到参数顺序不一致的问题，属于白白浪费半小时的坑。

环境变量集中在config.py里读，常见做法是这几项：

VISIBILITY_TIMEOUT是最需要跟业务对齐的参数，不是拍脑袋填的。它应当大于任务 P99 执行时间的 2 到 3 倍，短了会把正常执行中的任务误判为超时并重新入队，长了会让真正卡死的任务在队列里躺很久。我一般的做法是先按 P99 乘 3 设一个初值，上线后看回收次数这个指标再调。

2.2 键空间设计：五类 key 各管一段生命周期

这套 demo 的关键设计是：列表里只放任务 ID，任务正文单独存一个 hash。这个选择看起来多了一次读，但它换来了三件事——任务体可以很大而不撑爆列表、同一个任务可以被重复入队而不重复存正文、排查时可以直接按 ID 查状态。

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_responses=True) # 生产者:任务正文进 hash,ID 进 ready 列表 def enqueue(task: dict) -> str: task_id = task["id"] pipe = r.pipeline(transaction=True) pipe.hset( f"queue:task:meta:{task_id}", mapping={ "body": json.dumps(task, ensure_ascii=False), "state": "ready", "retry": 0, }, ) pipe.expire(f"queue: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", timeout=timeout, src="RIGHT", # ready 是 LPUSH 进的,从右取才是 FIFO dest="LEFT", )

pipeline(transaction=True)把 hset 和 lpush 包成一个 MULTI/EXEC，避免出现"正文写进去了但 ID 没进队列"的半截状态。src="RIGHT"这个方向不能反：生产者用 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_responses=True) 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_responses=True, socket_timeout=10, socket_connect_timeout=3, health_check_interval=30, # 主动探活,防空闲连接被中间设备掐断 ) 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是 zset，score 是超时时刻。回收进程只需要定期扫一遍"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 moved

ZREM的返回值在这里就是一把轻量锁，不需要额外加 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_responses=True) 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, start=0, num=batch) 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_responses=True) 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_interval=30和socket_keepalive=True，让 redis-py 在做命令前先探活；同时socket_timeout不能设太大，10 秒足够，设 60 秒只会让故障发现得更晚。

第二条：worker 扩到 8 个，connected_clients涨到四五百。现象是 Redis 的INFO clients里连接数远超 worker 进程数乘预期。原因通常是两处：一是在multiprocessingfork 之前就建好了连接池，子进程继承了父进程的 socket，多个进程共用一个 fd，命令串包；二是每个任务都redis.Redis()新建一个客户端。解决办法是把连接池的创建推迟到子进程入口函数里，并且用ConnectionPool(max_connections=32)限制单进程上限，全局只留一个客户端实例。

第三条：Ctrl+C杀不掉 worker，要等很久。现象是发 SIGTERM 后进程挂着不动，日志也不输出。原因是阻塞读用了timeout=0，永久阻塞在 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 的写法），同时给迁移进程加一个心跳 key，SET 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_responses=True) PREFIX = "queue:task:" def produce(n: int = 10000, delay_ratio: float = 0.1) -> None: pipe = r.pipeline(transaction=False) 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 单调上涨基本等于消费能力不足或者 worker 全挂了；在途超时数高说明VISIBILITY_TIMEOUT设小了或者有任务卡死；死信大于 0 就是有业务在持续失败，必须有人看。三个指标都能用redis-cli --latency之外的普通命令采，接进现成的监控体系即可，不需要额外组件。

我后来养成的习惯是：任何队列工程接手第一件事，不是读业务代码，而是先把这条守恒等式在测试环境跑一遍，跑不平就先别谈优化。这套动作帮我省下过好几次上线后半夜爬起来捞任务的时间。希望帮到你。

本文还有配套的精品资源，点击获取