在高级工作流程中获取celery任务ID

debugcn 发表于 Dev

炸弹

我需要实现以下情况：

执行任务A
- 与不同的参数并行执行多个任务B
- 等待所有任务完成
- 与不同的参数并行执行多个任务B
- 等待所有任务完成
- 执行任务C

我是通过实现和弦链来实现的，这是简化的代码：

＃内部的run（）方法ATask 
chord_chain = []
为taskB_group在taskB.groups.all（）：
  任务= [。BTask（）SI（ID = taskB_model.id），用于在taskB_model taskB_group.children.all（）]
  如果len（任务）：
    chord_chain.append（chord（任务，_dummy_callback.s（）））
chord_chain.append（CTask（）。si（execution_id））
chain（chord_chain）（）

问题是我需要能够在任何时间点对所有BTask调用revoke（terminate = True）。较低级别的问题是我无法访问BTask的芹菜ID。

试图通过chain获得BTask ID result = chain(chord_chain)()。但是我没有在返回的AsyncResult对象中找到该信息。是否可以从此对象获取链子ID？（result.children为None）
试图通过ATask AsyncResult获取BTask ID，但似乎children属性仅包含第一个和弦的结果，而不包含其余任务的结果。

>>> r = AsyncResult（＃ATask.id＃）
>>> r.children 
[<GroupResult：5599ae69-4de0-45c0-afbe-b0e573631abc [＃BTask.id＃，＃BTask.id＃]>，
<AsyncResult： ＃chord_unlock.id＃>]

炸弹

通过用中止的状态标志标记ATask相关模型并在BTask开始时添加检查来解决。

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。