rq 消息队列(python)

以你之姓@ 2023-06-25 07:42 41阅读 0赞

# 0.简介 #

RQ (*Redis Queue*)是一个简单的Python库，用于队列任务并在后台与工人（worker）一起处理它们。它由Redis提供支持，旨在降低入门门槛。它可以轻松集成到您的Web堆栈中。

RQ 要求 Redis >= 3.0.0.

## 开始 ##

首先，运行Redis服务。你可以使用现有的。将任务放在队列中，你不必执行任何特殊操作，只需定义一般冗长或阻塞的函数：

import requests
    
    def count_words_at_url(url):
        resp = requests.get(url)
        return len(resp.text.split())

然后，创建一个RQ队列：

from redis import Redis
    from rq import Queue
    
    q = Queue(connection=Redis())

再将函数调用放入队列中：

from my_module import count_words_at_url
    result = q.enqueue(
                 count_words_at_url, 'http://nvie.com')

## Worker ##

在后台开始执行入队函数，从项目的目录中启动一个worker：

$ rq worker
    *** Listening for work on default
    Got count_words_at_url('http://nvie.com') from default
    Job result = 818
    *** Listening for work on default

就是这样。

## 安装 ##

只需使用以下命令安装最新发布的版本：

pip install rq

如果你想要最新的版本（可能会破坏），请使用：

pip install -e git+git@github.com:nvie/rq.git@master#egg=rq

## 项目历史 ##

本项目的灵感来自[Celery][]，[Resque][]和[这个代码段][Link 1]的优点，作为现有队列框架的轻量级替代品，它具有较低的入门门槛。

--------------------

RQ由[Vincent Driessen][]撰写

在[BSD][]许可条款下开源

# 1.队列 #

任务是一个Python对象，表示在worker（后台）进程中异步调用的函数。只需将对函数及其参数的引用推送到队列中，就可以异步调用任何Python函数。这被称为入队。

## 入队 ##

将任务放在队列中，首先声明一个函数：

import requests
    
    def count_words_at_url(url):
        resp = requests.get(url)
        return len(resp.text.split())

注意到什么了？这个函数没什么特别的！任何Python函数调用都可以放在RQ队列中。

将该函数放入队列中，只需执行以下操作：

from rq import Queue
    from redis import Redis
    from somewhere import count_words_at_url
    import time
    
    # 生命Redis连接
    redis_conn = Redis()
    q = Queue(connection=redis_conn)  # 没有参数意味着使用默认队列
    
    # 延迟执行count_words_at_url('http://nvie.com')
    job = q.enqueue(count_words_at_url, 'http://nvie.com')
    print(job.result)   # => None
    
    # 稍等一会儿，直到worker完成
    time.sleep(2)
    print(job.result)   # => 889

如果想要将任务放在特定队列中，只需指定其名称：

q = Queue('low', connection=redis_conn)
    q.enqueue(count_words_at_url, 'http://nvie.com')

注意到上面例子中的`Queue('low')`了吗？你可以使用任何队列名称，因此你可以根据自己的意愿灵活地分配任务。常见的命名模式是使用优先级命名队列（例如：`high`，`medium`，`low`）

此外，你可以添加一些选项来修改队列任务的行为。默认情况下，这些可以从传递给任务函数的kwargs中获取。

*  `job_timeout`指定任务中断并标记为`failed`之前的最大运行时间。默认单位为秒，它可以是整数或表示整数的字符串（例如： `2`，`'2'`）。此外，它可以是具有指定单位，包括小时，分钟，秒的字符串（例如`'1h'`，`'3m'`，`'5s'`）。
 *  `result_ttl`指定成功任务及其结果的保留时间（以秒为单位）。过期的任务将被自动删除。默认为500秒。
 *  `ttl`指定任务被丢弃之前的最大排队时间。如果指定的值为`-1`表示无限任务ttl，它将无限期运行。默认为-1。
 *  `failure_ttl` 指定失败任务的保留时间（默认为1年）
 *  `depends_on` 指定在此任务排队之前必须完成的另一个任务（或任务ID）。
 *  `job_id` 允许手动指定此任务的 `job_id`
 *  `at_front`将任务放在队列的*前面*，而不是后面
 *  `description` 为排队任务添加其他说明。
 *  `args`和`kwargs`：使用这些来显式地将参数和关键字参数传递给基础任务函数。如果你的函数恰好与RQ具有冲突的参数名称，这很有用，例如`description`或`ttl`。

在最后一种情况下，如果你想传递`description`和`ttl`关键字参数到你的任务而不是RQ的enqueue函数，这就是你要做的：

q = Queue('low', connection=redis_conn)
    q.enqueue(count_words_at_url,
              ttl=30,  # ttl将被RQ使用
              args=('http://nvie.com',),
              kwargs={ 
                  'description': 'Function description', # 将被传递给count_words_at_url
                  'ttl': 15  # 将被传递给count_words_at_url
              })

