我相当确定只能通过创建自己的任务类来完成这个操作,但我想知道是否有其他人找到了其他方法。
我相当确定只能通过创建自己的任务类来完成这个操作,但我想知道是否有其他人找到了其他方法。
以下是完整的解决方案(适用于Celery 4+):
import celery
from celery.task import task
class MyBaseClassForTask(celery.Task):
def on_failure(self, exc, task_id, args, kwargs, einfo):
# exc (Exception) - The exception raised by the task.
# args (Tuple) - Original arguments for the task that failed.
# kwargs (Dict) - Original keyword arguments for the task that failed.
print('{0!r} failed: {1!r}'.format(task_id, exc))
@task(name="foo:my_task", base=MyBaseClassForTask)
def add(x, y):
raise KeyError()
资源:
def fun(self, exc, task_id, args, kwargs, einfo):
print('Failed!')
@task(name="foo:my_task", on_failure=fun)
def add(x, y):
raise KeyError()