python - 将一个返回列表的 Celery 任务链接到链中间的一个组中

Question

这个问题和这个问题是一样的: How to chain a Celery task that returns a list into a group?除了我需要这发生在链的中间，并且接受的解决方案仅在中间任务是链中的最终“链接”时才有效。

下面是同一个示例，稍微修改后重现了该问题:

from random import random

from celery import 

@app.task
def get_list(amount):
    return [i for i in range(amount)]

@app.task
def process_item(item):
    return [f'id-{item}', random() > .5]

@app.task
def dmap(it, callback):
    # Map a callback over an iterator and return as a group
    callback = subtask(callback)
    return group(callback.clone([arg,]) for arg in it)()

@app.task
def handle_results(results):
    for result in results:
        if result[1] == None:
            continue

        return result[1] # return the first True value

def foo():
    return chain(
        get_list.s(10),
        dmap.s(process_item.s()),
        handle_results.s() # <-- if I add this, it fails
    )

# in a terminal, or somewhere
foo()()

我得到的错误是这样的:

File "/usr/local/Cellar/python/3.7.4_1/Frameworks/Python.framework/Versions/3.7/lib/python3.7/json/encoder.py", line 179, in default raise TypeError(f'Object of type {o.class.name} ' kombu.exceptions.EncodeError: Object of type GroupResult is not JSON serializable

毕竟这是 dmap 的返回值..不，它不能被序列化..但请注意，如果我这样做:

>>> lst = [i for i in range(amount)]
>>> chain(group(process_item.s(i) for i in lst), handle_results.s())

然后就可以了。我对实际需要从链的一个成员传递到另一个成员的内容感到困惑。group(...) 的结果是:

>>> from app.manager_tasks import process_item
>>> group(process_item.s(e) for e in [1, 2, 3, 4])
group([app.manager_tasks.process_item(1), process_item(2), process_item(3), process_item(4)])
>>> group(process_item.s(e) for e in [1, 2, 3, 4]).delay()
<GroupResult: 07c9be1a-b3e3-4da2-af54-7177f3d91d0f [cf777f54-4763-46bd-a405-2c1993ddbf66, 103298fc-8f1f-4183-ba45-670224fcd319, 3ad87c2c-7b64-4309-a61b-e53ae17302b9, bf2766a3-662a-415d-a35b-037a0476f4a4]>

这是一个 GroupResult 本身(调用了延迟)，否则只是一个组。由于 dmap 本身就是一个签名，我猜这就是为什么需要在其中为 chain 调用 delay() 的原因...... 🤔

如果I invoke the result正如在另一个stackoverflow (same link as first) 中所做的那样examples我留下了一个 GroupResult，它只有在它是链的最后一个成员时才会成功(()、.delay()、.apply_async())。如果我在 GroupResult 上调用 .get() 以获取可序列化的内容，则会收到以下错误:RuntimeError: Never call result.get() within a task! 这给我带来了一个难题；我怎样才能做到这一点？

我对这个很困惑……但我对 celery 还是陌生的。非常感谢有关我如何/应该如何解决此问题的任何建议!

更多背景知识，我打算重复使用此链作为另一个链的一部分，该链位于顶层，指定管道中的阶段。

Answer 1

正如@DejanLekic 提到的，我应该一直使用chord。这将解决上述问题:

def foo():
    return chord(
        get_list.s(10),
        dmap.s(process_item.s())
    )(handle_results.s())

我希望它仍然是链的一部分，但它doesn't look like that is supported现在。

---

以下内容与问题关系不大，但可能对某些人有用。

使用来自那个 github 问题线程的解决方案，我仍然可以通过嵌套和弦和链来做我需要的(在弄清楚主要问题之后)。不是最干净的，但它可以工作......看起来像这样:

def foo():
    return chord(
        get_list.s(10),
        dmap.s(process_item.s())
    )(chain(handle_results.s(), log_stuff.s()))