对于Web进程无法访问在worker中运行的源代码的情况（即代码库X从代码库Y调用延迟函数），你也可以将该函数作为字符串引用传递。

q = Queue('low', connection=redis_conn)
    q.enqueue('my_package.my_module.my_func', 3, 4)

## 使用队列 ##

除了排队任务外，Queues还有一些有用的方法：

from rq import Queue
    from redis import Redis
    
    redis_conn = Redis()
    q = Queue(connection=redis_conn)
    
    # 获得队列的任务数
    print(len(q))
    
    # 检索任务
    queued_job_ids = q.job_ids # 从队列中获取任务ID列表
    queued_jobs = q.jobs # 从队列中获取任务实例列表
    job = q.fetch_job('my_id') # 返回任务ID为“my_id”的任务实例
    
    # 清空队列, 这将删除队列中的所有任务
    q.empty()
    
    # 删除队列
    q.delete(delete_jobs=True) # 传递`True`将删除队列中的所有任务
    # 队列现在无法使用。它可以通过将任务放入队列中来重新创建。

### 设计 ###

使用RQ，你不必预先设置任何队列，也不必指定任何通道，交换，路由规则或诸如此类的东西。你可以将任务放到任何你想要的队列中。只要将任务排队到尚不存在的队列，就会立即创建它。

RQ并没有采用先进的中间商（broker）能够做消息路由给你。你可能会认为这是一个很棒的优势或障碍，取决于你正在解决的问题。

最后，它不是便携式协议，因为它依赖于[pickle][] 来序列化任务，因此它是一个仅限Python使用的系统。

## 延迟结果 ##

当任务入队时，`queue.enqueue()`方法返回一个`Job`实例。这只不过是一个可用于检查实际任务结果的代理对象。

为此，它具有一个方便的`result`访问者属性，它将在任务尚未完成时返回`None`，或者在任务完成时返回非`None`值（假设任务具有返回值）。

## `@job`装饰器 ##

如果你熟悉Celery，你可能会习惯它的`@task`装饰器。从RQ> = 0.3开始，引入类似的装饰器：

from rq.decorators import job
    
    @job('low', connection=my_redis_conn, timeout=5)
    def add(x, y):
        return x + y
    
    job = add.delay(3, 4)
    time.sleep(1)
    print(job.result)

## 绕过worker ##

出于测试目的，你可以将任务放入队列，而无需将实际执行委派给worker（从版本0.3.1开始提供）。为此，将`is_async=False`参数传递给Queue构造函数：

>>> q = Queue('low', is_async=False, connection=my_redis_conn)
    >>> job = q.enqueue(fib, 8)
    >>> job.result
    21

上面的代码在没有活动工作程序的情况下运行，`fib(8)` 在同一进程中同步执行。你可能知道Celery的这种行为`ALWAYS_EAGER`。但请注意，你仍需要与redis实例建立工作连接，以存储与任务执行和完成相关的状态。

## 任务依赖 ##

RQ 0.4.0中的新功能是链接多个任务执行。执行依赖于其他任务的任务，使用以下`depends_on`参数：

q = Queue('low', connection=my_redis_conn)
    report_job = q.enqueue(generate_report)
    q.enqueue(send_report, depends_on=report_job)

处理任务依赖关系的能力允许你将大型任务拆分为几个较小的任务。依赖于另一个的任务仅在其依赖关系完成时才会排队。

## 任务的注意事项 ##

从技术上讲，你可以将任何Python函数调用放在队列中，但这并不意味着这样做总是明智的。在将任务放入队列之前需要考虑的一些事项：

*  确保该函数`__module__`可由worker导入。特别是，这意味着您无法将`__main__`模块中声明的函数排入队列。
 *  确保worker和任务生成器共享完全相同的源代码。
 *  确保函数调用不依赖于其上下文。特别是，全局变量是邪恶的（一如既往），但是当工作人员处理它时，函数所依赖的任何状态（例如“当前”用户或“当前”Web请求）都不存在。如果要为“当前”用户完成工作，则应将该用户解析为具体实例，并将对该用户对象的引用作为参数传递给任务。

## 限制 ##

RQ workers只能在实现`fork()`的系统上运行。最值得注意的是，这意味着如果不使用[Windows的Linux子系统][Windows_Linux]并在bash shell中运行，就无法在Windows上运行worker。

# 2.Workers #

worker是一个Python进程，通常在后台运行，并且仅作为子进程（work horse）存在，以执行你不希望在Web进程内执行的冗长或阻塞任务。

## 启动workers ##

开始计算工作，只需从项目目录的根目录启动一个worker：

