将 returns 列表的 Celery 任务链接到链中间的一个组中

Question

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

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

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 的 return 值..不，它不能序列化..但请注意，如果我这样做：

>>> 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 本身就是一个签名，我猜这就是为什么 delay() 需要在其中调用 chain..

的原因

如果 I invoke the result as done in the other Whosebug (same link as first) examples 我剩下一个 GroupResult，只有当它是链的最后一个成员时它才会成功 ((), .delay(), .apply_async()).如果我在 GroupResult 上调用 .get() 来获取可序列化的内容，则会出现以下错误： RuntimeError: Never call result.get() within a task! 这给我带来了一个难题；我怎样才能做到这一点？

对这个很困惑..但我也是芹菜的新手。非常感谢有关我如何 could/should 解决此问题的任何建议！

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

Answer 1

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

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

我本来希望它仍然是 chain 的一部分，但现在 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()))

将 returns 列表的 Celery 任务链接到链中间的一个组中

chain a Celery task that returns a list into a group in middle of chain

python

celery