企业官网建设流程全解析

简介：python-redis-queue-sample 面向希望快速上手分布式任务队列的 Python 开发者，以 RQ 搭配 Redis 给出一个最小可运行的任务队列示例，演示如何把爬虫抓取、邮件推送、图片处理等耗时操作从主流程中剥离，交给后台 worker 异步执行。它适合已具备 Python 基础、刚开始接触任务调度的初中级读者。压缩包共 5 个文件、约 3KB，主体是 2 个 Python 脚本，分别负责任务投递与任务处理逻辑；另附 1 个已编译的 pyc、1 份 README 说明文档和 1 个 gitignore 配置。压缩包体量小、结构直观，解压即可阅读。目前已有 385 人学习下载。读者可从中理清 Redis 连接初始化、Queue 入队、worker 监听与结果回收的完整链路，并借助 README 中的环境准备说明（Redis 服务、rq 依赖、rqworker 启动命令）快速复现实验环境；在此基础上，还可继续扩展重试、超时、优先级等调度策略，省去从零搭建脚手架的时间。

1. Redis 任务队列不是“装个 redis 就行”：自备 demo 的边界先说清

python redis 任务队列 demo 的价值不在“能跑”，而在“能改”。搜到这个主题的人，多半是想绕开重型框架，用 Python 直接控制入队、出队、重试和死信，把队列行为握在自己手里。这份 demo 的“自备”两个字很关键：Redis 服务、Python 环境、连接参数都要自己准备，它不替你兜底运维。适合想理解队列底层语义、需要轻量异步任务、做内部工具或练手的开发者；不适合上来就要可视化、复杂定时调度、多租户隔离的生产平台。先把它当一张可拆的骨架图，跑通双进程，再按业务补可靠性，比直接复制一个“万能队列”靠谱。

2. 队列结构选型：List、ZSet、Stream 三种模型怎么对号入座

2.1 List 的 LPUSH/BRPOP：最像 queue 的阻塞语义

Python 侧用 Redis 做任务队列，最先碰到的就是 List。它的语义简单：一端进、另一端出，配合阻塞命令，消费者没有任务时挂起，不空转。常见做法是生产者LPUSH从左边入队，消费者BRPOP从右边取，这样先入先出。下面先看命令层面的最小验证，不急着上 Python 代码。

# 生产者：从左侧入队，value 是任务的 JSON 字符串 redis-cli LPUSH demo:queue:tasks '{"id":"t1","name":"resize_image"}' # 消费者：从右侧阻塞取，最多等 5 秒；返回 [队列名, 任务内容] redis-cli BRPOP demo:queue:tasks 5

LPUSH的返回值是入队后的列表长度，适合做粗粒度监控；BRPOP的timeout是阻塞秒数，设为 0 会一直阻塞，生产环境不建议，因为连接会长期占住，Redis 重启或网络抖动时排错困难。BRPOP返回的是两个元素的数组：第一个是队列名，第二个是任务内容。很多 Python 示例只取第二个元素，忘了判空，结果在超时返回None时直接json.loads(None)，日志里就是一堆看不懂的序列化错误。

List 的优点是没有学习成本，缺点也明显：任务弹出后如果消费者进程挂了，这条任务就丢了，Redis 的 List 本身没有“未确认”状态。要补可靠性，常见做法是BRPOPLPUSH或较新版本推荐的LMOVE，把任务从待处理队列原子地挪到 processing 队列，处理成功再删。这个动作在第 4 章会展开。选型上，List 适合任务量不大、允许少量丢失、或者你愿意自己补 processing 队列的场景。别把它当“可靠消息队列”用，它只是一个阻塞链表。

2.2 ZSet 的 score 排序：延迟队列和重试的底座

任务队列不只有“立刻执行”，还有“失败后等 10 秒再试”“定时到点再跑”。这类需求用 List 硬做会很别扭，ZSet 更合适：member 存任务 JSON，score 存到期时间戳。消费者用ZRANGEBYSCORE取0到当前时间的那批，取到后再ZREM删除。先看命令：

# 入延迟队列：score 是到期时间戳，这里用当前时间 + 10 秒示意 redis-cli ZADD demo:queue:delay 1700000010 '{"id":"t2","retry":1}' # 取出已到期的前 20 条 redis-cli ZRANGEBYSCORE demo:queue:delay 0 1700000000 LIMIT 0 20 # 确认取走后删除，避免被其他调度器重复取 redis-cli ZREM demo:queue:delay '{"id":"t2","retry":1}'

