即使主要任务失败,也要运行和弦回调

Question

即使主要任务失败,是否可以运行和弦回调？

我创建了一个和弦,我添加了一堆任务并注册了一个回调.我的问题是,如果其中一个任务失败,则不会触发回调,但我希望以任一方式触发回调.

我试图用si()注册回调(不变性)

callback = tasks.run_delete_rule.si([timestamp])
header = [tasks.run_update_rule.s(i, timestamp) for i in item_ids]
result = chord(header)(callback)

我也尝试将param添加ignore_result=True到两个任务装饰器中,但没有成功.

Answer 1

从github问题#1881,如果回调具有link_error选项集,其中包含任务名称列表,那么当和弦的任务失败时,将执行link_error任务.

@task(name='super_task.good')
def good():
    return True

@task(name='super_task.raise_exception')
def raise_exception():
    raise ValueError('error')

@task(name='super_task.callback')
def callback(*args, **kwargs):
    logger.info('callback')
    logger.info(args)
    logger.info(kwargs)
    return 'finished'

@task(name='super_task.error_callback')
def error_callback(*args, **kwargs):
    logger.info('error_callback')
    logger.info(args)
    logger.info(kwargs)
    return 'error'

>>> c = chord(
        [raise_exception.s(), good.s(), raise_exception.s()], 
        callback.s().set(link_error=['super_task.error_callback'])
    )
>>> result = c()

这将执行和弦,在您的芹菜日志中,您将看到raise_exception任务失败,并且执行error_callback将在其中接收任务的task_id callback.

此时,值result将包含AsyncResult实例callback,并且因为在和弦中传播到回调的错误result.get()会引发任务异常并result.traceback为您提供回溯.

如果你想要一个回调,只需将和弦回调的名称传递给 link_error

callback.s().set(link_error='super_task.callback')

注意

设置的另一个选项CELERY_CHORD_PROPAGATES = False将恢复到芹菜前的行为并始终执行回调.

但这不是推荐的方法,因为你可以在github问题#1349中找到

Celery 3.1定义了和弦错误的处理方式,之前的行为从未被记录过,更多的是因为从来没有打算以这种方式工作.

我们无法更改错误修正版本中的行为,因此必须使用设置,但从来没有意图有人故意禁用新行为.

新行为可以防止发生此类问题,并且可以删除向后兼容设置.我建议你在这里找到一些其他方法来处理错误(如果你能为它发明一个很好的API,我不介意提议)