如何使用Celery autoretry_for实现优雅的异常处理？

我之前也碰到过这个一模一样的坑——用autoretry_for配置了预期异常自动重试，但重试次数耗尽后抛出的ExpectedException还是被Celery当成了意外异常，折腾了好一会儿才找到几个优雅的解决方案，不用写一堆重复的显式捕获代码：

方案1：用throws参数声明预期异常 #

这是最简洁的方案，只需要给任务添加throws参数，告诉Celery这些异常是任务预期会抛出的，即使重试耗尽后抛出，也不会被标记为意外异常。

示例代码：

from celery import shared_task

class ExpectedException(Exception):
    pass

@shared_task(
    autoretry_for=(ExpectedException,),
    max_retries=3,
    default_retry_delay=5,
    throws=(ExpectedException,)  # 核心配置
)
def my_task():
    # 你的任务逻辑，可能抛出ExpectedException
    raise ExpectedException("业务逻辑触发了预期异常")

原理：Celery默认会把未在throws中声明的异常视为意外异常，添加这个参数后，即使任务最终因重试耗尽抛出ExpectedException，Celery也会将其归类为「预期内的任务失败」，不会触发额外的意外异常告警或标记。

方案2：自定义任务基类封装重试与失败处理 #

如果你有多个任务都需要这个逻辑，可以封装一个自定义任务基类，把重试配置和失败处理逻辑统一起来，避免重复代码：

from celery import Task, shared_task

class ExpectedException(Exception):
    pass

class ExpectedRetryTask(Task):
    # 统一配置自动重试规则
    autoretry_for = (ExpectedException,)
    max_retries = 3
    default_retry_delay = 5
    # 声明预期异常
    throws = (ExpectedException,)

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        # 这里可以添加自定义的优雅处理逻辑，比如日志记录、告警通知等
        if isinstance(exc, ExpectedException):
            self.logger.info(f"任务 {task_id} 达到最大重试次数，预期内失败: {exc}")
        # 非预期异常交给父类处理
        super().on_failure(exc, task_id, args, kwargs, einfo)

# 使用自定义基类创建任务
@shared_task(base=ExpectedRetryTask)
def my_task():
    raise ExpectedException("业务逻辑触发了预期异常")

为什么你之前用on_failure无效？大概率是没结合throws参数，或者没正确判断异常类型——单纯重写on_failure只能处理失败后的逻辑，但如果不声明throws，Celery还是会把这个异常标记为意外。

方案3：手动控制重试逻辑（可选） #

如果你需要更精细的控制，可以在任务中手动捕获异常并调用retry()，同时指定max_retries_exceeded_callback来处理重试耗尽的情况：

from celery import shared_task
from celery.exceptions import MaxRetriesExceededError

class ExpectedException(Exception):
    pass

def handle_max_retries(task, exc, args, kwargs):
    # 重试耗尽后的自定义处理
    task.logger.info(f"任务 {task.request.id} 重试耗尽，预期内失败: {exc}")

@shared_task(max_retries=3, default_retry_delay=5)
def my_task():
    try:
        # 你的任务逻辑
        raise ExpectedException("业务逻辑触发了预期异常")
    except ExpectedException as exc:
        # 手动重试，指定耗尽后的回调
        my_task.retry(exc=exc, max_retries_exceeded_callback=handle_max_retries)

不过这个方案需要手动捕获异常，不如前两个方案简洁，适合需要极致定制的场景。

内容的提问来源于stack exchange，提问作者Greg Lever