ZADD的 score 用浮点数存 Unix 时间戳，秒或毫秒都行，但全项目要统一，混用会导致任务要么永远不到期，要么瞬间全到期。ZRANGEBYSCORE的LIMIT offset count用来批量取，不要一次取太多，否则 Lua 脚本或调度循环会卡住 Redis 单线程。ZREM的返回值是实际删除数量，只有返回 1 才说明当前消费者抢到了这条任务；多消费者场景下，先ZRANGEBYSCORE再ZREM存在竞争窗口，稳妥做法是用 Lua 把“取到期 + 删除”做成原子操作，第 6 章会给脚本。

ZSet 的坑在“重复调度”。两个调度器同时读到同一条到期任务，都执行ZREM，只有一个返回 1，另一个应该放弃。很多 demo 省略了这个判断，结果同一条重试任务被入队两次。另一个坑是 score 精度：用秒级时间戳时，同一秒到期的任务会扎堆，批量上限设小了会延迟，设大了会瞬时打满消费者。我一般会把批量控制在 20 到 50，按消费者数量和任务耗时调。ZSet 不适合直接当主队列，因为它是按分数排序，不是按入队顺序；它最适合做延迟层和重试层，和 List 主队列配合。

2.3 Stream 的消费组：需要 ACK 时再上，不要为了新而新

Redis Stream 提供消费组和 ACK，天然适合“任务不能丢”的场景。XADD追加消息，XGROUP CREATE建组，XREADGROUP带组名和消费者名读取，处理完XACK。先看命令序列：

# 追加任务，* 表示由 Redis 生成消息 ID redis-cli XADD demo:stream:tasks '*' name resize_image path /tmp/a.jpg # 创建消费组 g1，从 0 开始；MKSTREAM 表示 stream 不存在时一并创建 redis-cli XGROUP CREATE demo:stream:tasks g1 0 MKSTREAM # 消费者 c1 读取一条，BLOCK 5000 表示阻塞 5 秒 redis-cli XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 5000 STREAMS demo:stream:tasks '>' # 处理成功后确认 redis-cli XACK demo:stream:tasks g1 1700000000000-0

XADD的*让 Redis 生成时间戳加序号的消息 ID，生产环境也可以自己指定，但没必要。XGROUP CREATE的0表示从最早消息开始消费，$表示只消费新消息；重建消费组时选错会重放历史任务。XREADGROUP的>表示读新消息，读出来的消息会进入 Pending Entries List，没XACK就一直挂着，可以用XPENDING查。XACK的第三个参数是消息 ID，不是任务 ID，别混。

Stream 的优点是有 ACK 和 pending 查询，缺点是概念多、内存清理要配XTRIM或MAXLEN，否则 stream 会一直涨。Python 侧用 redis-py 的xadd、xreadgroup、xack也能写，但代码量比 List 多。选型建议：任务允许丢或能靠业务补偿，用 List；需要延迟和重试，List + ZSet；任务完全不能丢、需要消费组水平扩展，再上 Stream。不要因为 Stream 新就全套上，很多内部工具的并发量根本用不到消费组，反而增加排错成本。

3. 把 demo 跑成双进程：连接、生产、消费三段代码

3.1 依赖与 Redis 连接：连接池、超时、decode_responses

这份 demo 的运行前提是本地或可达的 Redis 服务，以及 Python 环境。依赖只装 redis 客户端即可，不需要额外框架。先建一个requirements.txt，再装依赖。这里不写死版本号，按当前环境可用的稳定版装，装完用pip freeze锁定到自己的环境里。

# requirements.txt redis

python -m venv .venv source .venv/bin/activate pip install -r requirements.txt

连接对象不要每次操作都新建，Python 侧用连接池复用。下面这段是后续生产者和消费者共用的连接工厂，参数按本地 demo 给，迁到内网时改host、port、password、db。

# redis_conn.py import redis def get_redis(): pool = redis.ConnectionPool( host="127.0.0.1", port=6379, db=0, password=None, decode_responses=True, # 让返回 str，方便 json.loads socket_timeout=5, # 单次命令读写超时 socket_connect_timeout=3, # 建连超时 health_check_interval=30, # 空闲连接健康检查 max_connections=50, # 连接池上限，按消费者线程数调 ) return redis.Redis(connection_pool=pool)

