基于Redis的Python任务队列：可靠消费与延迟重试实践

简介：面向具备一定 Python 基础的开发者，这份示例围绕 Redis 与 RQ 搭建轻量级异步任务队列，重点演示任务生产者、队列服务与 Worker 消费者之间的协作方式，可用于课程演示、原型验证或后台任务处理入门。压缩包共 5 个文件，以 2 个 Python 脚本为核心，分别负责任务逻辑定义与任务投递，另含 1 个编译缓存文件、1 个 Markdown 说明文档和 1 个 Git 忽略配置，整体约 3KB，结构精简、导入方便。示例涉及 Redis 服务准备、rqworker 启动监听、队列任务创建等环节，读者可据此理解任务入队、后台执行与结果获取的基本流程，并进一步扩展失败重试、多队列调度等需求。已有 385 人学习，适合作为快速上手 RQ 的参考脚手架。

1. 从一次脚本卡死说起：Python + Redis 任务队列到底解决什么问题

有个跑了半年的定时脚本，某天开始卡在某个环节不动，日志里什么都没有，进程还在，任务就是不往前走——这种黑匣子现场，用线程池兼做任务调度的 Python 脚本几乎都会遇到。进程一重启，正在处理的活全丢；想做延迟执行，只能time.sleep；想看还有多少活没干，只能翻日志。这套python-redis-queue-sample就是把这件事拆开重做：用 Redis 的 list 和 zset 搭一套最小可用的任务队列，生产者只管入队，消费者阻塞式取任务，任务状态、失败重试、延迟执行都有明确落点。

它不适合拿去扛百万级 QPS，但特别适合三类人：手里有一堆定时脚本想改成常驻 worker 的、被重量级框架的一堆配置劝退、想先搞懂队列原理的、以及需要一个能随手改的轻量骨架做内部工具的。整个 demo 不依赖任何重型中间件，一个 Redis 实例加几个 Python 文件就能跑，改起来没有负担。

2. 项目骨架与依赖：从目录结构看这份 demo 到底装了什么

拿到一份 demo 源码，我习惯先看目录再跑代码，因为目录结构基本决定了这套东西的野心有多大。这份 sample 的分层很克制，没有为了“看起来完整”硬塞抽象层，几个文件各管一件事，读起来不费劲。

2.1 目录结构与文件职责

展开之后大概是这个样子：

python-redis-queue-sample/ ├── requirements.txt ├── config.py # 常量与 Redis 连接参数 ├── task_queue/ │ ├── __init__.py │ ├── connection.py # 连接池，全局复用 │ ├── keys.py # 所有 Redis key 集中定义 │ ├── producer.py # 入队：立即任务 + 延迟任务 │ ├── consumer.py # 出队、执行、ack、重试 │ ├── scheduler.py # 延迟队列搬运 │ ├── reaper.py # 超时任务回收 │ └── registry.py # 函数名 -> 处理函数 的映射表 ├── tasks/ │ └── handlers.py # 业务处理函数都写这里 ├── scripts/ │ ├── enqueue_demo.py # 手动塞几条任务进去 │ └── run_worker.py # 启动消费者进程 └── tests/ └── test_queue.py

keys.py单独拆出来这点很值得抄。队列相关的 key 一旦散落在各个文件里，后期排查时你会花大量时间 grep 字符串，还容易写出tq:job和tq:jobs这种一字之差的前缀，导致生产者入的队和消费者取的队不是同一个。集中定义之后，改前缀只改一处。

# task_queue/keys.py PREFIX = "tq" QUEUE_KEY = f"{PREFIX}:jobs:default" # 待处理任务（list） DELAY_KEY = f"{PREFIX}:jobs:delayed" # 延迟任务（zset，score 是到期时间戳） PROCESSING_KEY = f"{PREFIX}:jobs:processing" # 已取出未确认（list） PROCESSING_TS = f"{PREFIX}:jobs:processing:ts" # 同上，zset 记录取出时刻 DEAD_KEY = f"{PREFIX}:jobs:dead" # 超过重试上限的死信 LOCK_PREFIX = f"{PREFIX}:lock:"

逻辑上分了四类 key：待处理、延迟、处理中、死信。处理中这一组用 list 加 zset 双写，list 存原始 payload 方便重新投递，zset 存时间戳方便按时间范围扫出超时的那些。这个组合是后面可靠性设计的基础，先记住它。

2.2 依赖与连接配置：连接池和 decode_responses

依赖很少，redis客户端加一个可选的python-dotenv：