$ rq worker high default low
    *** Listening for work on high, default, low
    Got send_newsletter('me@nvie.com') from default
    Job ended normally without result
    *** Listening for work on high, default, low
    ...

Workers将在无限循环中读取给定队列中的任务（顺序很重要），当所有任务完成后等待新工作到达。

每个worker一次只能处理一份工作。在一个worker中，没有并发处理。如果你想同时执行任务，只需启动更多workers。

你可以使用[Supervisor][]或 [systemd][]等流程管理器在生产中运行RQ worker。

### Burst模式 ###

默认情况下，workers将立即开始工作，在他们完成所有任务时阻止并等待新任务。workers也可以以\*\*\*突发模式\*\*\*启动以完成所有当前可用的任务，在所有给定队列为空后立即退出。

$ rq worker --burst high default low
    *** Listening for work on high, default, low
    Got send_newsletter('me@nvie.com') from default
    Job ended normally without result
    No more work, burst finished.
    Registering death.

这对于需要定期处理的批处理任务非常有用，或者只是在高峰期临时扩展workers。

### worker参数 ###

除此`--burst`之外，`rq worker`还接受以下参数：

*  `--url`或`-u`：描述Redis连接详细信息的URL（例如`rq worker --url redis://:secrets@example.com:1234/9`）
 *  `--path`或`-P`：支持多个导入路径（例如`rq worker --path foo --path bar`）
 *  `--config`或`-c`：包含RQ设置的模块的路径。
 *  `--results-ttl`：任务结果将保留此秒数（默认为500）。
 *  `--worker-class`或`-w`：使用的RQ Worker类（例如`rq worker --worker-class 'foo.bar.MyWorker'`）
 *  `--job-class`或`-j`：使用的RQ Job类。
 *  `--queue-class`：使用的RQ队列类。
 *  `--connection-class`：使用的Redis连接类，默认为`redis.StrictRedis`。
 *  `--log-format`：工作日志的格式，默认为 `'%(asctime)s %(message)s'`
 *  `--date-format`：工作日志的日期时间格式，默认为 `'%H:%M:%S'`
 *  `--disable-job-desc-logging`：关闭任务说明日志记录。
 *  `--max-jobs`：执行的最大任务数。

## worker内部 ##

### worker生命周期 ###

worker的生命周期包括几个阶段：

1.  **启动**：加载Python环境。
2.  **注册**：worker将自己注册到系统，以便知道该worker。
3.  **监听**：从任何给定的Redis队列中获取任务。如果所有队列都为空且worker以突发模式运行，则立即退出。否则，等待任务到来。
4.  **准备执行任务**：worker告诉系统它将通过设置其状态为`busy`，并在`StartedJobRegistry`中注册任务来开始工作。
5.  **fork子进程**：复制一个子进程（work horse）以在故障安全上下文中执行实际任务。
6.  **处理工作**：在子进程（work horse）中执行实际的工作。
7.  **清理任务执行**。worker将其状态设置为`idle`，并设置任务及其结果的过期时间（基于`result_ttl`）。任务成功后，会从`StartedJobRegistry`中删除任务并且添加到`FailedJobRegistry`中，任务失败后，添加到`FailedJobRegistry`中。
8.  **循环**：从第3步开始重复。

### 性能说明 ###

基本上`rq worker`shell脚本是一个简单的fetch-fork-execute循环。当你的很多工作都进行冗长的设置，或者它们都依赖于同一组模块时，每次运行任务时都要支付开销（因为你在fork之后进行了导入）。这很干净，因为RQ不会以这种方式泄漏内存，但也会很慢。

可用于提高这类任务的吞吐量性能的模式可以是在fork之前导入必要的模块。无法告诉RQ worker为你执行此设置，但你可以在开始工作循环之前自己完成此操作。

为此，请提供你自己的任务脚本（而不是使用`rq worker`）。一个简单的实现示例：

#!/usr/bin/env python
    import sys
    from rq import Connection, Worker
    
    # 预先导入模块
    import library_that_you_want_preloaded
    
    # 提供队列名称，类似于rq worker
    with Connection():
        qs = sys.argv[1:] or ['default']
    
        w = Worker(qs)
        w.work()

### worker名称 ###

worker以其名称注册到系统，这些名称在实例化期间随机生成（请参阅[监控][Link 2]）。要覆盖此默认值，请在启动worker时指定名称，或使用`--name`客户端选项。

from redis import Redis
    from rq import Queue, Worker
    
    redis = Redis()
    queue = Queue('queue_name')
    
    # Start a worker with a custom name
    worker = Worker([queue], connection=redis, name='foo')

### 检索worker信息 ###

*在版本0.10.0中更新。*

`Worker`实例将其运行时信息存储在Redis中。以下是检索它们的方法：

