在Python生态中，构建异步任务队列最轻量的方案之一就是Redis与RQ的组合。RQ全称Redis Queue，它利用Redis作为消息存储，以极简的API让开发者将函数调用转化为后台任务。与Celery等重量级框架相比，RQ没有复杂的Broker配置，没有繁多的中间件依赖，甚至不需要编写单独的任务文件，只要你的代码里有一个普通函数，就可以把它丢进队列。
Redis与RQ的核心特性与适用场景
Redis本身就是一款高性能的内存键值数据库，它的列表结构天然适合做队列。LPUSH和BRPOP这两个指令配合起来，恰好能实现一个线程安全、支持阻塞读取的消息队列。RQ则在这个基础上封装了任务序列化、worker进程管理、失败重试等逻辑，让开发者无需关心底层协议，直接用Python函数就能定义任务。
适用场景非常明确：需要异步处理但不想引入庞大基础设施的项目。比如发送注册验证邮件、生成缩略图、导出报表、爬虫网页下载等耗时操作。这些任务通常对实时性要求不高，用户只需要知道请求已经接受，后台慢慢执行即可。RQ的轻量化特性特别适合中小型项目，或者作为大型项目中某个子模块的局部异步方案。
相比自己使用Redis列表手写队列，RQ最大的优势在于它内置了任务状态管理和worker并发控制。你可以在一个进程中运行多个worker，也可以把worker分布到多台机器上，只需要它们共享同一个Redis实例。任务执行失败后，RQ会记录异常堆栈并自动放入失败队列，方便后续排查。这些能力大幅度降低了维护成本。
快速上手：安装与最简单的任务入队
安装RQ非常简单，只需要pip安装redis和rq两个包。值得注意的是，RQ依赖Redis服务，因此你还需要在本机或远程环境准备好Redis。在下面的例子中，我们假设Redis已经运行在默认端口6379上。
# 安装
# pip install rq redis
import redis
from rq import Queue
# 连接Redis
redis_conn = redis.Redis(host='localhost', port=6379, db=0)
q = Queue(connection=redis_conn) # 创建默认队列
# 定义一个普通函数作为任务
def send_email(user_id):
print("正在给用户发送邮件:", user_id)
# 模拟耗时操作
import time
time.sleep(2)
return "邮件发送成功"
# 入队：调用q.enqueue即可，无需预先注册任务
job = q.enqueue(send_email, user_id=123)
print(job.id)
在上方代码中，q.enqueue会将函数对象、参数以及执行所需的模块信息序列化后存入Redis。job.id是任务的唯一标识，你可以稍后用这个ID查询执行状态。如果任务函数位于其他模块，RQ会尝试导入该模块，因此建议将任务函数放在可以被worker进程导入的路径中。
要让任务真正执行，需要启动一个worker进程。worker的作用是监听队列，一旦发现新任务就取出并执行。启动方式非常简单，在命令行中执行rq worker命令，并指定队列名。默认情况下，RQ会从当前目录的Python模块中导入任务函数。
rq worker --url redis://localhost:6379/0
worker启动后，你会看到控制台打印出监听中的队列名称。当通过生产者脚本入队任务时，worker会立即接收并运行send_email函数。运行结果可以通过job.result获取，也可以定义回调实现更复杂的处理。如果你想在同一个队列里运行多个worker，只需要打开多个终端执行同样命令即可。
深入理解RQ任务生命周期与worker机制
一个RQ任务大致经历四个阶段：排队、执行、成功或失败、结果保留。任务入队后，先以序列化形式存储在Redis的列表中，状态为queued。worker通过BRPOP命令阻塞弹出任务，此时反序列化函数和参数，开始执行，状态变为started。执行完成后，如果返回正常结果，RQ会将返回值保存到Redis中，状态变为finished。如果抛出异常，状态变为failed，并把异常信息存入失败队列。
这个生命周期设计使得任务状态可以独立于worker进程而存在。即使worker中途崩溃，被取出的任务没有执行完成，Redis中的任务记录仍然保留。不过需要注意的是，默认情况下任务一旦被worker取出就视为开始执行，如果worker崩溃，任务可能丢失。要防止这种情况，可以启用RQ的Sentry模式或使用更高级的可靠性配置，但在简单场景中，这个风险是可控的。
worker内部采用Fork方式创建子进程处理每个任务。这种设计带来了一个好处：即使某个任务引起内存泄漏或段错误，子进程退出后不会影响主进程。但代价是任务函数及其依赖的模块必须在fork之前就已导入，因此RQ规定任务函数必须定义在模块顶层，且模块需要可以被worker导入。你可以在任务中定义局部函数吗？可以，但序列化时可能会出现问题，所以最好将所有任务函数放到独立的tasks.py文件中。
worker支持多并发，默认每个worker进程只能同时执行一个任务。如果你想提高吞吐量，可以使用--workers参数指定worker进程数量，或者配合gevent或eventlet使用协程模式。RQ也提供了--burst模式，执行完当前队列中的全部任务后自动退出，非常适合在定时任务或CI脚本中临时启动worker。
进阶用法：延迟任务、定时任务与优先级调度
除了一般的即时入队，RQ还支持延迟执行任务。通过enqueue_in和enqueue_at方法，可以让任务在指定时间后或指定时间点执行。延迟任务在实现上并不复杂，它利用Redis的有序集合保存任务的时间戳，由一个额外的调度器进程定期检查并将到期任务移入主队列。
from rq import Queue from datetime import timedelta, datetime import redis conn = redis.Redis() q = Queue(connection=conn) # 延迟10秒执行 q.enqueue_in(timedelta(seconds=10), send_email, user_id=456) # 在某个固定时间点执行 scheduled_time = datetime(2025, 5, 20, 10, 30) q.enqueue_at(scheduled_time, send_email, user_id=789)
使用延迟功能需要启动独立的调度进程，命令是rq scheduler。这个进程会周期性地扫描延迟队列，将已经到期的任务重新入队。如果你不需要延迟任务，则可以不启动它，不会影响普通队列的正常工作。
关于优先级，RQ的体系相对简单：每一个队列实例本质上对应一个队列名称，不同名称的队列拥有不同的优先级。你可以创建high、default、low三个队列，worker启动时按顺序监听high、default、low。同一个队列内部不区分优先级，这与Celery的优先级机制有所不同，但已经可以满足多数业务需求。
high_q = Queue('high', connection=conn)
default_q = Queue('default', connection=conn)
low_q = Queue('low', connection=conn)
high_q.enqueue(send_email, user_id=1)
default_q.enqueue(send_email, user_id=2)
low_q.enqueue(send_email, user_id=3)
# 启动worker时指定队列顺序，前面的优先
# rq worker high default low
这种多队列模式也在某种程度上替代了优先级调度。你可以根据业务紧急程度，把不同类型的任务分散到不同的队列，并调整worker对每个队列的监听顺序。如果需要更细粒度的优先级控制，则需要考虑改用Celery或自定义实现。
RQ与Celery的对比：何时选择RQ
Celery是Python中最受欢迎的分布式任务队列，支持RabbitMQ、Redis、Amazon SQS等多种Broker，并且内置定时调度、任务编排、结果后端、限流等丰富功能。然而这些能力也带来了更高的学习成本和运维复杂度。RQ的目标恰恰相反：它只支持Redis作为Broker，功能集中而克制，概念数量很少，代码库也非常小。
对于一个小型博客站的邮件服务，或者一个内部工具的异步报表生成，使用Celery无异于用高射炮打蚊子。RQ的配置只需要几行代码，任务定义就是普通Python函数，几乎没有任何约定。如果项目已经引入了Redis，那么RQ的额外成本几乎为零。相反，如果你的系统未来需要跨语言的消息通信、复杂的工作流编排或大规模分布式调度，Celery或许是更稳妥的选择。
从性能上看，RQ在处理大量短小任务时的吞吐量并不差，因为Redis本身速度很快，任务序列化使用的是pickle，开销也不大。但RQ缺少Celery那样的任务路由和幂等保障，也不提供基于时间或频率的复杂调度表。RQ的定时调度功能相对简陋，仅能指定延迟和固定时间，而Celery的celery beat支持crontab表达式，可以实现复杂的周期任务。
开发者在选型时，可以考虑一个简单原则：如果队列只是你项目中的一个辅助工具，希望十分钟内集成完毕，那么选RQ；如果队列是系统的轴心，需要长期演进和精细控制，那么选Celery。当然，RQ也提供了许多扩展机制，例如自定义job类、自定义worker类，以及通过middleware实现额外的逻辑。这些扩展让RQ在保持简洁的同时不至于完全没有灵活性。
最后需要提醒的是，RQ并不适用于所有异步场景。如果你需要的是跨进程共享状态，而不是任务分发，那么请使用Redis本身的发布订阅或数据结构。RQ的定位非常明确：将函数调用变成后台任务。把握住这一点，你就能在正确的地方使用它。
RedisRQPython简单队列修改时间：2026-08-15 10:37:13