# requirements.txt redis>=4.5,<6.0 python-dotenv>=1.0

配置文件把连接参数和环境变量对齐，本地开发和容器里部署用同一套代码：

# config.py import os from dotenv import load_dotenv load_dotenv() REDIS_HOST = os.getenv("REDIS_HOST", "127.0.0.1") REDIS_PORT = int(os.getenv("REDIS_PORT", 6379)) REDIS_DB = int(os.getenv("REDIS_DB", 0)) REDIS_PASSWORD = os.getenv("REDIS_PASSWORD") or None MAX_RETRY = 3 # 单条任务最大重试次数 VISIBLE_TIMEOUT = 60 # 任务被取走多久没 ack 就认为超时，单位秒 BATCH_SIZE = 50 # 延迟队列一次搬运的最大条数

VISIBLE_TIMEOUT这个值要结合业务耗时来定。它太小会把还在正常跑的长任务误判成卡死然后重复投递，太大会让真正卡住的任务迟迟不回队列。我一般的起点是“单任务 P99 耗时 × 3”，拿不准就先设 60 秒观察一段时间。

连接池那层没什么花活，但decode_responses是个必须想清楚的开关：

# task_queue/connection.py import redis import config _pool = redis.ConnectionPool( host=config.REDIS_HOST, port=config.REDIS_PORT, db=config.REDIS_DB, password=config.REDIS_PASSWORD, decode_responses=True, # 返回 str 而不是 bytes max_connections=32, # 和 worker 并发数对齐，别超过 Redis maxclients ) def get_redis(): """所有模块共用同一个连接池，进程内只建一次""" return redis.Redis(connection_pool=_pool)

decode_responses=True意味着拿到的 payload 直接是str，可以直接json.loads。代价是没法存二进制，如果你的任务参数里有 bytes 或者要用 pickle 序列化，这里就得关掉，然后在消费端手动payload.decode()。这个开关一旦和序列化方式对不上，报错信息会非常含糊，第 5 章会单独说。

max_connections=32不是拍脑袋的数字。每个 worker 进程至少占一条阻塞连接（BRPOPLPUSH会一直占着），再算上 ack、重试、巡检用的普通连接，一个进程大概吃 2 到 3 条。三个 worker 就是十条左右，留出余量给监控脚本，32 够用。

2.3 三条命令把它跑起来

Redis 起好之后，验证顺序是这样的：先确认能连上，再入队，最后起 worker。

# 1. 确认 Redis 可达 redis-cli -h 127.0.0.1 -p 6379 ping # 期望输出 PONG # 2. 安装依赖 pip install -r requirements.txt # 3. 塞几条任务进去（立即任务 + 延迟 5 秒的任务） python -m scripts.enqueue_demo # 4. 起一个消费者 python -m scripts.run_worker

入队脚本里会打印任务 id，worker 起来后能立刻看到消费日志。想确认队列当前状态，直接看 key：

redis-cli llen tq:jobs:default # 待处理数量 redis-cli zcard tq:jobs:delayed # 延迟任务数量 redis-cli llen tq:jobs:processing # 处理中数量 redis-cli llen tq:jobs:dead # 死信数量

这四条命令建议存成 alias，后面排查基本全靠它们。正常情况下processing的数字应该是个位数并且在波动，如果它只涨不跌，说明 ack 环节出问题了，直接跳到第 5 章。

3. 核心链路拆解：一条任务从入队到被消费的完整路径

队列的所有设计取舍，最终都体现在“一条任务走完全程”这条路径上。这一章按时间顺序拆：任务怎么构造、怎么进队列、怎么被取走、失败了怎么回去。

3.1 任务载荷：固定字段加 JSON 序列化

任务在 Redis 里就是一段字符串，所以载荷设计的第一原则是自描述——消费者拿到它，不需要查库就能知道该调哪个函数、传什么参数、重试过几次。

# task_queue/producer.py import json, time, uuid from task_queue.connection import get_redis from task_queue import keys, config def _build_job(func, args, kwargs, max_retry): return { "id": uuid.uuid4().hex, # 全局唯一，排查时按这个搜日志 "func": func, # 处理函数名，靠 registry 映射 "args": args or [], "kwargs": kwargs or {}, "retry": 0, # 已重试次数 "max_retry": config.MAX_RETRY if max_retry is None else max_retry, "enqueue_at": time.time(), # 入队时刻，用来算排队延迟 "trace": uuid.uuid4().hex[:8], # 简单的链路标记 }

