python - 将任务签名设置为 celery 中多链的起始任务

Question

假设 A 到 G 是异步任务，我们想实现一个像 Canvas 这样的任务工作流

     /---> B()
A()  ----> C() -> D()
     \---> E() -> F() -> G()

根据问题，我们需要使用A()的结果来启动多个链。

在代码中:

from celery import chain

# A (params) ?????

B.apply_async(params)
chain(
    C.s(params), D.s(params)
).apply_async()
chain(
    E.s(params), F.s(params), G.s(params)
).apply_async()

在 Celery Canvas 中，如何在不运行 A() 三次的情况下将 A() 的结果异步用于多个链的第一个元素？

(我使用 Celery 4.1.0 和 RabbitMQ 作为代理)

Answer 1

我知道! b, c, e 不保留顺序，但 a 肯定会在所有这些之前完成，因为有一个链.我只想将a的返回值作为b、c和e的输入参数。