from redis import Redis
    from rq import Queue, Worker
    
    # Returns all workers registered in this connection
    redis = Redis()
    workers = Worker.all(connection=redis)
    
    # Returns all workers in this queue (new in version 0.10.0)
    queue = Queue('queue_name')
    workers = Worker.all(queue=queue)
    worker = workers[0]
    print(worker.name)

除`worker.name`之外，worker人还具有以下属性：

*  `hostname` \- 运行此工作程序的主机
 *  `pid` \- worker的进程ID
 *  `queues` \- worker正在侦听工作的队列
 *  `state`\-可能的状态是`suspended`，`started`，`busy`和`idle`
 *  `current_job` \- 正在执行的任务（如果有的话）
 *  `last_heartbeat` \- worker最后的更新时间
 *  `birth_date` \- worker实例化的时间
 *  `successful_job_count` \- 成功完成的任务数量
 *  `failed_job_count` \- 处理的失败任务数
 *  `total_working_time` \- 执行任务所花费的时间，以秒为单位

*版本0.10.0中的新功能。*

如果您只想知道用于监控目的的worker数量， `Worker.count()`那么性能要高得多。

from redis import Redis
    from rq import Worker
    
    redis = Redis()
    
    # 计算redis中的worker连接数
    workers = Worker.count(connection=redis)
    
    # 计算指定队列中的worker数
    queue = Queue('queue_name', connection=redis)
    workers = Worker.all(queue=queue)

### worker统计 ###

*0.9.0版中的新功能。*

如果要检查队列的利用率，`Worker`实例会存储一些有用的信息：

from rq.worker import Worker
    worker = Worker.find_by_key('rq:worker:name')
    
    worker.successful_job_count  # 成功的任务书
    worker.failed_job_count # 失败的任务书
    worker.total_working_time  # 执行任务所花费的时间 (单位：秒)

## 更好的worker处理标题 ##

安装第三方软件包`setproctitle`后，工作进程将具有更好的标题（由ps和top等系统工具显示）：

pip install setproctitle

## 取下worker ##

如果工作人员在任何时候接收到`SIGINT`（通过Ctrl + C）或`SIGTERM`（通过 `kill`），worker等待当前正在运行的任务完成，停止工作循环并优雅地注册自己的死亡。

如果，在删除阶段，再次收到`SIGINT`或`SIGTERM`，worker将强行终止子进程（发送它`SIGKILL`），但仍将尝试注册自己的死亡。

## 使用配置文件 ##

如果您想通过配置文件而不是通过命令行参数进行配置`rq worker`，可以通过创建Python文件来完成此操作 `settings.py`：

REDIS_URL = 'redis://localhost:6379/1'
    
    # 你也可以指定使用的redis数据库
    # REDIS_HOST = 'redis.example.com'
    # REDIS_PORT = 6380
    # REDIS_DB = 3
    # REDIS_PASSWORD = 'very secret'
    
    # 监听的队列
    QUEUES = ['high', 'default', 'low']
    
    # 如果您使用Sentry收集运行时异常，则可以使用此方法去配置RQ
    # The 'sync+' prefix is required for raven: https://github.com/nvie/rq/issues/350#issuecomment-43592410
    SENTRY_DSN = 'sync+http://public:secret@example.com/1'
    
    # 自定义worker名称
    # NAME = 'worker-1024'

上面的示例展示了当前支持的所有选项。

*注意：* `QUEUES` *和* `REDIS_PASSWORD` *设置是0.3.3以后的新特性。*

指定从哪个模块读取设置，请使用以下`-c`选项：

$ rq worker -c settings

## 自定义Worker类 ##

有时您想要自定义worker的行为。到目前为止，一些更常见的请求是：

1.  在运行任务之前管理数据库连接。
2.  使用不需要的任务执行模型`os.fork`。
3.  能够使用不同的并发模型，如 `multiprocessing`或`gevent`。

您可以使用该`-w`选项指定要使用的其他工作类：

$ rq worker -w 'path.to.GeventWorker'

## 自定义任务和队列类 ##

你可以告诉worker使用自定义类来处理任务和队列`--job-class`或`--queue-class`。

$ rq worker --job-class 'custom.JobClass' --queue-class 'custom.QueueClass'

在添加队列任务时不要忘记使用相同的类。

例如：

from rq import Queue
    from rq.job import Job
    
    class CustomJob(Job):
        pass
    
    class CustomQueue(Queue):
        job_class = CustomJob
    
    queue = CustomQueue('default', connection=redis_conn)
    queue.enqueue(some_func)

## 自定义DeathPenalty类 ##

当任务超时时，worker将尝试使用提供的方法`death_penalty_class`（默认值: `UnixSignalDeathPenalty`）将其终止。如果你希望尝试以特定于应用程序或“更清洁”的方式杀死任务，则可以覆盖此项。