字段逐个说：id是排查的抓手，日志里统一带上它，出问题能一路追；func存函数名字符串而不是序列化函数对象，是为了让生产者和消费者彻底解耦，两边不用共享代码；args/kwargs必须是 JSON 可序列化的，这一点很关键，传个 datetime 进去会直接抛TypeError，得自己转成时间戳；enqueue_at用来算“任务从入队到开始执行隔了多久”，队列积压时这个指标比队列长度更直观。

3.2 入队：LPUSH 与 ZADD 两条路

入队只有两个分支：立刻要跑的和延迟要跑的。

def enqueue(func, args=None, kwargs=None, delay=0, max_retry=None): job = _build_job(func, args, kwargs, max_retry) payload = json.dumps(job, ensure_ascii=False, separators=(",", ":")) r = get_redis() if delay and delay > 0: # 延迟任务进 zset，score 是绝对到期时间戳 r.zadd(keys.DELAY_KEY, {payload: time.time() + delay}) else: # 立即任务进 list，从左端推 r.lpush(keys.QUEUE_KEY, payload) return job["id"]

separators=(",", ":")去掉 JSON 里的空格，一条任务能省几个字节，量大的时候不是小数。ensure_ascii=False让中文参数原样存储，redis-cli里看的时候可读性好很多，代价是必须保证连接层decode_responses=True，否则取出来是 bytes，json.loads会报一个看起来完全不相干的错。

方向也要对上：LPUSH从左推，消费端就得从右取，这样才是先进先出。如果两边都写左，队列就变成了后进先出，表现出来是“任务顺序莫名其妙反了”，查半天查不出原因——这种方向错配是新手最容易翻的车之一。

延迟任务用整段 payload 当 zset 的 member，这带来一个隐含约束：member 必须唯一。因为id是 uuid，正常不会重复，但如果你手动构造 payload 测试，两条一模一样的会被ZADD覆盖成一条。我在测试环境就吃过这个亏，塞了十条一样的任务，结果只跑了一条。

3.3 出队：BRPOPLPUSH 保证任务不悬空

消费端的核心只有一行，但这一行是整个方案可靠性的来源：

# task_queue/consumer.py import time from task_queue.connection import get_redis from task_queue import keys def pop(timeout=5): """阻塞取任务，同时原子地挪进 processing 队列""" r = get_redis() payload = r.brpoplpush(keys.QUEUE_KEY, keys.PROCESSING_KEY, timeout=timeout) if payload is None: return None # 超时，队列空了，正常现象 # 额外记一笔取出时刻，供 reaper 判断是否超时 r.zadd(keys.PROCESSING_TS, {payload: time.time()}) return payload

BRPOPLPUSH source destination timeout的语义是：从source右端弹出一个元素，推入destination左端，整个过程在 Redis 服务端原子完成。原子性这点非常关键——如果先RPOP再LPUSH，中间进程崩了，这条任务就从世界上消失了，没有任何痕迹。用BRPOPLPUSH，任务至少在processing里有备份，进程挂了还能捞回来。

timeout=5是阻塞等待的上限，不是任务执行超时。队列空的时候它会挂起 5 秒然后返回None，主循环拿到None就继续下一轮，CPU 占用几乎为零。这个值设太短会导致频繁轮询，设太长会让 worker 对停止信号反应迟钝，5 秒是个折中。

取出来之后的执行与确认：

def ack(payload): """任务处理成功，从 processing 里彻底移除""" r = get_redis() pipe = r.pipeline() pipe.lrem(keys.PROCESSING_KEY, 1, payload) pipe.zrem(keys.PROCESSING_TS, payload) pipe.execute()

LREM count=1而不是全部删，是因为理论上可能存在两条相同 payload（虽然 uuid 保证不会），保守一点。用 pipeline 把两个删除打包成一次往返，减少网络开销。这里有个顺序问题：先删 list 还是先删 zset？实际上只要两个都删了就行，但如果中间崩了，会出现 list 里没有、zset 里还留着的“幽灵记录”，reaper 扫到之后会尝试LREM一个不存在的元素，LREM返回 0，不报错，属于可接受的残留。

3.4 失败重试与死信队列

任务抛异常是常态，重点是别让它无声无息地消失。

