celery - 让芹菜分布在所有任务中

Question

我有多个任务要设置为并行执行。

• 更新包：返回包列表。
• 更新版本：接受一个包并返回一个版本列表。
• 更新版本：获取版本（用于包）并获取版本。

这让我得到了类似的东西：

@task()
def update_packages():
    return [1, 2, 3]

@task()
def update_versions(package):
    # Get versions
    return [1, 2, 3]

@task()
def update_releases(version):
    # Get releases

我能做的是按顺序执行它们并等待结果，但我宁愿将中间结果向前推进，就像在 shell 上一样：

update_packages | update_versions | update_releases

什么魔法调用可以做到这一点？

Answer 1

我认为，您正在寻找组和Scatter-Gather模式：

@task()
def update_packages():
    res = group(update_versions.s(i) for i in [1, 2, 3])() # run tasks in parallel (Scatter)
    res.get() # wait for all results (Gather)
    return res

@task()
def update_versions(package):
    # Get versions
    res = group(update_packages.s(i) for i in [1, 2, 3])() # run tasks in parallel (Scatter)
    res.get() # wait for all results (Gather)
    return res

@task()
def update_releases(version):
    # Get releases
    return <what you want to see in final>

现在您可以简单地运行update_packages并等待所有结果：

res = update_packages()

您不需要使用.delay，因为update_packages它本身不做任何工作。