简介这是一份基于 Python 与 Redis 的轻量级任务队列示例资源主要面向需要快速实现异步任务处理的 Python 开发者。项目利用 RQRedis Queue库展示了如何将耗时任务放入队列并由独立 Worker 消费执行适用于定时任务、后台数据清洗、消息通知等常见场景。资源共 5 个文件包含 create.py 任务创建脚本、handlers.py 任务处理函数、编译生成的 pyc 文件以及 README.md 说明文档和 gitignore 配置整体压缩包仅 3KB结构精简、易于阅读和改造。目前已有 385 人学习下载。通过该示例你可以理解 RQ 的基本工作流程连接 Redis、定义队列、注册任务函数、启动 rqworker 监听并执行同时还能掌握依赖安装与 Redis 容器化启动等环境准备细节适合入门级开发者作为学习参考或项目脚手架。1. 一个 Python Redis 任务队列 demo几个文件就能撑起异步消费任务队列这个概念很多新手以为一定要上重型分布式框架其实一个 Redis 的 List 就能解决大多数场景。这个项目就是一个非常典型的 Python Redis 任务队列 demo生产者往队列里丢任务消费者阻塞拉取任务并执行失败的任务自动重试超过次数就进死信队列。整套代码只有几个文件不依赖额外的消息中间件适合刚接触任务队列的 Python 开发者也适合那些为了“发个邮件、导出个报表”不想引入大型框架的轻量业务场景。跑通这个 demo你对队列模型的生产、消费、重试、消息格式设计会有一个完整的认知之后再接触分布式调度也不会觉得黑匣子那么多。2. 系统骨架五个文件一条队列为什么这套结构能跑2.1 技术选型为什么 Redis 的 List 够用先看实现思路。任务队列要回答三个问题任务放哪里、怎么通知消费者、任务处理失败怎么办。这个项目把答案简化成任务放在一个 Redis 的 List 里消费者用 BRPOP 在队列上阻塞等待失败任务把 attempt 次数加一后重新入队超过上限就丢进死信队列。List 在这个场景里其实是一个天然的消息通道。LPUSH 从左边写入BRPOP 从右边取出数据先进先出顺序可控BRPOP 是阻塞读队列空时不消耗 CPU一旦有数据写入立刻返回比轮询 Redis 键优雅得多。这两个命令都是原子的起多个消费者并发取任务时不会出现两个 worker 取到同一个任务的场景天然满足多消费者分摊负载的需求。为什么不直接上更重的分布式任务框架因为那些框架动辄几十张配置表、一堆 worker 和调度器的概念要理解。如果你只是想让耗时操作脱离 HTTP 请求线程一个 Redis 实例加几个 Python 文件真的是最轻的一条路。这个 demo 就卡在这层粒度上不贪多够用。2.2 目录结构五个文件各管一段python-redis-queue-sample/ ├── config.py # Redis 连接参数、队列键名、最大重试次数 ├── add_task.py # 生产者脚本构造任务并 LPUSH 入队 ├── worker.py # 消费者脚本BRPOP 取任务、分发、重试 ├── requirements.txt # 依赖声明 └── README.md # 运行说明文件职责拆开看更清楚文件职责config.py集中管理 Redis 连接参数与队列键名生产者和消费者共用add_task.py模拟业务侧推送任务实际项目中可以替换成 API 接口worker.py消费队列、调用对应 handler、失败重试和死信转移requirements.txt运行依赖核心只有 redis 库config.py 是这段设计的核心。队列键名、重试次数、Redis 地址都收敛到一个文件里而不是散落在两个脚本中这样把 worker 从一台机器换到另一台机器时只改环境变量就够了。实际项目里我一般还会加一个日志配置demo 为了保持可读性没有放进去。这套 demo 的消息不是原生的 Redis 字符串而是把任务结构化之后塞进去。每个任务是一个 JSON 字符串里面包含 task_id、task_type、payload、attempt、created_at 五个字段。task_id 用time_ns()生成在幂等扩展时可以用它去重task_type 决定 worker 调用哪个处理函数payload 是业务参数attempt 记录已经执行失败了几次created_at 记录入队时间方便后续排查积压。数据流也非常直白add_task 脚本把任务 LPUSH 进sample:task:queueworker 进程 BRPOP 取出根据 task_type 查表执行对应 handler执行异常时 attempt 加一重新入队超过 MAX_RETRY 后 LPUSH 进sample:task:dead死信队列。整个链路没有一个多余的环节。2.3 快速跑通Redis 自备两个命令把队列转起来这个 demo 不打包 Redis 环境标题里的“自备”指的就是它。想跑起来先确认本机有一个可用的 redis-server没有就自己装一个。如果 Redis 在远程机器或带密码把环境变量改掉就行。pip install -r requirements.txt # 终端 1启动 Redis常见系统里直接执行 redis-server # 终端 2启动消费者 python worker.py # 终端 3推送测试任务 python add_task.pyworker 启动后控制台打印“worker 已启动正在监听队列”然后阻塞等待add_task 脚本运行后会打印“已推送 2 个任务”终端 2 紧接着打印邮件日志表示任务已经被消费。生产者和消费者是两个独立进程之间只通过 Redis 通信这就是一个最朴素的异步任务闭环。注意如果 redis-server 不在 PATH 里用绝对路径启动或者用 docker 起一个 Redis 容器端口、密码、db 参数在 config.py 里通过环境变量配置。3. 核心代码拆解一个任务从入队到被消费的完整链路3.1 生产者把任务序列化成 JSON 再 LPUSH 入队import json import time from config import get_redis, QUEUE_KEY def add_task(task_type: str, payload: dict) - dict: task { task_id: str(time.time_ns()), task_type: task_type, payload: payload, attempt: 0, created_at: time.time(), } r get_redis() r.lpush(QUEUE_KEY, json.dumps(task, ensure_asciiFalse)) return task if __name__ __main__: add_task(send_email, {to: userexample.com, content: 欢迎注册}) add_task(send_email, {to: adminexample.com, content: 新用户注册}) print(已推送 2 个任务到队列可启动 worker 消费)生产者脚本的核心是 add_task 函数。构造 task 字典时task_id 直接用time_ns()在单机 demo 里不会撞attempt 初始化为 0后面消费者每失败一次就加一。json.dumps时开了ensure_asciiFalse这样任务里带中文时队列里依然是可读文本而不是\uXXXX转义排查问题时用 redis-cli 看一眼队列内容会舒服很多。Redis 只认字符串所以一切结构化数据最终都要序列化。选择 JSON 而不是 pickle 有一个简单理由JSON 跨语言通用队列里万一混进脏数据也容易定位如果任务里全是自定义对象序列化接入其它语言或调试时的迁移成本会高不少。参数说明也顺带提一下LPUSH 把 JSON 字符串从 List 左侧写入后续消费者从右侧 BRPOP整体先进先出。如果业务要求 LIFO把 LPUSH 改成 RPUSH 就能颠倒顺序其它代码不用动这也是队列键名收敛在 config.py 里的好处。3.2 消费者BRPOP 阻塞拉取与按类型分发import json import time from config import get_redis, QUEUE_KEY, DEAD_QUEUE_KEY, MAX_RETRY def handle_email(payload: dict): # 模拟邮件发送真实场景这里会调用外部邮件接口 print(给, payload[to], 发送邮件:, payload[content]) time.sleep(0.5) HANDLERS { send_email: handle_email, } def process(task_json: str): task json.loads(task_json) handler HANDLERS.get(task[task_type]) if handler is None: raise ValueError(unknown task_type: task[task_type]) handler(task[payload]) def run_worker(): r get_redis() print(worker 已启动正在监听队列, QUEUE_KEY) while True: item r.brpop(QUEUE_KEY, timeout1) if item is None: # 队列暂时为空BRPOP 超时返回 None继续下一轮 continue key, task_json item[0], item[1] print(收到任务:, task_json) try: process(task_json) except Exception as exc: print(任务执行失败:, exc) task json.loads(task_json) task[attempt] 1 if task[attempt] MAX_RETRY: r.lpush(QUEUE_KEY, json.dumps(task, ensure_asciiFalse)) print(准备第, task[attempt], 次重试) else: r.lpush(DEAD_QUEUE_KEY, json.dumps(task, ensure_asciiFalse)) print(超过最大重试次数任务进入死信队列) if __name__ __main__: run_worker()BRPOP 返回的是一个二元组第一个元素是键名第二个是弹出的值。因为 Redis 的 BRPOP 支持同时监听多个队列所以返回值里带着 key 表示消息来自哪条队列这个 demo 只监听一条但结构上保留了扩展空间后面接多队列只需要把队列名按优先级排进 BRPOP 参数列表就行。timeout1的含义是队列里暂时没数据时阻塞最多 1 秒超时后返回 None。demo 里用 1 秒是为了 CtrlC 退出时响应快长驻场景建议直接传 0BRPOP 会一直阻塞到有新任务不耗 CPU。分发逻辑是这段代码的亮点。HANDLERS 是一个字典把 task_type 映射到处理函数。process 取出 task_type 后查表找到就执行找不到就抛异常。新增一种任务类型只需要写一个 handler 函数并注册到字典里while 循环、重试、死信逻辑完全不用碰。真实项目里我一般还会在 handler 外层套一层耗时统计方便定位慢任务。3.3 失败重试与死信attempt、MAX_RETRY 与兜底队列except Exception as exc: print(任务执行失败:, exc) task json.loads(task_json) task[attempt] 1 if task[attempt] MAX_RETRY: r.lpush(QUEUE_KEY, json.dumps(task, ensure_asciiFalse)) print(准备第, task[attempt], 次重试) else: r.lpush(DEAD_QUEUE_KEY, json.dumps(task, ensure_asciiFalse)) print(超过最大重试次数任务进入死信队列)这段逻辑做的事情很直白任务执行抛异常时把 attempt 加一如果还没超过 MAX_RETRY就把任务重新入队如果超过上限说明这个任务一直处理失败再留在主队列只会反复污染消费流程于是转进死信队列留给人去排查。这里最值得讨论的不是 if 本身而是重试任务回到了哪个位置。r.lpush 会把失败任务插到队列头部问题是它会阻塞队列里其它新任务对邮件这种低频任务无所谓对高吞吐场景就需要 rpush 到队尾或者用一个单独的 retry 列表配合延迟处理。这一点在下一章避坑部分继续展开。另一个容易被忽略的是重试的幂等性。处理函数在重试时会被再次执行如果 handler 里的操作不是幂等的比如重复扣款、重复插入记录就会出问题。实际改造时我会要求每个任务带上业务侧的唯一标识在 handler 里先做一次查重保证同一个 task_id 最多只生效一次。4. 常见问题与避坑跑任务队列最容易翻车的五个细节4.1 Redis 没起来连接报错的第一现场现象运行 add_task.py 或 worker.py 时直接抛出redis.exceptions.ConnectionError提示连接 127.0.0.1:6379 失败。原因标题里的“自备”指的就是 Redis 环境很多新手忽略了本地需要有一个正在运行的 redis-server。端口 6379 没有进程在监听客户端连接自然失败。解决先执行redis-cli ping返回 PONG 说明 Redis 正常如果没起来启动 redis-server。Redis 在远程机器或者带密码时通过环境变量注入连接信息比如REDIS_HOST192.168.x.x REDIS_PASSWORDxxx python worker.pyconfig.py 里已经预留了这些变量的读取入口。4.2 decode_responses 没开json.loads 处理 bytes 的怪问题现象消费任务时json.loads抛TypeError或者打印出来的任务内容带b...前缀。原因redis.Redis 默认decode_responsesFalse从 Redis 读出来的数据是 bytes 类型而我习惯在加任务时直接序列化成字符串两边类型对不上。真正踩坑的人常以为是自己 JSON 写错了其实只是编码问题。解决在 get_redis() 里统一加decode_responsesTrue让所有读写都按字符串处理。如果不开这个参数每次 BRPOP 出来的值都要手动.decode()轮询一多非常容易漏。提示这个参数改动虽然不起眼但直接影响整个二次开发阶段的数据处理方式建议一开始就固定下来。4.3 BRPOP 的 timeout 用 1队列空闲时进程 CPU 偏高现象队列里没任务时worker 进程的 CPU 占用比预期的活跃或者日志里不断刷“等待”相关的空转输出。原因BRPOP timeout1 表示阻塞 1 秒后返回 None代码里 continue 后马上又进入下一轮 BRPOP。表面上看是阻塞实际是高频循环加短暂阻塞空转频率明显高于 timeout0 的写法。解决常驻进程把 timeout 改成 0BRPOP 会一直阻塞到数据到达等待期间不消耗 CPU。demo 里写 1 仅仅是为了测试时 CtrlC 响应快长驻服务一定记得改掉。4.4 失败任务重试插队新任务被反复阻塞现象某个任务一直执行失败同一时间段入队的新任务迟迟不被消费队列长度看起来正常但就是不出活。原因重试逻辑用 lpush 把失败任务插到队列头部下次 BRPOP 又取到它。如果它持续失败就会反复占用消费者的处理时间后面排着的新任务全部被堵住。解决低峰场景无所谓高峰期把重试任务 rpush 到队尾或者进入单独的延迟队列等若干秒再回主队列。更重要的是确认重试的幂等性处理函数要能接受同一个任务执行多次不产生副作用。4.5 多 worker 真的不会丢任务吗BRPOP 弹出即删除现象把 worker 起成多个进程后某个 handler 在 glitch 场景下崩溃退出发现这个任务永久消失了重试逻辑完全没机会执行。原因BRPOP 把消息从队列取出的同时就删除了处理期间没有任何“正在处理中”的记录。进程在 handler 执行一半时挂掉消息已经离开主队列自然谁也没法消费它。解决可靠场景换成 RPOPLPUSH先原子地把消息搬到 processing 列表处理完再用 LREM 删除这是“至少一次投递”的雏形。生产上还要配合 task_id 做幂等保证重复执行不产生副作用。5. 二次开发从 Demo 到能用的任务系统5.1 延迟任务用 ZSET 当时间轮想把“30 秒后发送邮件”这种延迟任务做出来常见做法是加一个 Redis ZSETscore 存到期时间戳值存任务内容。worker 每秒取一次已到期的任务重新投递到主队列import time from config import get_redis, DELAY_QUEUE_KEY, QUEUE_KEY def poll_delay_queue(): r get_redis() now time.time() tasks r.zrangebyscore(DELAY_QUEUE_KEY, 0, now, start0, num10) for task_json in tasks: if r.zrem(DELAY_QUEUE_KEY, task_json): r.lpush(QUEUE_KEY, task_json)5.2 优先级多个队列按序 BRPOP如果需要高优任务先处理可以建 high、normal、low 三条 ListBRPOP 支持一次传多个 key越靠左优先级越高r.brpop(queue:high, queue:normal, queue:low, timeout0)优先级队列在真正的业务场景里非常常用这段代码已经是在 demo 基础上最小成本的扩展了。5.3 更可靠的确认机制RPOPLPUSH 备份BRPOP 弹出即删除的缺点可以通过 RPOPLPUSH 弥补它原子地把主队列末尾的任务搬到 processing 列表处理逻辑成功后再移除备份task_json r.rpoplpush(QUEUE_KEY, PROCESSING_KEY) try: process(task_json) r.lrem(PROCESSING_KEY, 1, task_json) except Exception: # 备份队列里的数据不会因宕机丢失可以重启 worker 后重新处理 pass我自己的习惯是收到新需求先问一句这个队列真的需要多完整之前有个团队一上来就搭了个带 web 界面、多节点调度的任务中心结果业务量每天不到上百条运维成本比开发成本还高。从那以后我每写一个任务系统都先强制走一遍这份 demo 的流程生产者、消费者、失败兜底三个环节跑通再加功能。排序、延迟、可靠投递都是在这三个环节上叠加的基础没立住加再多都是空中楼阁。希望本次拆解对你有帮助。本文还有配套的精品资源点击获取 
长河君 · 2026/10/16 3:17:37 · 阅读 1.2k 用Python和Redis手写一个轻量任务队列，附完整代码
简介这是一份基于 Python 与 Redis 的轻量级任务队列示例资源主要面向需要快速实现异步任务处理的 Python 开发者。项目利用 RQRedis Queue库展示了如何将耗时任务放入队列并由独立 Worker 消费执行适用于定时任务、后台数据清洗、消息通知等常见场景。资源共 5 个文件包含 create.py 任务创建脚本、handlers.py 任务处理函数、编译生成的 pyc 文件以及 README.md 说明文档和 gitignore 配置整体压缩包仅 3KB结构精简、易于阅读和改造。目前已有 385 人学习下载。通过该示例你可以理解 RQ 的基本工作流程连接 Redis、定义队列、注册任务函数、启动 rqworker 监听并执行同时还能掌握依赖安装与 Redis 容器化启动等环境准备细节适合入门级开发者作为学习参考或项目脚手架。1. 一个 Python Redis 任务队列 demo几个文件就能撑起异步消费任务队列这个概念很多新手以为一定要上重型分布式框架其实一个 Redis 的 List 就能解决大多数场景。这个项目就是一个非常典型的 Python Redis 任务队列 demo生产者往队列里丢任务消费者阻塞拉取任务并执行失败的任务自动重试超过次数就进死信队列。整套代码只有几个文件不依赖额外的消息中间件适合刚接触任务队列的 Python 开发者也适合那些为了“发个邮件、导出个报表”不想引入大型框架的轻量业务场景。跑通这个 demo你对队列模型的生产、消费、重试、消息格式设计会有一个完整的认知之后再接触分布式调度也不会觉得黑匣子那么多。2. 系统骨架五个文件一条队列为什么这套结构能跑2.1 技术选型为什么 Redis 的 List 够用先看实现思路。任务队列要回答三个问题任务放哪里、怎么通知消费者、任务处理失败怎么办。这个项目把答案简化成任务放在一个 Redis 的 List 里消费者用 BRPOP 在队列上阻塞等待失败任务把 attempt 次数加一后重新入队超过上限就丢进死信队列。List 在这个场景里其实是一个天然的消息通道。LPUSH 从左边写入BRPOP 从右边取出数据先进先出顺序可控BRPOP 是阻塞读队列空时不消耗 CPU一旦有数据写入立刻返回比轮询 Redis 键优雅得多。这两个命令都是原子的起多个消费者并发取任务时不会出现两个 worker 取到同一个任务的场景天然满足多消费者分摊负载的需求。为什么不直接上更重的分布式任务框架因为那些框架动辄几十张配置表、一堆 worker 和调度器的概念要理解。如果你只是想让耗时操作脱离 HTTP 请求线程一个 Redis 实例加几个 Python 文件真的是最轻的一条路。这个 demo 就卡在这层粒度上不贪多够用。2.2 目录结构五个文件各管一段python-redis-queue-sample/ ├── config.py # Redis 连接参数、队列键名、最大重试次数 ├── add_task.py # 生产者脚本构造任务并 LPUSH 入队 ├── worker.py # 消费者脚本BRPOP 取任务、分发、重试 ├── requirements.txt # 依赖声明 └── README.md # 运行说明文件职责拆开看更清楚文件职责config.py集中管理 Redis 连接参数与队列键名生产者和消费者共用add_task.py模拟业务侧推送任务实际项目中可以替换成 API 接口worker.py消费队列、调用对应 handler、失败重试和死信转移requirements.txt运行依赖核心只有 redis 库config.py 是这段设计的核心。队列键名、重试次数、Redis 地址都收敛到一个文件里而不是散落在两个脚本中这样把 worker 从一台机器换到另一台机器时只改环境变量就够了。实际项目里我一般还会加一个日志配置demo 为了保持可读性没有放进去。这套 demo 的消息不是原生的 Redis 字符串而是把任务结构化之后塞进去。每个任务是一个 JSON 字符串里面包含 task_id、task_type、payload、attempt、created_at 五个字段。task_id 用time_ns()生成在幂等扩展时可以用它去重task_type 决定 worker 调用哪个处理函数payload 是业务参数attempt 记录已经执行失败了几次created_at 记录入队时间方便后续排查积压。数据流也非常直白add_task 脚本把任务 LPUSH 进sample:task:queueworker 进程 BRPOP 取出根据 task_type 查表执行对应 handler执行异常时 attempt 加一重新入队超过 MAX_RETRY 后 LPUSH 进sample:task:dead死信队列。整个链路没有一个多余的环节。2.3 快速跑通Redis 自备两个命令把队列转起来这个 demo 不打包 Redis 环境标题里的“自备”指的就是它。想跑起来先确认本机有一个可用的 redis-server没有就自己装一个。如果 Redis 在远程机器或带密码把环境变量改掉就行。pip install -r requirements.txt # 终端 1启动 Redis常见系统里直接执行 redis-server # 终端 2启动消费者 python worker.py # 终端 3推送测试任务 python add_task.pyworker 启动后控制台打印“worker 已启动正在监听队列”然后阻塞等待add_task 脚本运行后会打印“已推送 2 个任务”终端 2 紧接着打印邮件日志表示任务已经被消费。生产者和消费者是两个独立进程之间只通过 Redis 通信这就是一个最朴素的异步任务闭环。注意如果 redis-server 不在 PATH 里用绝对路径启动或者用 docker 起一个 Redis 容器端口、密码、db 参数在 config.py 里通过环境变量配置。3. 核心代码拆解一个任务从入队到被消费的完整链路3.1 生产者把任务序列化成 JSON 再 LPUSH 入队import json import time from config import get_redis, QUEUE_KEY def add_task(task_type: str, payload: dict) - dict: task { task_id: str(time.time_ns()), task_type: task_type, payload: payload, attempt: 0, created_at: time.time(), } r get_redis() r.lpush(QUEUE_KEY, json.dumps(task, ensure_asciiFalse)) return task if __name__ __main__: add_task(send_email, {to: userexample.com, content: 欢迎注册}) add_task(send_email, {to: adminexample.com, content: 新用户注册}) print(已推送 2 个任务到队列可启动 worker 消费)生产者脚本的核心是 add_task 函数。构造 task 字典时task_id 直接用time_ns()在单机 demo 里不会撞attempt 初始化为 0后面消费者每失败一次就加一。json.dumps时开了ensure_asciiFalse这样任务里带中文时队列里依然是可读文本而不是\uXXXX转义排查问题时用 redis-cli 看一眼队列内容会舒服很多。Redis 只认字符串所以一切结构化数据最终都要序列化。选择 JSON 而不是 pickle 有一个简单理由JSON 跨语言通用队列里万一混进脏数据也容易定位如果任务里全是自定义对象序列化接入其它语言或调试时的迁移成本会高不少。参数说明也顺带提一下LPUSH 把 JSON 字符串从 List 左侧写入后续消费者从右侧 BRPOP整体先进先出。如果业务要求 LIFO把 LPUSH 改成 RPUSH 就能颠倒顺序其它代码不用动这也是队列键名收敛在 config.py 里的好处。3.2 消费者BRPOP 阻塞拉取与按类型分发import json import time from config import get_redis, QUEUE_KEY, DEAD_QUEUE_KEY, MAX_RETRY def handle_email(payload: dict): # 模拟邮件发送真实场景这里会调用外部邮件接口 print(给, payload[to], 发送邮件:, payload[content]) time.sleep(0.5) HANDLERS { send_email: handle_email, } def process(task_json: str): task json.loads(task_json) handler HANDLERS.get(task[task_type]) if handler is None: raise ValueError(unknown task_type: task[task_type]) handler(task[payload]) def run_worker(): r get_redis() print(worker 已启动正在监听队列, QUEUE_KEY) while True: item r.brpop(QUEUE_KEY, timeout1) if item is None: # 队列暂时为空BRPOP 超时返回 None继续下一轮 continue key, task_json item[0], item[1] print(收到任务:, task_json) try: process(task_json) except Exception as exc: print(任务执行失败:, exc) task json.loads(task_json) task[attempt] 1 if task[attempt] MAX_RETRY: r.lpush(QUEUE_KEY, json.dumps(task, ensure_asciiFalse)) print(准备第, task[attempt], 次重试) else: r.lpush(DEAD_QUEUE_KEY, json.dumps(task, ensure_asciiFalse)) print(超过最大重试次数任务进入死信队列) if __name__ __main__: run_worker()BRPOP 返回的是一个二元组第一个元素是键名第二个是弹出的值。因为 Redis 的 BRPOP 支持同时监听多个队列所以返回值里带着 key 表示消息来自哪条队列这个 demo 只监听一条但结构上保留了扩展空间后面接多队列只需要把队列名按优先级排进 BRPOP 参数列表就行。timeout1的含义是队列里暂时没数据时阻塞最多 1 秒超时后返回 None。demo 里用 1 秒是为了 CtrlC 退出时响应快长驻场景建议直接传 0BRPOP 会一直阻塞到有新任务不耗 CPU。分发逻辑是这段代码的亮点。HANDLERS 是一个字典把 task_type 映射到处理函数。process 取出 task_type 后查表找到就执行找不到就抛异常。新增一种任务类型只需要写一个 handler 函数并注册到字典里while 循环、重试、死信逻辑完全不用碰。真实项目里我一般还会在 handler 外层套一层耗时统计方便定位慢任务。3.3 失败重试与死信attempt、MAX_RETRY 与兜底队列except Exception as exc: print(任务执行失败:, exc) task json.loads(task_json) task[attempt] 1 if task[attempt] MAX_RETRY: r.lpush(QUEUE_KEY, json.dumps(task, ensure_asciiFalse)) print(准备第, task[attempt], 次重试) else: r.lpush(DEAD_QUEUE_KEY, json.dumps(task, ensure_asciiFalse)) print(超过最大重试次数任务进入死信队列)这段逻辑做的事情很直白任务执行抛异常时把 attempt 加一如果还没超过 MAX_RETRY就把任务重新入队如果超过上限说明这个任务一直处理失败再留在主队列只会反复污染消费流程于是转进死信队列留给人去排查。这里最值得讨论的不是 if 本身而是重试任务回到了哪个位置。r.lpush 会把失败任务插到队列头部问题是它会阻塞队列里其它新任务对邮件这种低频任务无所谓对高吞吐场景就需要 rpush 到队尾或者用一个单独的 retry 列表配合延迟处理。这一点在下一章避坑部分继续展开。另一个容易被忽略的是重试的幂等性。处理函数在重试时会被再次执行如果 handler 里的操作不是幂等的比如重复扣款、重复插入记录就会出问题。实际改造时我会要求每个任务带上业务侧的唯一标识在 handler 里先做一次查重保证同一个 task_id 最多只生效一次。4. 常见问题与避坑跑任务队列最容易翻车的五个细节4.1 Redis 没起来连接报错的第一现场现象运行 add_task.py 或 worker.py 时直接抛出redis.exceptions.ConnectionError提示连接 127.0.0.1:6379 失败。原因标题里的“自备”指的就是 Redis 环境很多新手忽略了本地需要有一个正在运行的 redis-server。端口 6379 没有进程在监听客户端连接自然失败。解决先执行redis-cli ping返回 PONG 说明 Redis 正常如果没起来启动 redis-server。Redis 在远程机器或者带密码时通过环境变量注入连接信息比如REDIS_HOST192.168.x.x REDIS_PASSWORDxxx python worker.pyconfig.py 里已经预留了这些变量的读取入口。4.2 decode_responses 没开json.loads 处理 bytes 的怪问题现象消费任务时json.loads抛TypeError或者打印出来的任务内容带b...前缀。原因redis.Redis 默认decode_responsesFalse从 Redis 读出来的数据是 bytes 类型而我习惯在加任务时直接序列化成字符串两边类型对不上。真正踩坑的人常以为是自己 JSON 写错了其实只是编码问题。解决在 get_redis() 里统一加decode_responsesTrue让所有读写都按字符串处理。如果不开这个参数每次 BRPOP 出来的值都要手动.decode()轮询一多非常容易漏。提示这个参数改动虽然不起眼但直接影响整个二次开发阶段的数据处理方式建议一开始就固定下来。4.3 BRPOP 的 timeout 用 1队列空闲时进程 CPU 偏高现象队列里没任务时worker 进程的 CPU 占用比预期的活跃或者日志里不断刷“等待”相关的空转输出。原因BRPOP timeout1 表示阻塞 1 秒后返回 None代码里 continue 后马上又进入下一轮 BRPOP。表面上看是阻塞实际是高频循环加短暂阻塞空转频率明显高于 timeout0 的写法。解决常驻进程把 timeout 改成 0BRPOP 会一直阻塞到数据到达等待期间不消耗 CPU。demo 里写 1 仅仅是为了测试时 CtrlC 响应快长驻服务一定记得改掉。4.4 失败任务重试插队新任务被反复阻塞现象某个任务一直执行失败同一时间段入队的新任务迟迟不被消费队列长度看起来正常但就是不出活。原因重试逻辑用 lpush 把失败任务插到队列头部下次 BRPOP 又取到它。如果它持续失败就会反复占用消费者的处理时间后面排着的新任务全部被堵住。解决低峰场景无所谓高峰期把重试任务 rpush 到队尾或者进入单独的延迟队列等若干秒再回主队列。更重要的是确认重试的幂等性处理函数要能接受同一个任务执行多次不产生副作用。4.5 多 worker 真的不会丢任务吗BRPOP 弹出即删除现象把 worker 起成多个进程后某个 handler 在 glitch 场景下崩溃退出发现这个任务永久消失了重试逻辑完全没机会执行。原因BRPOP 把消息从队列取出的同时就删除了处理期间没有任何“正在处理中”的记录。进程在 handler 执行一半时挂掉消息已经离开主队列自然谁也没法消费它。解决可靠场景换成 RPOPLPUSH先原子地把消息搬到 processing 列表处理完再用 LREM 删除这是“至少一次投递”的雏形。生产上还要配合 task_id 做幂等保证重复执行不产生副作用。5. 二次开发从 Demo 到能用的任务系统5.1 延迟任务用 ZSET 当时间轮想把“30 秒后发送邮件”这种延迟任务做出来常见做法是加一个 Redis ZSETscore 存到期时间戳值存任务内容。worker 每秒取一次已到期的任务重新投递到主队列import time from config import get_redis, DELAY_QUEUE_KEY, QUEUE_KEY def poll_delay_queue(): r get_redis() now time.time() tasks r.zrangebyscore(DELAY_QUEUE_KEY, 0, now, start0, num10) for task_json in tasks: if r.zrem(DELAY_QUEUE_KEY, task_json): r.lpush(QUEUE_KEY, task_json)5.2 优先级多个队列按序 BRPOP如果需要高优任务先处理可以建 high、normal、low 三条 ListBRPOP 支持一次传多个 key越靠左优先级越高r.brpop(queue:high, queue:normal, queue:low, timeout0)优先级队列在真正的业务场景里非常常用这段代码已经是在 demo 基础上最小成本的扩展了。5.3 更可靠的确认机制RPOPLPUSH 备份BRPOP 弹出即删除的缺点可以通过 RPOPLPUSH 弥补它原子地把主队列末尾的任务搬到 processing 列表处理逻辑成功后再移除备份task_json r.rpoplpush(QUEUE_KEY, PROCESSING_KEY) try: process(task_json) r.lrem(PROCESSING_KEY, 1, task_json) except Exception: # 备份队列里的数据不会因宕机丢失可以重启 worker 后重新处理 pass我自己的习惯是收到新需求先问一句这个队列真的需要多完整之前有个团队一上来就搭了个带 web 界面、多节点调度的任务中心结果业务量每天不到上百条运维成本比开发成本还高。从那以后我每写一个任务系统都先强制走一遍这份 demo 的流程生产者、消费者、失败兜底三个环节跑通再加功能。排序、延迟、可靠投递都是在这三个环节上叠加的基础没立住加再多都是空中楼阁。希望本次拆解对你有帮助。本文还有配套的精品资源点击获取