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

【问题标题】：Run a chord callback even if the main tasks fail即使主要任务失败也运行和弦回调
【发布时间】：2015-06-04 00:33:54
【问题描述】：

即使主要任务失败，是否可以运行chord 回调？

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

我尝试用 si() (immutability) 注册回调

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

我还尝试将参数 ignore_result=True 添加到两个任务装饰器中，但没有成功。

【问题讨论】：

我正在寻找同样的解决方案。

标签： python callback celery chord

【解决方案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()

这将执行和弦，在你的 celery 日志中，你会看到 raise_exception 任务失败，并且 error_callback 的执行将在它的 args 中收到 callback 的 task_id。

此时result 的值将包含callback 的AsyncResult实例，并且因为在和弦中错误传播到执行result.get() 的回调将引发任务异常和result.traceback给你回溯。

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

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

注意

另一个选项是设置CELERY_CHORD_PROPAGATES = False，它将恢复到前 celery 3.1 的行为并始终执行回调。

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

Celery 3.1 定义了如何处理和弦错误，以前的行为从未记录过以及更多的意外，因为它从来没有打算那样工作。

我们无法更改错误修复版本中的行为，因此必须改用设置，但绝不是有人故意禁用新行为。

新行为是为了防止发生此类问题，并且可能会删除向后兼容的设置。我建议你在这里找到一些其他方法来处理错误（如果你可以为它发明一个好的 api，我不介意提出建议）

【讨论】：

如何在error_callback中得到好任务的结果？
@Carl 我不是 100% 肯定，但我认为你不能。我认为您可以获得的唯一属性是失败任务的异常和回溯。如果您需要稍后检索特定的内容，即使一切都失败了，您也可以使用数据存储。我们只需要它来发送和发送电子邮件，所以我没有玩太多。
根据接受代码，我得到错误回调运行，但弦函数仍然引发异常，而不是从错误回调返回值...
如何将参数传递给link_error的函数？
我认为您不应该将 link_error 附加到回调中。 link_error 应该附加到和弦（任务）等等。

【解决方案2】：

你只需要改变link_error 的调用方式。使用您想要的参数传递签名，而不是字符串引用。

在上面的例子中，你可以传递如下参数

c = chord(
    [raise_exception.s(), good.s(), raise_exception.s()], 
    callback.s().set(link_error=[error_callback.s(<arguments_here>)])
)

请记住，第一个参数是task_id，后面的参数是签名中定义的参数。

【讨论】：