Celery - 运行一组具有复杂依赖关系的任务

Question

在我正在处理的应用程序中，用户可以执行由“步骤”组成的“转换”。一个步骤可以对其他步骤具有任意数量的依赖关系。我希望能够调用转换并将这些步骤作为单独的 Celery 任务并行执行。

理想情况下，我想要一些类似 celery-tasktree 的东西。 ，除了一般的有向无环图，而不仅仅是树，但似乎还没有这样的库存在。

想到的第一个解决方案是标准拓扑排序的并行适应 - 我们不是确定满足依赖关系的步骤的线性顺序，而是确定可以在开始时并行执行的整个步骤集，然后是可以在第 2 轮中执行的整个步骤集，依此类推。

但是，当任务花费的时间可变且工作人员必须空闲等待运行时间更长的任务而有些任务现在已准备好运行时，这不是最佳选择。 (对于我的特定应用程序，这个解决方案现在可能很好，但我仍然想弄清楚如何优化它。)

如 https://cs.stackexchange.com/questions/2524/getting-parallel-items-in-dependency-resolution 中所述，更好的方法是直接在 DAG 之外运行 - 在每个任务完成后，检查它的任何依赖任务现在是否能够运行，如果是，则运行它们。

实现这样的事情的最佳方法是什么？我不清楚是否有一种简单的方法可以做到这一点。

据我所知，Celery 的组/链/和弦基元不够灵活，无法表达完整的 DAG - 尽管我在这里可能错了？

我想我可以为任务创建一个包装器，一旦当前任务完成，它就会通知相关任务 - 我不确定处理这种通知的最佳方法是什么。访问应用程序的 Django 数据库并不是特别简洁，并且很难将其分解为通用库，但 Celery 本身并没有为此提供明显的机制。

Answer 1

我也遇到了这个问题，但除了一个库，我真的找不到更好的解决方案或库，对于任何仍然感兴趣的人，您可以结帐
https://github.com/selinon/selinon .虽然它仅适用于 python 3，但它似乎是唯一可以完全满足您要求的东西。

气流是另一种选择，但气流与其他 dag 库一样用于更静态的环境中。