【问题标题】：Create and execute a workflow of tasks with networkx (Python)使用 networkx (Python) 创建和执行任务工作流
【发布时间】：2020-08-18 20:35:40
【问题描述】：

总结：

目标是运行从一组步骤（图节点）及其依赖项（图边）创建的工作流。

是否可以创建networkx.DiGraph() 并运行遍历，在不同节点上执行回调？

示例：

请查看示例图：

对于这组任务和依赖项，执行的可能性是：

步骤a作为入口点
步骤b 紧跟在步骤a 之后
步骤c 和d 在步骤b 之后并行（步骤c 首先完成）
在步骤 c 之后执行步骤 g（而步骤 d 仍在进行中）
步骤 e 紧跟在步骤 d 之后（因为步骤 c 已经完成）
步骤f 紧跟在步骤e 之后

或者，

步骤a作为入口点
步骤b 紧跟在步骤a 之后
步骤c 和d 在步骤b 之后并行（步骤d 首先完成）
步骤g和e并行（因为步骤d在步骤c之前完成）
步骤f 紧跟在步骤e 之后

下面，请在networkx中找到用于创建此示例图的代码sn-p：

import networkx as nx

W = nx.DiGraph()

nodes = ["a", "b", "c", "d", "e", "f"]
edges = [("a", "b"), ("b", "c"), ("c", "g"), ("b","d"), ("c", "e"), ("d", "e"), ("e", "f")]
W.add_nodes_from(nodes)
W.add_edges_from(edges)

是否有已知的解决方案以前面描述的方式遍历图形，打印节点的值（步骤名称）并随机休眠几秒钟以模拟任务运行并执行一些计算？（使用基本的多线程）

谢谢。

【问题讨论】：

不清楚你需要什么。 NetworkX 中有一些遍历算法，例如 BFS 或 DFS。您可以编写一个自定义遍历函数，您可以在其中设置延迟。
简而言之，我正在尝试模拟运行依赖任务的层次结构。我的想法是用一个图来表示工作流，看看是否可以使用networkx来构建图并遍历它。我认为 BFS 比 DFS 更接近可以实现这一点的遍历算法，但我担心这还不够。我将尝试审查我对帖子的表述，希望能进一步澄清。

标签： python multithreading graph task networkx

【解决方案1】：

拓扑排序 (nx.topological_sort) 将返回任务的有效序列。
示例：

>>> list(nx.topological_sort(W))
['a', 'b', 'd', 'c', 'e', 'f', 'g']

如果您想要一组可以同时执行的任务，您可以稍微更改一下以执行分组。

def topological_sort_grouping(g):
    # copy the graph
    _g = g.copy()
    res = []
    # while _g is not empty
    while _g:
        zero_indegree = [v for v, d in _g.in_degree() if d == 0]
        res.append(zero_indegree)
        _g.remove_nodes_from(zero_indegree)
    return res

例子：

>>> topological_sort_grouping(W)
[['a'], ['b'], ['c', 'd'], ['e', 'g'], ['f']]

给定组，您可以迭代它们并同时执行同一组中的组。

【讨论】：