decode_responses=True是关键参数：redis-py 默认返回 bytes，任务 JSON 用 bytes 也能json.loads，但日志和异常里全是b'...'，排查时很烦。开了之后所有字符串命令返回 str，二进制场景要单独处理。socket_timeout不要设太大，消费者用BRPOP时阻塞时间要小于或等于这个超时，否则客户端可能先报超时。max_connections按并发消费者数量留余量，设太小会出现“拿不到连接”的报错，看起来像 Redis 挂了，实际是池子耗尽。health_check_interval用来清理空闲坏连接，内网环境建议保留。

3.2 生产者：任务封包、入队和字段约束

生产者负责把业务参数封成统一结构，再LPUSH到主队列。任务结构建议至少包含id、name、payload、retry、max_retry、enqueue_ts。id用 UUID，后面做幂等键和日志追踪都要用；name是任务类型，消费者据此分发；retry从 0 开始，失败时累加；max_retry控制上限；enqueue_ts用来算排队耗时。下面是一个可直接运行的生产者。

# producer.py import json import time import uuid import redis from redis_conn import get_redis QUEUE = "demo:queue:tasks" def build_task(name: str, payload: dict, max_retry: int = 3) -> dict: return { "id": uuid.uuid4().hex, "name": name, "payload": payload, "retry": 0, "max_retry": max_retry, "enqueue_ts": time.time(), } def enqueue(r: redis.Redis, task: dict) -> int: raw = json.dumps(task, ensure_ascii=False, separators=(",", ":")) return r.lpush(QUEUE, raw) def main(): r = get_redis() for i in range(5): task = build_task("resize_image", {"path": f"/tmp/img_{i}.jpg", "width": 800}) size = enqueue(r, task) print(f"enqueued id={task['id']} queue_size={size}") if __name__ == "__main__": main()

json.dumps用ensure_ascii=False保留中文，separators去掉多余空格，减小 payload。lpush返回队列长度，可以直接打印，但不要把它当精确积压指标，因为消费者可能同时在取。build_task里的payload建议只放可 JSON 序列化的基础类型，不要把文件对象、数据库连接、datetime直接塞进去；datetime先转 ISO 字符串或时间戳。max_retry按业务定，纯计算任务可以 3 次，调用外部接口的任务要更谨慎，重试太多次会把下游打挂。

生产者的常见误用是“先查长度再入队”或“入队前做重活”。队列只负责传递，业务校验放在消费者或前置服务里，生产者越轻越好。另一个细节是队列名加业务前缀，比如demo:queue:tasks，不要直接用queue，多个项目共用一个 Redis 时容易串。入队失败要处理异常，至少打日志并保留任务 JSON，不要静默吞掉。

3.3 消费者：阻塞循环、异常隔离与优雅退出

消费者是 demo 的核心。它要阻塞取任务、反序列化、按name分发、捕获异常、响应退出信号。下面这版是“能跑”的基础版，先不处理 processing 队列和重试，目的是把链路跑通，后续章节再补可靠性。

# consumer.py import json import signal import time import redis from redis_conn import get_redis QUEUE = "demo:queue:tasks" STOP = False def handle_signal(signum, frame): global STOP STOP = True def run_task(task: dict) -> dict: name = task["name"] payload = task.get("payload", {}) if name == "resize_image": # demo 用 sleep 模拟耗时，真实业务替换成图像处理 time.sleep(0.2) return {"ok": True, "path": payload.get("path")} raise ValueError(f"unknown task name: {name}") def main(): signal.signal(signal.SIGINT, handle_signal) signal.signal(signal.SIGTERM, handle_signal) r = get_redis() while not STOP: item = r.brpop(QUEUE, timeout=5) if item is None: continue _, raw = item try: task = json.loads(raw) result = run_task(task) print(f"done id={task['id']} result={result}") except Exception as exc: # 基础版只打日志，第 4 章再补重试和死信 print(f"failed raw={raw[:200]} error={exc!r}") if __name__ == "__main__": main()

brpop返回None表示超时，不是错误，直接continue进入下一轮。timeout=5让循环有机会检查STOP，如果设 0 会一直阻塞，按Ctrl+C后进程可能不退出。信号处理里只改标志位，不直接做清理，避免在信号上下文里执行 Redis 命令。run_task按name分发，未知任务抛异常，进入失败分支。基础版失败只打日志，任务会丢，这是故意的，用来对照第 4 章的补丁。

