如何阻止Celery在RabbitMQ异常时无限重复执行任务？

针对你的环境（celery==4.2.2、Django==3.0.4、amqp==2.6.1），可以从全局配置和任务级配置两个层面入手，阻止Broker宕机后的无限重复执行：

1. 全局配置（Django settings.py） #

修改Celery相关配置，同时限制连接重试和任务重试：

# 任务重试全局配置
CELERY_TASK_MAX_RETRIES = 3  # 所有任务的最大重试次数
CELERY_TASK_RETRY_BACKOFF = True  # 启用指数退避（重试间隔逐渐增加）
CELERY_TASK_RETRY_BACKOFF_MAX = 300  # 最大重试间隔（5分钟）
CELERY_TASK_REJECT_ON_WORKER_LOST = True  # worker意外退出时拒绝任务，避免重复入队
CELERY_TASK_ACKS_LATE = True  # 任务执行完成后再发送ACK，防止Broker宕机时任务丢失后重复

# RabbitMQ Broker连接配置
CELERY_BROKER_TRANSPORT_OPTIONS = {
    'max_retries': 3,  # 连接Broker的最大重试次数
    'interval_start': 0,
    'interval_step': 0.5,
    'interval_max': 3,
    'connection_timeout': 30,  # 连接超时时间
}

# 禁用Broker连接无限重试（针对启动阶段）
CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP = False

2. 任务级配置（单个shared_task） #

如果需要对特定任务单独设置重试规则，可以在装饰器中指定参数：

from celery import shared_task

@shared_task(
    max_retries=3,
    retry_backoff=True,
    retry_backoff_max=300,
    reject_on_worker_lost=True
)
def your_shared_task():
    # 任务业务逻辑
    pass

关键说明 #

• CELERY_TASK_MAX_RETRIES：直接限制任务的总重试次数，超过后任务会标记为失败，不再重复执行。
• broker_transport_options中的max_retries：控制worker与RabbitMQ连接失败时的重试次数，超过后worker会停止尝试连接，避免无限循环连接导致的任务重复调度。
• task_acks_late和task_reject_on_worker_lost：确保只有任务成功执行完成后才会确认ACK，同时在worker异常退出时拒绝任务，避免RabbitMQ无限制地将任务重新放回队列。

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