简介：这是一份基于 Python 与 Redis 的轻量级任务队列示例资源，主要面向需要快速实现异步任务处理的 Python 开发者。项目利用 RQ（Redis 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:queue，worker 进程 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_ascii=False)) return task if __name__ == "__main__": add_task("send_email", {"to": "user@example.com", "content": "欢迎注册"}) add_task("send_email", {"to": "admin@example.com", "content": "新用户注册"}) print("已推送 2 个任务到队列，可启动 worker 消费")生产者脚本的核心是 add_task 函数。构造 task 字典时，task_id 直接用time_ns()，在单机 demo 里不会撞；attempt 初始化为 0，后面消费者每失败一次就加一。json.dumps时开了ensure_ascii=False，这样任务里带中文时队列里依然是可读文本，而不是\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, timeout=1) 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_ascii=False)) print("准备第", task["attempt"], "次重试") else: r.lpush(DEAD_QUEUE_KEY, json.dumps(task, ensure_ascii=False)) print("超过最大重试次数，任务进入死信队列") if __name__ == "__main__": run_worker()BRPOP 返回的是一个二元组，第一个元素是键名，第二个是弹出的值。因为 Redis 的 BRPOP 支持同时监听多个队列，所以返回值里带着 key 表示消息来自哪条队列；这个 demo 只监听一条，但结构上保留了扩展空间，后面接多队列只需要把队列名按优先级排进 BRPOP 参数列表就行。
timeout=1的含义是：队列里暂时没数据时，阻塞最多 1 秒，超时后返回 None。demo 里用 1 秒是为了 Ctrl+C 退出时响应快，长驻场景建议直接传 0，BRPOP 会一直阻塞到有新任务，不耗 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_ascii=False)) print("准备第", task["attempt"], "次重试") else: r.lpush(DEAD_QUEUE_KEY, json.dumps(task, ensure_ascii=False)) 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_HOST=192.168.x.x REDIS_PASSWORD=xxx python worker.py，config.py 里已经预留了这些变量的读取入口。
4.2 decode_responses 没开，json.loads 处理 bytes 的怪问题
现象：消费任务时json.loads抛TypeError，或者打印出来的任务内容带b'...'前缀。
原因：redis.Redis 默认decode_responses=False，从 Redis 读出来的数据是 bytes 类型，而我习惯在加任务时直接序列化成字符串，两边类型对不上。真正踩坑的人常以为是自己 JSON 写错了，其实只是编码问题。
解决：在 get_redis() 里统一加decode_responses=True，让所有读写都按字符串处理。如果不开这个参数，每次 BRPOP 出来的值都要手动.decode()，轮询一多非常容易漏。
提示：这个参数改动虽然不起眼，但直接影响整个二次开发阶段的数据处理方式，建议一开始就固定下来。
4.3 BRPOP 的 timeout 用 1，队列空闲时进程 CPU 偏高
现象：队列里没任务时，worker 进程的 CPU 占用比预期的活跃，或者日志里不断刷“等待”相关的空转输出。
原因：BRPOP timeout=1 表示阻塞 1 秒后返回 None，代码里 continue 后马上又进入下一轮 BRPOP。表面上看是阻塞，实际是高频循环加短暂阻塞，空转频率明显高于 timeout=0 的写法。
解决：常驻进程把 timeout 改成 0，BRPOP 会一直阻塞到数据到达，等待期间不消耗 CPU。demo 里写 1 仅仅是为了测试时 Ctrl+C 响应快，长驻服务一定记得改掉。
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 ZSET，score 存到期时间戳，值存任务内容。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, start=0, num=10) 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 三条 List，BRPOP 支持一次传多个 key，越靠左优先级越高：
r.brpop("queue:high", "queue:normal", "queue:low", timeout=0)优先级队列在真正的业务场景里非常常用，这段代码已经是在 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 的流程：生产者、消费者、失败兜底，三个环节跑通再加功能。排序、延迟、可靠投递都是在这三个环节上叠加的，基础没立住，加再多都是空中楼阁。希望本次拆解对你有帮助。
本文还有配套的精品资源，点击获取
