python - 如何处理 celery 的 Task.map 中的错误

Question

假设我有两个 celery 任务:

@celery.task
def run_flakey_things(*args, **kwargs):
    return run_flakey_and_synchronous_thing.map(
        xrange(10)
    ).apply_async()


@celery.task
def run_flakey_and_synchronous_thing(a):
    if a % 5:
        return a
    raise RuntimeError(a)

因此，当您运行 run_flakey_things 时，它会立即失败，因为序列中的第一项会引发异常。我想要的是像 map 一样按顺序为序列中的所有项目运行任务，但继续运行异常，一旦所有这些都完成就引发一个新的异常。

理想情况是我可以在应用它之前向 xmap 对象添加一个 on_failure，但是 xmap 似乎不是一个完整的任务对象。

Answer 1

您可以更改返回值以指示和传播错误。像这样:

import traceback

@celery.task
def run_flakey_things(*args, **kwargs):
    return run_flakey_and_synchronous_thing.map(
        xrange(10)
    ).apply_async()


@celery.task
def run_flakey_and_synchronous_thing(a):
    d = {'value': None, 'error': None}
    try:
        if a % 5:
            d['value'] = a
    except:
        d['error'] = traceback.format_exc()
    return d

然后你可以做一些事情:

1) 更改您的 run_flakey_things 以将有错误和无错误的事物分组，返回没有错误的事物并报告有错误的事物。

2) 在调用 run_flakey_things 时处理该行为