def handle_failure(job, exc): """任务失败后的分流：还能重试就退避重投，超限就进死信""" r = get_redis() job["retry"] += 1 job["last_error"] = repr(exc)[:500] # 截断，避免死信队列被堆栈撑爆 payload = json.dumps(job, ensure_ascii=False, separators=(",", ":")) if job["retry"] > job["max_retry"]: r.lpush(keys.DEAD_KEY, payload) return "dead" # 指数退避：1s, 2s, 4s ... 上限 60s backoff = min(60, 2 ** job["retry"]) r.zadd(keys.DELAY_KEY, {payload: time.time() + backoff}) return "retry"

三个细节值得注意。repr(exc)而不是str(exc)，因为很多异常str()出来是空字符串，排查时等于没有信息。[:500]截断是防止某些框架的异常带超长堆栈，几百条就能把 Redis 内存吃掉一大块。指数退避是为了应对下游服务短时间抖动——如果下游挂了 30 秒，固定间隔重试会把队列瞬间打爆，退避能让压力平滑下来。

重试次数存在任务载荷里，而不是用 Redis 的某个计数器，好处是任务在队列之间搬来搬去时状态跟着走，不会因为换了个 key 就丢了重试记录。坏处是 payload 会随重试变大，但加了截断之后影响可控。

4. 可靠性设计：丢任务、重复跑、内存涨这三个坑怎么堵

上面那条链路在一切正常时跑得很好，真正考验设计的是异常路径。这一章讲三个必答题：worker 崩了任务怎么办、同一条任务跑了两遍怎么办、队列越堆越大怎么办。

4.1 超时回收：用 zset 的时间戳把卡住的任务捞回来

processing队列解决的是“任务不丢”，但没解决“任务卡住”。worker 拿到任务后如果进程被 kill，这条任务会永远躺在processing里，谁也不知道它该不该继续。

回收逻辑靠的是取出时记的那个时间戳：

# task_queue/reaper.py import time from task_queue.connection import get_redis from task_queue import keys, config def reap_once(limit=50): """把超过 VISIBLE_TIMEOUT 还没 ack 的任务重新投回主队列""" r = get_redis() deadline = time.time() - config.VISIBLE_TIMEOUT stale = r.zrangebyscore(keys.PROCESSING_TS, 0, deadline, start=0, num=limit) recovered = 0 for payload in stale: # zrem 返回 1 表示抢到了这条记录的处置权，避免多个 reaper 重复搬运 if not r.zrem(keys.PROCESSING_TS, payload): continue r.lrem(keys.PROCESSING_KEY, 1, payload) r.lpush(keys.QUEUE_KEY, payload) recovered += 1 return recovered

ZREMRANGEBYSCORE看起来更省事，但它不返回被删的元素内容，没法重新投递，所以必须用ZRANGEBYSCORE查出来再逐个ZREM。ZREM放在前面当锁用——多个巡检进程同时扫到同一条时，只有ZREM返回 1 的那个才继续搬运，其余直接跳过。这个技巧比上分布式锁轻得多，适合这种“做得重复也无所谓、但不能重复太多”的场景。

limit=50是防止某次故障后积压了几万条超时任务，一次性全捞回来把主队列打爆。分批处理，让系统有个缓冲。

提示：VISIBLE_TIMEOUT必须大于业务代码的最长执行时间。如果你的任务里有跑十分钟的报表生成，这个值至少要设到 1800 秒，否则正常任务会被当成卡死反复重投，表现出来就是“同一条任务跑了三遍”。

4.2 幂等：锁只能兜底，业务侧必须自己扛

超时回收带来一个必然结果：同一条任务可能被执行两次——第一次的 worker 其实没死，只是慢，任务被回收后又投递给了另一个 worker，两边同时跑。这是至少一次语义的固有代价，任何队列方案都躲不掉。

队列层能做的只有兜底：

# task_queue/consumer.py import functools, uuid from task_queue.connection import get_redis from task_queue import keys def task_lock(name, ttl=300): """同 name 的任务在 ttl 内只允许一个实例执行""" def decorator(fn): @functools.wraps(fn) def wrapper(*args, **kwargs): r = get_redis() key = f"{keys.LOCK_PREFIX}{name}" token = uuid.uuid4().hex if not r.set(key, token, nx=True, ex=ttl): return None # 已有实例在跑，直接放弃本次 try: return fn(*args, **kwargs) finally: # 只删自己加的锁，防止误删别人的 if r.get(key) == token: r.delete(key) return wrapper return decorator

