Linux下如何处理Celery的任务失败-乐工具技术知识

在 Linux 下处理 Celery 任务失败，可以采取以下几种方法：

使用任务重试（Retry）：

当一个任务失败时，Celery 可以自动重试该任务。你可以在任务定义中设置重试次数和重试间隔：

from celery import Celery

app = Celery('tasks', broker='pyamqp://guest@localhost//')

@app.task(bind=True, default_retry_delay=30, max_retries=5)
def my_task(self, *args, **kwargs):
    try:
        # Your task code here
        pass
    except Exception as exc:
        raise self.retry(exc=exc)

这里，default_retry_delay 设置了重试间隔（单位为秒），max_retries 设置了最大重试次数。

使用错误回调（Error Callback）：

当任务失败时，你可以定义一个错误回调函数来处理失败的任务。错误回调函数接收任务的异常信息、任务 ID、任务参数等作为参数。

from celery import Celery

app = Celery('tasks', broker='pyamqp://guest@localhost//')

@app.task(bind=True)
def my_task(self, *args, **kwargs):
    try:
        # Your task code here
        pass
    except Exception as exc:
        self.on_failure(exc, task_id=self.request.id, args=args, kwargs=kwargs)
        raise

@app.task
def on_failure(self, exc, task_id, args, kwargs, einfo):
    # Handle the failed task here
    print(f"Task {task_id} failed with exception: {exc}")