在 Heroku 上的 RQ 中处理异常处理和重新排队

Question

我有一个用 Python 在 Heroku 上运行的网站，我有一个工作人员作为后台进程来处理我不想阻止网页传递的任务，因此不适合 web dynos。为此，我使用 rq 和 redis 设置了一个队列。

在我的过程中，有时可能会出现自定义异常。对于其中的特定子集，我不想让作业直接进入“失败”队列，而是要重新排队几次。我一直在看rq 主页上的exception handlers page，但有些事情我不清楚。具体来说，它描述了以下编写异常处理程序的方法：

def my_handler(job, exc_type, exc_value, traceback):
    # do custom things here
    # for example, write the exception info to a DB
    ...

现在，我打算按照以下方式做一些事情：

   from rq import requeue_job
   def my_handler(job, exc_type, exc_value, traceback):
        if exec_type == "MyCustomError":
           job.meta['MyErrorCount'] += 1
           job.save()

           if job.meta['MyErrorCount'] >= 10:
               return True
           else:
               requeue_job(job.id)
               return False

问题：

• exc_type、exc_value 和 traceback 是哪些类型的对象？ （例如，if exec_type == "MyCustomError" 行完全正确吗？）
• 我的错误处理程序是否会有效地检测它是否是一个特定的错误，将这些作业重新排队，直到它失败 10 次，然后让它落到failed？是否还会让所有其他错误归于failed？

Answer 1

Here’s my solution

queues = []

def retry_handler(job, exc_type, exc_value, traceback):
    # Returning True moves the job to the failed queue (or continue to
    # the next handler)

    job.meta.setdefault('failures', 1)
    job.meta['failures'] += 1
    if job.meta['failures'] > 3 or isinstance(exc_type, (LookupError, CorruptImageError)):
        job.save()
        return True

    job.status = Status.QUEUED
    for queue_ in queues:
        if queue_.name == job.origin:
            queue_.enqueue_job(job, timeout=job.timeout)
            break
    else:
        return True  # Queue has disappeared, fail job

    return False  # Job is handled. Stop the handler chain.

queues.append(Queue(exc_handler=retry_handler))

除非遇到某种已知的异常类型，否则我决定重试所有错误 3 次。这使我能够尊重可以理解的故障，例如如果在创建作业之后但在执行作业之前删除了用户，或者在图像调整大小作业的情况下，不再找到提供的图像 (HTTP 404)以可读的格式（基本上只要我知道代码永远无法处理这项工作）。

回答你的问题：exc_type 是类，exc_value 是异常实例。 traceback 对于日志记录很有用。如果您关心这一点，请查看 Sentry。如果在上下文中使用SENTRY_DSN 运行，Worker 会自动配置一个 Sentry 错误处理程序。比用错误日志污染自己的数据库要干净得多。

Answer 2

Jökull here 分享的答案是正确的，只是答案（和帖子）已经很老了。我有类似的情况，每当作业失败时，我都需要将重试逻辑放入我的 redis-queue 中。这是队列工作人员的 sn-p（必须将逻辑转移到工作人员）从 Jökull 的答案（和 cmets）redis-queue docs 和 this 中获得灵感：

from rq import Worker, Connection, Queue
from redis import Redis
from app.config import Config

conn = Redis()
max_retries = 3

def retry_handler(job, exc_type, exc_value, traceback):
    # Returning True moves the job to the failed queue (or continue to
    # the next handler)
    job.meta.setdefault('failures', 1)
    job.meta['failures'] += 1
    if job.meta['failures'] > max_retries:
        job.save()
        return True
    # I was unable to locate Status in rq so added the status 'queued' that rq actually expects
    job.status = 'queued' 
    for queue_ in Queue.all(connection=conn):
        if queue_.name == job.origin:
            # at_front=True enqueues the job at the front of the queue for immediate retry.
            queue_.enqueue_job(job, at_front=True) 
            break
    else:
        return True  # Queue has disappeared, fail job

    return False  # Job is handled. Stop the handler chain.

if __name__ == "__main__":
    with Connection(conn):
            worker = Worker(list(map(Queue, ['your_queue_name'])), exception_handlers=[retry_handler])
            worker.work()

Answer 3

1. 更多信息，请阅读sys的文档。

2. False 表示停止处理异常，True 表示继续并落入堆栈中的下一个异常处理程序

同一个作业的多个异常处理程序，type 是异常类型（一个类），你应该更正你的代码，其他错误将返回 None，解释为True，正如 rq 文档所说。