SET key value NX EX ttl是加锁的标准写法，NX保证只有第一个能成功，EX保证进程崩溃后锁会自动过期而不是永久占着。finally里那句if r.get(key) == token是必须的——如果任务执行时间超过了ttl，锁已经过期被别人拿走了，这时候直接DEL会误删别人的锁。用 token 比对虽然仍有极小的竞态窗口（GET 和 DEL 之间），但比不比对安全得多，真要做到严格需要 Lua 脚本。

更根本的做法是在业务层做幂等：处理前先查task_id是否已经处理过，或者用数据库唯一索引挡住重复写入。锁只是减少重复概率，不能消除，指望它完全解决问题迟早翻车。

4.3 并发与批量：worker 数量、连接数和搬运节奏

单 worker 串行处理，吞吐取决于任务耗时。要提速最直接的是起多个进程：

# 三个 worker 并行消费 for i in 1 2 3; do python -m scripts.run_worker & done wait

多进程比多线程省心，因为 Python 有 GIL，CPU 密集型任务用线程根本没收益；而且每个进程有独立的连接池，崩一个不影响其他。BRPOPLPUSH本身是原子的，多个 worker 同时阻塞在同一个 key 上也不会取到同一条，这点不用担心。

延迟队列的搬运要注意节奏：

# task_queue/scheduler.py LUA_MOVE = """ local jobs = redis.call('ZRANGEBYSCORE', KEYS[1], 0, ARGV[1], 'LIMIT', 0, ARGV[2]) if #jobs == 0 then return 0 end for _, job in ipairs(jobs) do redis.call('ZREM', KEYS[1], job) redis.call('LPUSH', KEYS[2], job) end return #jobs """ def move_due(batch=50): """把到期的延迟任务搬到主队列，Lua 保证原子性""" r = get_redis() return r.eval(LUA_MOVE, 2, keys.DELAY_KEY, keys.QUEUE_KEY, time.time(), batch)

用 Lua 而不是在 Python 里循环ZRANGEBYSCORE加ZREM，是因为多实例部署时会出问题：两个 scheduler 同时扫到同一条到期任务，各自ZREM一次，第二个ZREM返回 0 但代码没检查，结果同一条被LPUSH了两次。Lua 脚本在 Redis 里单线程执行，整段搬运是原子的，这个坑就不存在了。

搬运间隔一般设 1 秒轮询一次，batch设 50 到 100。间隔太短会给 Redis 增加无谓压力，太长会让延迟任务的精度变差——如果你需要毫秒级定时，这套轮询方案就不合适，得换别的思路。

5. 常见问题排查：队列不动、任务卡住时先看这几处

前面讲的都是设计，真跑起来该踩的坑一个不会少。这一章是我自己遇到过、也帮别人排查过的几类典型问题，按“现象、原因、解决”记下来。

5.1 启动就 ConnectionError 或者 NOAUTH

现象：run_worker一起来就抛redis.exceptions.ConnectionError: Error 111 connecting to 127.0.0.1:6379，或者NOAUTH Authentication required。

原因通常有三种。一是 Redis 压根没起，或者起在别的端口上；二是 Docker 部署时把127.0.0.1写死在配置里，容器内的127.0.0.1指的是容器自己，连不到宿主机上的 Redis；三是 Redis 设了密码但配置里REDIS_PASSWORD没读到，os.getenv返回None之后连接被拒。

解决：先用redis-cli在同一个环境里验证连通性，别在宿主机验完就以为容器里也能通。容器场景把REDIS_HOST换成服务名或者宿主机 IP。密码问题在启动日志里打一行脱敏后的配置（只打 host、port、db，密码只打“已设置/未设置”），比盲猜快得多。

5.2 同一条任务被反复执行

现象：日志里同一个task_id出现两三次，业务数据被重复写入。

原因：最常见的是VISIBLE_TIMEOUT设得比任务实际耗时短。任务跑了 90 秒，超时是 60 秒，reaper 在 60 秒时就把它捞回去重投了，第二个 worker 接着跑，第一个 worker 跑完回来 ack，发现 processing 里已经没有了。次要原因是 worker 在处理过程中被重启，任务留在 processing 里到期回收，这属于设计内的正常行为。

解决：先把VISIBLE_TIMEOUT调到最长任务耗时的三倍以上，观察是否还复现。如果复现频率降下来但没归零，那说明确实存在真实的重投场景，这时候只能靠业务幂等兜住——按task_id建唯一索引，或者在处理函数入口查一次“这条任务是否已完成”。

5.3 队列积压，Redis 内存持续上涨

现象：llen tq:jobs:default一路涨，used_memory同步上涨，任务排队延迟越来越大。

