如何避免Celery在任务抛出异常时自动确认任务？

我希望防止Celery在任务抛出异常时确认任务，但设置acks_late=True并未生效。

代码示例：

@app.task(acks_late=True)
def send_mail(data):
    raise Exception

我尝试了自定义任务基类的方式：

class CeleryTask(celery.Task):

    def on_failure(self, exc, task_id, args, kwargs, einfo : ExceptionInfo):
        
        app.send_task(self.name, args=args, kwargs=kwargs)

    def run(self, *args, **kwargs):
        pass


@app.task(base=CeleryTask)
def send_mail(data):
    raise Exception

但这会导致无限递归，想知道是否有更优的解决方案？

1. 利用Celery原生重试机制（推荐） #

无需手动重新发送任务，直接通过Celery自带的重试配置实现异常时任务重新入队，同时避免无限递归：

@app.task(acks_late=True, autoretry_for=(Exception,), retry_backoff=3, retry_kwargs={'max_retries': 5})
def send_mail(data):
    raise Exception

• autoretry_for：指定触发自动重试的异常类型
• retry_backoff：设置重试间隔的指数退避时间（单位：秒），避免短时间内重复重试
• retry_kwargs：限制最大重试次数，防止无限循环

2. 正确实现自定义任务基类 #

如果必须使用自定义基类，要在on_failure中添加重试次数限制，并用Celery内置的retry方法而非手动发送任务：

class CeleryTask(celery.Task):
    max_retries = 5

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        retries = kwargs.get('retries', 0)
        if retries < self.max_retries:
            kwargs['retries'] = retries + 1
            self.retry(args=args, kwargs=kwargs, exc=exc)
        else:
            # 超过重试次数后执行的收尾逻辑，比如记录日志、告警等
            pass

@app.task(base=CeleryTask, acks_late=True)
def send_mail(data):
    raise Exception

self.retry()会由Celery统一处理任务的重试逻辑，自带重试次数追踪，不会触发无限递归。

3. 检查全局配置与Broker设置 #

确认Celery全局配置中开启了task_acks_late，同时确保使用的Broker（如RabbitMQ）没有强制自动确认任务：

app = Celery('tasks', broker='amqp://guest@localhost//')
app.conf.update(
    task_acks_late=True,
)

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