DeathPenalty类使用以下参数构造 `BaseDeathPenalty(timeout, JobTimeoutException, job_id=job.id)`

## 自定义异常处理 ##

如果你需要针对不同类型的任务以不同方式处理错误，或者只是想要自定义RQ的默认错误处理行为，使用`--exception-handler`选项运行`rq worker`：

$ rq worker --exception-handler 'path.to.my.ErrorHandler'
    
    # Multiple exception handlers is also supported
    $ rq worker --exception-handler 'path.to.my.ErrorHandler' --exception-handler 'another.ErrorHandler'

如果要禁用RQ的默认异常处理程序，请使用`--disable-default-exception-handler`选项：

$  rq worker --exception-handler 'path.to.my.ErrorHandler' --disable-default-exception-handler

# 3.结果 #

排队任务是延迟执行函数。这意味着我们正在解决一个问题，但正在获得结果。

## 处理结果 ##

Python函数可能有返回值，因此任务也可以拥有它们。如果任务返回非`None`返回值，则worker会将该返回值写入任务的`result`键下的Redis哈希值。任务完成后，任务的Redis哈希本身将在500秒后失效。

排队任务的一方`Job`因排队本身而返回一个实例。这样的`Job`对象是与任务ID相关联的代理对象，以便能够轮询结果。

**在返回值的TTL** 返回值被写回Redis有限的生命周期（通过Redis到期密钥），这只是为了避免不断增长的Redis数据库。

从RQ> = 0.3.1，可以使用`result_ttl`关键字参数传递给`enqueue()`和`enqueue_call()`指定任务结果的TTL值 。它也可以用来完全禁用到期时间。然后，你自己负责清理工作，所以要小心使用它。

你可以执行以下操作：

q.enqueue(foo)  # 结果在500秒后过期 (默认)
    q.enqueue(foo, result_ttl=86400)  #结果在1天后过期
    q.enqueue(foo, result_ttl=0)  # 立即删除
    q.enqueue(foo, result_ttl=-1)  # 不过期--需要手动删除

此外，你可以使用它来保留没有返回值的已完成任务，默认情况下会立即删除这些值。

q.enqueue(func_without_rv, result_ttl=500)  # 任务明确保存

## 处理异常 ##

任务失败并抛出异常。这是生活的事实。RQ以下列方式处理此问题。

此外，应该可以重试失败的任务。通常，这是需要手动解释的事情，因为没有自动或可靠的方式让RQ判断某些任务是否可以安全地重试。

当在任务中抛出异常时，它会被worker捕获，序列化并存储在任务的Redis哈希的`exc_info`键下。对工作的引用放在`FailedJobRegistry`。默认情况下，失败的任务将保留1年。

任务本身具有一些有用的属性，可用于辅助检查：

*  任务的原始创作时间
 *  最后的入队日期
 *  原始队列
 *  所需函数调用的文本描述
 *  异常信息

这使得可以手动检查和解释问题，并可能重新提交作任务

## 处理中断 ##

当worker以礼貌的方式被杀（Ctrl + C或`kill`）时，RQ努力不丢失任何工作。当前的工作完成后，工人将停止进一步处理工作。这确保了任务总能获得公平机会完成自己。

但是，工人可以被强行杀死`kill -9`，这不会让worker有机会优雅地完成工作或将工作放在`failed` 队列中。因此，强行杀死一名worker可能会导致损害。

## 处理任务超时 ##

默认情况下，任务应在180秒内执行。之后，worker杀死子进程（work horse）并将任务放入`failed`队列，表明任务超时。

如果任务需要更多（或更少）时间来完成，则可以通过将其指定为`enqueue()`调用的关键字参数来放宽（或收紧）默认超时时间，如下所示：

q = Queue()
    q.enqueue(mytask, args=(foo,), kwargs={ 'bar': qux}, job_timeout=600)  # 10 分钟

你还可以一次更改，通过特定队列实例排队任务的默认超时时间，这对于这样的模式非常有用：

# high队列有8秒超时时间，low队列有10秒
    high = Queue('high', default_timeout=8)  # 8 secs
    low = Queue('low', default_timeout=600)  # 10 mins
    
    # 单个任务仍可以覆盖这些默认值
    low.enqueue(really_really_slow, job_timeout=3600)  # 1 hr

单个任务仍然可以指定替代超时时间，因为worker会遵守这些。

# 4.任务 #

对于某些用例，可以从任务函数本身访问当前任务ID或实例。或者在任务中存储任意数据。

## 从Redis检索任务 ##

所有任务信息都存储在Redis中。您可以使用`Job.fetch()`检查任务及其属性。

from redis import Redis
    from rq.job import Job
    
    redis = Redis()
    job = Job.fetch('my_job_id', connection=redis)
    print('Status: %s' $ job.get_status())