原因排查顺序：先看 worker 是不是真的在跑（ps加日志），再看单个任务耗时是不是变长了（打点统计），最后看tq:jobs:dead是不是堆了几十万条。死信队列没有上限是最常见的隐形内存黑洞——任务一直失败，一直往死信里塞，没人清理，几个月后能吃掉几个 G。

解决：给死信队列加个上限，用LTRIM保留最近 N 条：

if job["retry"] > job["max_retry"]: pipe = get_redis().pipeline() pipe.lpush(keys.DEAD_KEY, payload) pipe.ltrim(keys.DEAD_KEY, 0, 9999) # 只保留最近一万条 pipe.execute()

同时给used_memory和三个队列长度配上监控，涨的时候能第一时间发现，而不是等用户投诉。

5.4 延迟任务到点了不执行

现象：zadd进去的延迟任务，到期后一直躺在tq:jobs:delayed里，zcard不下降。

原因：八成就没起scheduler。很多人只跑了 worker，以为 worker 会自己处理延迟队列——不会，搬运是独立进程干的。第二个原因是时间单位搞混，time.time()是秒，time.time() * 1000是毫秒，入队用毫秒、搬运用秒，任务就得等一千年；或者反过来，任务立刻全到期，等于没有延迟。第三个原因是 zset 的 member 重复被覆盖，前面提过。

解决：run_worker.py里顺手把 scheduler 的循环也放进同一个进程（用线程或者 asyncio 任务），别让部署时漏掉。时间戳统一在producer.py里算好，其他地方只读不写。起服务后先塞一条delay=5的任务，五秒后确认zcard归零，这个自检步骤固定写进启动脚本。

5.5 中文参数乱码或者 json 解析失败

现象：消费者报json.decoder.JSONDecodeError: Expecting value: line 1 column 1，或者拿到的payload是b'{"func": ...}'这种带b前缀的 bytes。

原因：decode_responses和序列化方式没对齐。连接池设了decode_responses=False，BRPOPLPUSH返回 bytes，json.loads虽然能接受 bytes，但如果中间做过字符串拼接或者截断就会出错。反过来，如果存进去的时候用的是 pickle，读的时候按 str 解，那就是彻底对不上。

解决：全链路统一用 JSON 加str。ConnectionPool(decode_responses=True)配上json.dumps(..., ensure_ascii=False)，读写两端都按文本处理。如果确有二进制参数，不要塞进 payload，转成 base64 字符串存，消费端再解回来。这个约定写进 README，比事后排查省事。

6. 二次开发：把 demo 改成自己项目里能长期跑的 worker

这个 sample 的定位是骨架，直接用会缺东西。我一般会在它基础上做三处改造，成本都不高，但决定了它能不能长期跑下去。

第一处是把registry.py从字典改成装饰器注册：

# task_queue/registry.py _HANDLERS = {} def task(name=None): """用 @task("send_mail") 注册处理函数，避免手写映射表""" def decorator(fn): _HANDLERS[name or fn.__name__] = fn return fn return decorator def get_handler(name): if name not in _HANDLERS: raise KeyError(f"未注册的任务类型: {name}") return _HANDLERS[name]

业务代码里写@task("send_mail")就能注册，生产者那边传enqueue("send_mail", args=[...])。好处是新加任务类型不用改两处，坏处是必须有地方 import 到tasks/handlers.py，否则装饰器不会执行、注册表是空的——这一点在run_worker.py里显式import tasks.handlers解决。

第二处是加信号处理做优雅退出。现在的while True收到Ctrl+C会直接把正在跑的任务打断，任务留在 processing 里等超时回收。改成收到SIGTERM后置一个标志位，主循环跑完当前任务就退出，能在容器滚动更新时少制造一堆重投。配合VISIBLE_TIMEOUT适当放宽，效果更明显。

第三处是补监控，最小实现是定时把四个队列长度打给日志或者报表系统。我把几个关键指标的合理区间记成了表，超了就告警：

最后说个习惯。我现在每次改完消费者逻辑，都会先手动塞三类任务做一遍回归：一条正常任务、一条必失败的任务（验证重试和死信）、一条带延迟的任务（验证 scheduler）。三条跑完，processing和dead的计数都对得上，才敢上环境。这套自检流程看着笨，但帮我省过好几次“上线才发现 ack 写错 key”的后悔药。这个 demo 的价值不在于它能跑多快，而在于每个环节都看得见、改得动，拿去当自己项目的起点正合适，希望帮到你。

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