运行顺序：先起 Redis，再开两个终端，一个跑python producer.py，一个跑python consumer.py。消费者先启动会阻塞等待，生产者入队后立刻能消费到。如果消费者日志一直为空，先redis-cli LLEN demo:queue:tasks看队列长度，再redis-cli MONITOR看有没有LPUSH和BRPOP命令到达。MONITOR只在排查时短开，生产环境不要长期开着，会影响性能。优雅退出的验证方法是消费者运行时按Ctrl+C，进程应该在当前任务处理完后退出，而不是卡在brpop里。

4. 可靠性补丁：重试、死信与幂等怎么塞进 demo

4.1 重试计数与指数退避：别在异常里原地打转

基础版消费者一异常就把任务丢了，真实场景至少要重试。重试不能简单while循环原地重跑，否则下游故障时会把 CPU 和日志打满。常见做法是失败后把retry加 1，算出退避时间，把任务放进 ZSet 延迟队列，到点再由调度器挪回主队列。下面这段是重试逻辑，接在消费者异常分支里。

# retry.py import json import time DELAY_QUEUE = "demo:queue:delay" DEAD_QUEUE = "demo:queue:dead" def backoff_seconds(retry: int) -> int: # 2、4、8、16 秒递增，封顶 60 秒 return min(60, 2 ** retry) def retry_or_dead(r, task: dict, exc: Exception) -> str: task["retry"] = int(task.get("retry", 0)) + 1 task["last_error"] = repr(exc)[:500] if task["retry"] > int(task.get("max_retry", 3)): raw = json.dumps(task, ensure_ascii=False, separators=(",", ":")) r.lpush(DEAD_QUEUE, raw) return "dead" delay = backoff_seconds(task["retry"]) task["next_ts"] = time.time() + delay raw = json.dumps(task, ensure_ascii=False, separators=(",", ":")) r.zadd(DELAY_QUEUE, {raw: task["next_ts"]}) return f"retry_after_{delay}s"

backoff_seconds用 2 的 retry 次方，封顶 60 秒，避免第 10 次重试等几个小时。last_error截断到 500 字符，防止异常栈把 Redis value 撑大。max_retry从任务里读，生产者和消费者对上限的理解要一致，别一个写 3 一个写 5。zadd的 member 是完整任务 JSON，score 是next_ts，这样调度器只需要按分数取。重试任务再次失败时，retry继续累加，直到进死信队列。

消费者异常分支改成调用retry_or_dead，同时要把基础版里“弹出即丢”的问题一起处理。最简单的方式是消费者弹出后先不删，失败时直接进延迟队列；但BRPOP已经把任务从主队列移除了，失败后进延迟队列不会重复。真正会丢的场景是消费者在处理中途崩溃，任务既不在主队列也不在延迟队列，所以第 5 章会讲 processing 队列。这里先把重试链路跑通，用redis-cli ZRANGE demo:queue:delay 0 -1 WITHSCORES能看到带分数的任务。

4.2 死信队列：给失败任务留一个后悔药

死信队列不是“垃圾桶”，而是人工补偿入口。任务超过重试上限后进demo:queue:dead，运维或开发可以查内容、修数据、重新入队。死信队列用 List 存，配合一个简单的查看脚本，别让它无限增长。下面是一个最小死信查看和重投脚本。

# dead_letter.py import json import redis from redis_conn import get_redis DEAD_QUEUE = "demo:queue:dead" TASK_QUEUE = "demo:queue:tasks" def peek_dead(r: redis.Redis, count: int = 10): # lrange 只读不删，方便排查 return r.lrange(DEAD_QUEUE, 0, count - 1) def requeue_one(r: redis.Redis): # 从右侧取一条死信，重置 retry 后重新入主队列 raw = r.rpop(DEAD_QUEUE) if raw is None: return None task = json.loads(raw) task["retry"] = 0 task.pop("last_error", None) task.pop("next_ts", None) r.lpush(TASK_QUEUE, json.dumps(task, ensure_ascii=False, separators=(",", ":"))) return task["id"] if __name__ == "__main__": r = get_redis() for raw in peek_dead(r, 5): print(raw[:300])

lrange从左侧读，不删除，适合排查；rpop从右侧取，和生产者lpush配合形成先进先出。重投前把retry清零，清掉last_error和next_ts，避免旧字段干扰。死信队列要有告警，至少每天看一次长度，长度持续上涨说明下游有稳定故障。不要写自动无限重投脚本，否则死信会变成循环队列。