一些有趣的任务属性包括：

*  `job.get_status()`
 *  `job.func_name`
 *  `job.args`
 *  `job.kwargs`
 *  `job.result`
 *  `job.enqueued_at`
 *  `job.started_at`
 *  `job.ended_at`
 *  `job.exc_info`

*版本1.1.0中的新功能。*

如果要高效获取大量任务，请使用`Job.fetch_many()`。

jobs = Job.fetch_many(['foo_id', 'bar_id'], connection=redis)
    for job in jobs:
        print('Job %s: %s' % (job.id, job.func_name))

## 访问“当前”任务 ##

由于任务函数是常规Python函数，因此你必须向RQ询问当前任务ID（如果有）。为此，你可以使用：

from rq import get_current_job
    
    def add(x, y):
        job = get_current_job()
        print('Current job: %s' % (job.id,))
        return x + y

## 在任务上存储任意数据 ##

*版本0.8.0中的新功能。*

在任务上添加/更新自定义状态信息，你可以访问该`meta`属性，该属性允许你在任务本身上存储任意可选择的数据：

import socket
    
    def add(x, y):
        job = get_current_job()
        job.meta['handled_by'] = socket.gethostname()
        job.save_meta()
    
        # do more work
        time.sleep(1)
        return x + y

## 在队列中工作的时间 ##

*版本0.4.7中的新功能。*

一个任务有两个TTL，一个用于任务结果，一个用于任务本身。这意味着如果你有一段时间后不应该执行的工作，你可以这样定义一个TTL：

#创建任务时:
    job = Job.create(func=say_hello, ttl=43)
    
    # 或者排队时:
    job = q.enqueue(count_words_at_url, 'http://nvie.com', ttl=43)

## 失败任务 ##

如果任务在执行期间失败，则worker将把任务放在FailedJobRegistry中。在Job实例上，`is_failed`属性将为true。FailedJobRegistry可以通过访问`queue.failed_job_registry`。

from redis import StrictRedis
    from rq import Queue
    from rq.job import Job
    
    
    def div_by_zero(x):
        return x / 0
    
    
    connection = StrictRedis()
    queue = Queue(connection=connection)
    job = queue.enqueue(div_by_zero, 1)
    registry = queue.failed_job_registry
    
    worker = Worker([queue])
    worker.work(burst=True)
    
    assert len(registry) == 1  # 失败任务被保存在FailedJobRegistry
    
    registry.requeue(job)  # 重新入队
    
    assert len(registry) == 0
    
    assert queue.count == 1

默认情况下，失败的任务将保留1年。你可以排队时通过指定`failure_ttl`（单位：秒）来更改此设置 。

job = queue.enqueue(foo_job, failure_ttl=300)  # 5 minutes in seconds

## 重新排队失败的任务 ##

RQ还提供了一个客户端工具，可以轻松地重新排队失败的任务。

# 将重新排队foo_job_id和bar_job_id
    rq requeue --queue myqueue -u redis://localhost:6379 foo_job_id bar_job_id
    
    # 将重新排队所有任务
    rq requeue --queue myqueue -u redis://localhost:6379 --all

# 5.监控 #

监控是RQ闪耀的地方。

最简单的方法可能是使用[RQ仪表板][RQ]，这是一个单独分发的工具，它是RQ的基于Web的轻量级监视器前端，如下所示：

