Postgres重启引发Celery OperationalError时，如何自动重试任务？

这个问题我之前帮不少开发者解决过——Postgres重启导致Celery任务批量触发OperationalError确实头疼，不用给每个数据库命令套try/except的话，有几个优雅的方案可以实现自动重试：

方法1：全局或任务级自动重试配置 #

Celery本身提供了自动重试装饰器参数，可以直接指定针对特定异常自动重试，无需手动写try/except块。

单个任务配置 #

如果只想给特定任务开启这个机制，直接在任务装饰器里加参数：

from celery import shared_task
from psycopg2 import OperationalError  # 用原生psycopg2时导入
# 若使用Django ORM，替换为 from django.db.utils import OperationalError

@shared_task(
    autoretry_for=(OperationalError,),  # 指定要自动重试的异常类型
    retry_backoff=30,  # 重试间隔（秒）
    retry_jitter=False,  # 关闭随机抖动，确保每次精确30秒后重试
    retry_kwargs={'max_retries': 5}  # 最大重试次数，设为None则无限重试
)
def your_database_task():
    # 你的数据库操作代码，无需额外加try/except
    pass

全局配置 #

如果想让所有任务都默认处理这个异常，直接在Celery实例的配置里设置：

from celery import Celery
from psycopg2 import OperationalError

app = Celery('your_app')
app.conf.update(
    task_autoretry_for=(OperationalError,),
    task_retry_backoff=30,
    task_retry_jitter=False,
    task_retry_kwargs={'max_retries': 5}
)

方法2：自定义任务基类 #

如果部分任务需要不同的重试策略（比如有的任务要更长的延迟），可以创建一个自定义任务基类，让需要的任务继承它：

from celery import Task
from psycopg2 import OperationalError

class DBRetryTask(Task):
    # 基类里统一定义重试规则
    autoretry_for = (OperationalError,)
    retry_backoff = 30
    retry_jitter = False
    retry_kwargs = {'max_retries': 5}

# 继承基类创建任务
@app.task(base=DBRetryTask)
def another_db_task():
    # 数据库操作代码
    pass

关键注意事项 #

• 异常类要对应正确：用Django ORM时捕获django.db.utils.OperationalError，原生psycopg2则用psycopg2.OperationalError，别导错了。
• 指数退避vs固定延迟：Celery的retry_backoff默认是指数退避（第一次1秒，第二次2秒，以此类推），需要固定30秒延迟的话，一定要加retry_jitter=False。
• 重试次数控制：根据业务场景调整max_retries，如果任务必须执行成功可以设为None无限重试，但要注意Postgres长期不可用的极端情况。

这套方案完全兼容Redis作为broker的场景，不用额外修改中间件配置，就能轻松解决Postgres重启后的任务重试问题。

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