4.3 幂等键：重复消费时先问“这事干过没”

队列的“至少一次”语义意味着重复消费不可避免：消费者处理成功但确认前崩溃、重试任务和原任务同时到达、人工重投，都可能让同一个任务跑两次。幂等键是后悔药。常见做法是用任务id或业务唯一键做 RedisSET NX EX，执行前占位，成功保留，失败删除。下面是一个包装函数。

# idempotent.py import redis IDEM_PREFIX = "demo:idem:" def run_once(r: redis.Redis, task: dict, handler, ttl: int = 3600): key = f"{IDEM_PREFIX}{task['id']}" # nx=True 只有 key 不存在才设置成功；ex 是过期时间 acquired = r.set(key, "1", nx=True, ex=ttl) if not acquired: return {"skipped": True, "reason": "duplicate"} try: result = handler(task) except Exception: # 失败删 key，允许后续重试重新执行 r.delete(key) raise return result

ttl按业务执行时间设，至少要覆盖“任务执行 + 重试窗口”，设太短会失去幂等保护，设太长会占内存。handler是真正的业务函数，run_once只负责占位和释放。注意这个方案是“先占位再执行”，如果业务执行很久，ttl到期后重复任务仍可能进来，所以长任务要把ttl放大，或者用后台续期。更严格的幂等要靠数据库唯一索引或状态机，Redis 键只是第一道闸。

5. 避坑与排查：五个翻车现场和对应排查顺序

5.1 连接池耗尽：报错像网络问题，根因在池子

现象：消费者跑一段时间后开始报ConnectionError: Too many connections或拿连接超时，重启进程能好一阵，随后复发。看 Redis 服务端INFO clients，connected_clients持续上涨，但业务量并没有明显增加。

原因：每次任务都新建redis.Redis(...)，或者连接池max_connections设得比实际并发小，加上BRPOP长时间阻塞占住连接，池子很快见底。另一个隐蔽原因是异常分支里创建了新连接却没释放。

解决：全进程共用一个ConnectionPool，消费者数量乘以阻塞连接数不超过max_connections。用redis-cli INFO clients和CLIENT LIST观察连接来源，确认是不是同一台机器疯狂建连。把max_connections调到并发消费者数量的 1.5 倍左右，并给socket_timeout设合理值，避免坏连接长期占位。

5.2 BRPOP 空转：timeout 设 0 与设 1 的差别

现象：Redis CPU 不高，但消费者日志疯狂打印空循环，或者按Ctrl+C后进程迟迟不退出。用MONITOR看，BRPOP命令一条接一条，几乎没有间隔。

原因：timeout设了 0，消费者一直阻塞，信号处理函数没机会执行；或者设了很小的值又在循环里没有sleep，超时后立刻重试，形成命令风暴。还有一种情况是队列名写错，消费者一直阻塞在一个不存在的队列上。

解决：timeout设 3 到 5 秒，超时返回None后continue，循环自然检查退出标志。队列名统一用常量，生产和消费引同一份配置。排查时先用redis-cli LLEN 队列名确认队列存在且有数据，再用CLIENT LIST看阻塞连接数。不要用KEYS *找队列，改用SCAN。

5.3 序列化翻车：JSON 里塞 bytes 和 datetime

现象：生产者入队正常，消费者一取就报TypeError: Object of type datetime is not JSON serializable，或者取出来是b'{"id":...}'，json.loads后字段全是 bytes。

原因：任务 payload 里直接放了datetime、Decimal、文件对象或 bytes；或者连接没开decode_responses，读出来是 bytes，代码里又按 str 处理。

解决：入队前统一序列化，时间转 ISO 字符串或时间戳，金额转字符串或整数分，二进制用 base64。连接池统一decode_responses=True，如果确实要存二进制，单独用一个不开解码的连接。消费者反序列化失败时，把原始 raw 截断打日志，不要直接吞掉，方便定位是哪条任务。

5.4 消费者重启：任务在 BRPOP 弹出后丢了

现象：消费者处理到一半被 kill，重启后任务没重试，也没进死信，业务方说“少了一单”。查主队列长度是 0，延迟队列也没有。

原因：BRPOP是“弹出即删除”，任务离开主队列后，消费者崩溃就彻底丢了。基础版 demo 没有 processing 队列，所有可靠性补丁都建立在“任务还在”的前提上。