[\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-1eykKf3X-1577369744997)(https://luolingchun.github.io/rq-docs-cn/chapter/assets/dashboard.png)\]][img-1eykKf3X-1577369744997_https_luolingchun.github.io_rq-docs-cn_chapter_assets_dashboard.png]

安装，只需执行：

$ pip install rq-dashboard
    $ rq-dashboard

它也可以轻松集成到Flask应用程序中。

## 在控制台监控 ##

查看存在哪些队列以及哪些worker处于活动状态，只需键入`rq info`：

$ rq info
    high       |██████████████████████████ 20
    low        |██████████████ 12
    default    |█████████ 8
    3 queues, 45 jobs total
    
    Bricktop.19233 idle: low
    Bricktop.19232 idle: high, default, low
    Bricktop.18349 idle: default
    3 workers, 3 queues

## 通过队列名称查询 ##

如果你正在寻找特定的队列，你还可以查询队列的子集：

$ rq info high default
    high       |██████████████████████████ 20
    default    |█████████ 8
    2 queues, 28 jobs total
    
    Bricktop.19232 idle: high, default
    Bricktop.18349 idle: default
    2 workers, 2 queues

## 按队列组织worker ##

默认情况下，`rq info`打印当前处于活动状态的worker以及它们正在侦听的队列，如下所示：

$ rq info
    ...
    
    Mickey.26421 idle: high, default
    Bricktop.25458 busy: high, default, low
    Turkish.25812 busy: high, default
    3 workers, 3 queues

查看相同的数据，但按队列组织，使用`-R`（或`--by-queue`）标志：

$ rq info -R
    ...
    
    high:    Bricktop.25458 (busy), Mickey.26421 (idle), Turkish.25812 (busy)
    low:     Bricktop.25458 (busy)
    default: Bricktop.25458 (busy), Mickey.26421 (idle), Turkish.25812 (busy)
    failed:  –
    3 workers, 4 queues

## 区间轮询 ##

默认情况下，`rq info`将打印统计信息并退出。你可以使用`--interval`标志指定轮询间隔。

$ rq info --interval 1

`rq info`将每秒更新一次屏幕。你可以指定一个浮点值来指示秒的分数。当然，请注意，低间隔值会增加Redis的负载。

$ rq info --interval 0.5

# 6.连接 #

为方便起见，尽管RQ使用了 `use_connection()`命令，但由于它污染了全局命名空间，因此不推荐使用它。相反，更喜欢使用`with Connection(...):`上下文管理器进行显式连接管理，或者直接将Redis连接引用传递给队列。

## 单Redis连接（简单） ##

[![img][]][img 1]\*\*注意：\*\*使用`use_connection`过时了。请不要在脚本中使用`use_connection`。

在开发模式下，连接到默认的本地Redis服务器：

from rq import use_connection
    use_connection()

在生产环境中，连接到特定的Redis服务器：

from redis import Redis
    from rq import use_connection
    
    redis = Redis('my.host.org', 6789, password='secret')
    use_connection(redis)

请注意`use_connection`污染全局命名空间的事实。它还意味着你只能使用单个连接。

## 多个Redis连接 ##

无论如何，单一连接模式仅适用于连接到单个Redis实例的情况，以及影响全局上下文的情况（通过将现有连接替换为`use_connection()`调用）。只有在完全控制Web堆栈时才能使用此模式。

在任何其他情况下，或者当你想要使用多个连接时，你应该使用`Connection`上下文或明确地传递连接。

### 显式连接（精确，但是乏味） ###

每个RQ对象实例（队列，worker，任务）都有一个`connection`可以传递给构造函数的关键字参数。使用它，你不需要使用`use_connection()`。相反，你可以像这样创建队列：

from rq import Queue
    from redis import Redis
    
    conn1 = Redis('localhost', 6379)
    conn2 = Redis('remote.host.org', 9836)
    
    q1 = Queue('foo', connection=conn1)
    q2 = Queue('bar', connection=conn2)

队列中排队的每个任务都将知道它所属的连接。worker也一样。

这种方法非常精确，但相当冗长，因此很乏味。

### 连接上下文（精确、简洁） ###

如果要使用多个连接，有一种更好的方法。每个RQ对象实例在创建时将使用RQ连接堆栈上最顶层的Redis连接，这是一种临时替换要使用的默认连接的机制。

一个例子将有助于理解它：

from rq import Queue, Connection
    from redis import Redis
    
    with Connection(Redis('localhost', 6379)):
        q1 = Queue('foo')
        with Connection(Redis('remote.host.org', 9836)):
            q2 = Queue('bar')
        q3 = Queue('qux')
    
    assert q1.connection != q2.connection
    assert q2.connection != q3.connection
    assert q1.connection == q3.connection

你可以将此视为在`Connection`上下文中，每个新创建的RQ对象实例都将`connection`隐式设置参数。将某个任务`q2`排入队列会将其排入第二个（远程）Redis后端，即使在连接上下文之外也是如此。

### 推送/弹出连接 ###

如果你的代码不允许你使用`with`语句，例如，如果您想使用它来设置单元测试，则可以使用`push_connection()`和 `pop_connection()`方法而不是使用上下文管理器。

import unittest
    from rq import Queue
    from rq import push_connection, pop_connection
    
    class MyTest(unittest.TestCase):
        def setUp(self):
            push_connection(Redis())
    
        def tearDown(self):
            pop_connection()
    
        def test_foo(self):
            """Any queues created here use local Redis."""
            q = Queue()

### 哨兵支持 ###

使用redis sentinel，必须在配置文件中指定字典。将此设置与带有自动重启选项的systemd或docker容器结合使用，允许worker和RQ具有与redis的容错连接。

SENTINEL: { 'INSTANCES':[('remote.host1.org', 26379), ('remote.host2.org', 26379), ('remote.host3.org', 26379)],
               'SOCKET_TIMEOUT': None,
               'PASSWORD': 'secret',
               'DB': 2,
               'MASTER_NAME': 'master'}

# 7.异常 #

由于发生异常，任务可能会失败。当你的RQ worker在后台运行时，你如何得到这些异常的通知？

## 默认： `FailedJobRegistry` ##

RQ的默认安全网是`FailedJobRegistry`。每个未成功执行的任务都会存储在此处，以及其异常信息（类型，值，回溯）。虽然这可以确保没有失败的任务“迷路”，但是主动得到关于任务失败的通知没有的。

## 自定义异常处理 ##

RQ支持注册自定义异常处理程序。这样就可以将自己的错误处理逻辑注入到worker身上。

这是你将自定义异常处理程序注册到RQ worker的方法：

from exception_handlers import foo_handler, bar_handler
    
    w = Worker([q], exception_handlers=[foo_handler, bar_handler])

句柄本身是一个函数，采用下列参数：`job`， `exc_type`，`exc_value`和`traceback`：

def my_handler(job, exc_type, exc_value, traceback):
        # 自定义事情
        # 例如, 写入异常到数据库中

你可能还会看到三个异常参数编码为：

def my_handler(job, *exc_info):
        # do custom things here
    from exception_handlers import foo_handler
    
    w = Worker([q], exception_handlers=[foo_handler],
               disable_default_exception_handler=True)

## 链接异常处理 ##

处理程序本身负责决定是否完成异常处理，或者应该落到堆栈上的下一个处理程序。处理程序可以通过返回布尔值来指示这一点。`False`表示停止处理异常，`True`表示继续并进入堆栈的下一个异常处理程序。

重要的是要知道实现者，默认情况下，当处理程序没有显式返回值（`None`）时，这将被解释为`True`（即继续使用下一个处理程序）。

防止处理程序链中的下一个异常处理程序执行，请使用不会丢失的自定义异常处理程序，例如：

def black_hole(job, *exc_info):
        return False

# 8.测试 #

## worker单元测试 ##

你可能希望在单元测试中包含RQ任务。然而，许多框架（例如Django）使用内存数据库，这些数据库与RQ的`fork()`默认行为不能很好地兼容。

因此，你必须使用SimpleWorker类来避免fork（）;

from redis import Redis
    from rq import SimpleWorker, Queue
    
    queue = Queue(connection=Redis())
    queue.enqueue(my_long_running_job)
    worker = SimpleWorker([queue], connection=queue.connection)
    worker.work(burst=True)  # Runs enqueued job
    # Check for result...

## 任务单元测试 ##

另一个用于测试目的的解决方案是使用`is_async=False`queue参数，该参数指示它在同一个线程中立即执行任务，而不是将其分派给worker。不再需要worker了。另外，我们可以使用fakeredis来模拟redis实例，因此我们不必单独运行redis服务器。伪redis服务器的实例可以直接作为连接参数传递给队列：

from fakeredis import FakeStrictRedis
    from rq import Queue
    
    queue = Queue(is_async=False, connection=FakeStrictRedis())
    job = queue.enqueue(my_long_running_job)
    assert job.is_finished
    回布尔值来指示这一点。`False`表示停止处理异常，`True`表示继续并进入堆栈的下一个异常处理程序。
    
    重要的是要知道实现者，默认情况下，当处理程序没有显式返回值（`None`）时，这将被解释为`True`（即继续使用下一个处理程序）。
    
    防止处理程序链中的下一个异常处理程序执行，请使用不会丢失的自定义异常处理程序，例如：
    
    ```python
    def black_hole(job, *exc_info):
        return False

# 8.测试 #

## worker单元测试 ##

你可能希望在单元测试中包含RQ任务。然而，许多框架（例如Django）使用内存数据库，这些数据库与RQ的`fork()`默认行为不能很好地兼容。

因此，你必须使用SimpleWorker类来避免fork（）;

## 任务单元测试 ##

[Celery]: http://www.celeryproject.org/
[Resque]: https://github.com/defunkt/resque
[Link 1]: http://flask.pocoo.org/snippets/73/
[Vincent Driessen]: http://nvie.com/about
[BSD]: https://raw.github.com/nvie/rq/master/LICENSE
[pickle]: http://docs.python.org/library/pickle.html
[Windows_Linux]: https://docs.microsoft.com/en-us/windows/wsl/about
[Supervisor]: https://python-rq.org/patterns/supervisor/
[systemd]: https://python-rq.org/patterns/systemd/
[Link 2]: https://python-rq.org/docs/monitoring/
[RQ]: https://github.com/nvie/rq-dashboard
[img-1eykKf3X-1577369744997_https_luolingchun.github.io_rq-docs-cn_chapter_assets_dashboard.png]: https://luolingchun.github.io/rq-docs-cn/chapter/assets/dashboard.png
[img]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9sdW9saW5nY2h1bi5naXRodWIuaW8vcnEtZG9jcy1jbi9jaGFwdGVyL2Fzc2V0cy93YXJuaW5nLnBuZw?x-oss-process=image/format,png
[img 1]: https://luolingchun.github.io/rq-docs-cn/chapter/assets/warning.png