在 Celery 中分别按 id 检索所有链式任务结果

Question

我正在尝试检索存储在 mysql 结果后端中的 celery 中所有链式任务的结果。

比如我有以下两个芹菜任务，

@celery.task(name='celery_fl.add')
def add(x, y, value=None):
    if value is None:
        try:
            return x + y
        except TypeError:
            return None
    return value

@celery.task(name='celery_fl.mul')
def mul(x, y, value=None):
    if value is None:
        try:
            return x * y
        except TypeError:
            return None
    return value

这就是我如何链接它们，

parent = (add.s(2, 2) | mul.s(8)).apply_async()

这里 parent.get() 的输出将是最终链式任务的结果。 parent.parent.get() 会给我第一个链式任务的输出。

我想要实现的是，我想在后期使用任务 ID 获得相同的输出。

task_id = 'bc5fc4b1-613e-4ef0-b5c8-900999d9a6f1'
parent = AsyncResult(task_id, app=celery)

假设我拥有的 task_id 属于链式事件中的第二个任务（父级）。然后，如果我键入 parent.parent.get()，我应该得到第一个链式任务的结果。但不知何故，我得到 None 作为值。是否有另一种方法我应该使用 task_id 而不是 AsyncResult() 来获取任务？

Answer 1

当使用 mysql 后端存储结果时，每个链式任务的结果是分开存储的。但是任务实例不再可用，没有它就不可能使用主任务检索子任务的结果（参考 - Celery tasks）。

所以为了检索所有任务的结果，每个任务的任务ID应该存储在数据库中的某处。

使用烧瓶（python）的示例，

chain = (s3_init.s(order.name, order.id)|create_order_sheet.s(order.id, order.name) | create_order_info.s(order.id, order.name))
res = chain()
process = {
   's3_init': res.parent.parent.parent.parent.parent.parent.id,
   'order_sheet': res.parent.parent.parent.parent.id,
   'order_info': res.parent.parent.parent.id
}
order.update(process_id=json.dumps(process))

然后您可以简单地从数据库中获取任务 ID 并使用 celery.result.AsyncResult(task_id) 按 ID 检索每个任务（参考 - Async results）。