解决：用BLMOVE或BRPOPLPUSH把任务原子地挪到demo:queue:processing，处理成功再LREM删除；失败时从 processing 移除并进重试队列。消费者启动时可以先扫 processing 里超过一定时间的任务，做补偿。注意LREM要按完整 raw 删除，不要只按任务 id，因为 List 没有按字段删除的能力。

5.5 监控盲区：队列长度不等于积压任务数

现象：看板显示主队列长度为 0，以为没积压，实际 processing 队列堆了几百条，延迟队列里还有一堆到期没调度的任务。用户侧表现为任务越来越慢。

原因：只监控LLEN 主队列，忽略了 processing、delay、dead 三个结构。主队列长度为 0 可能只是消费者取走了，不代表处理完了。延迟队列的到期任务如果调度器挂了，也不会体现在主队列长度上。

解决：至少监控四个值：主队列长度、processing 长度、延迟队列ZCARD、死信长度。再加一个“最老任务排队时长”，用任务里的enqueue_ts算。调度器本身要有一个心跳键，比如SET demo:beat 1 EX 30，看心跳是否存在判断调度器活没活。排查顺序是：先看死信有没有涨，再看 processing 是否只增不减，最后看延迟队列到期任务有没有被搬走。

6. 进阶：用 Lua 把“取到期任务”做成原子操作，再补两个指标

6.1 Lua 调度脚本与 EVALSHA 缓存

第 2 章的延迟队列先查后删有竞争窗口，多调度器同时跑会重复入队。把“取到期 + 删除 + 入主队列”写进 Lua，Redis 单线程执行，天然原子。脚本放move_due.lua，Python 用register_script调用，后续走EVALSHA减少脚本传输。

-- move_due.lua -- KEYS[1] 延迟队列，KEYS[2] 主队列 -- ARGV[1] 当前时间戳，ARGV[2] 批量上限 local src = KEYS[1] local dst = KEYS[2] local now = tonumber(ARGV[1]) local batch = tonumber(ARGV[2]) local items = redis.call('ZRANGEBYSCORE', src, 0, now, 'LIMIT', 0, batch) for i, raw in ipairs(items) do -- 只有真正删除成功才入主队列，避免重复 if redis.call('ZREM', src, raw) == 1 then redis.call('LPUSH', dst, raw) end end return #items

# scheduler.py import time import redis from redis_conn import get_redis DELAY_QUEUE = "demo:queue:delay" TASK_QUEUE = "demo:queue:tasks" def load_script(r: redis.Redis): with open("move_due.lua", "r", encoding="utf-8") as f: return r.register_script(f.read()) def tick(r: redis.Redis, script, batch: int = 20) -> int: # keys 和 args 顺序与 Lua 里一致 return script(keys=[DELAY_QUEUE, TASK_QUEUE], args=[time.time(), batch]) if __name__ == "__main__": r = get_redis() s = load_script(r) print("moved", tick(r, s))

register_script第一次会SCRIPT LOAD，之后自动用EVALSHA，Redis 重启后脚本缓存丢失会自动回退到EVAL。batch不要太大，单次 Lua 执行时间过长会阻塞其他命令。这个脚本只负责搬运，不负责重试计数，重试逻辑仍然在消费者异常分支里。

6.2 两个必看指标与验证脚本

调度器跑起来后，至少盯两个指标：延迟队列到期未搬走的数量和 processing 队列长度。前者持续大于 0，说明调度器挂了或批量太小；后者持续上涨，说明消费者处理不过来或确认逻辑有 bug。用下面几条命令做日常验证，别只看主队列。

redis-cli LLEN demo:queue:tasks redis-cli LLEN demo:queue:processing redis-cli ZCARD demo:queue:delay redis-cli LLEN demo:queue:dead redis-cli ZRANGEBYSCORE demo:queue:delay 0 "$(date +%s)" LIMIT 0 5

最后一条会列出已经到期但还没被搬走的任务，正常应该很快被调度器清掉。我的血泪经验是：队列类 demo 最容易被“能跑”骗过去，真正上线前一定要用 kill -9 模拟消费者崩溃，再看任务能不能从 processing 里捞回来。从那以后我每次改队列代码，都强制走一遍“入队、消费、崩溃、重启、补偿”这五步，再补一个死信查看入口，才敢放到共享环境里。希望帮到你。

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