DAG-有向无环图-拓扑排序

1. 场景

通过当前节点与依赖节点列表描述一个有向无环图DAG节点依赖问题，适合流程图中节点依赖关系的定义，适合存在明确的依赖关系并且按依赖顺序执行的领域

• 项目管理与任务调度
• 工作流与审批流程

2. 数据描叙

• name：描叙流程图中节点名称
• dependencies: 描叙当前节点依赖的父级节点列表，dependencies为空则表示根节点

[{"dependencies": ["task-0711-1761200645304"],"name": "task-0711-1761200698814"},{"name": "task-0711-1761200645304"},{"dependencies": ["task-0711-1761200698814"],"name": "task-0711-1761204095457"},{"dependencies": ["task-0711-1761204095457"],"name": "task-0711-1761204132857"}
]

3. 输出从根节点到末尾节点（排好序的）

• 方法1思路：
  • a. 首先遍历找出首节点
  • b. 找出首节点后依次找出后继节点添加到末尾
  • c. 时间复杂度:O(n^2) 空间复杂度: O(n)
  • d. 缺陷: 无法遍历出多节点的依赖的

task_dag = [{"dependencies": ["task-0711-1761200645304"],"name": "task-0711-1761200698814"},{"name": "task-0711-1761200645304"},{"dependencies": ["task-0711-1761200698814"],"name": "task-0711-1761204095457"},{"dependencies": ["task-0711-1761204095457"],"name": "task-0711-1761204132857"}
]def sort_dag(dag: list) -> []:nodes = []pre_node = Nonewhile len(dag) > 0:for index in range(len(dag)):deps = dag[index].get("dependencies", [])# 没有依赖的认为是第一个节点if not deps:nodes.insert(0, dag[index])pre_node = dag.pop(index)break# 存在依赖，依照顺序添加节点if pre_node and pre_node.get("name") in deps:pre_node = dag.pop(index)nodes.append(pre_node)breakreturn nodesif __name__ == '__main__':nodes = sort_dag(task_dag)print([n.get('name') for n in nodes])print(sort_dag(task_dag))pass

• 方法2-拓扑排序思路
  • 依次建立拓扑的正向依赖和反向依赖
  • 计算正向依赖的入度，入度为0表示没有前置依赖，可以立即执行，进入队列
  • 执行完当前任务，减少其他依赖该任务的入度，将入度为0的任务添加到待执行任务列表
  • 时间复杂度 O(V+E) 空间复杂度 O(V+E) V表示任务数，E表示依赖边数

from collections import dequetask_dag = [{"dependencies": ["task-0711-1761200645304"],"name": "task-0711-1761200698814"},{"name": "task-0711-1761200645304"},{"dependencies": ["task-0711-1761204095457"],"name": "task-0711-1761204132857"},{"dependencies": ["task-0711-1761200698814"],"name": "task-0711-1761204095457"},
]def topology_sort(dag):# 1. 获取所有任务ID标识tasks = [task.get("name") for task in dag]# 2. 建立正向依赖：当前任务执行时，需要执行的前置任务(当前任务依赖前置任务完成)# 3. 建立反向依赖：当前任务执行时，需要执行的后置任务（后置任务依赖当前任务完成）dependencies_forward = {}dependencies_resolve = {}for node in dag:# 使用task_name作为唯一标识task_name = node.get("name")dependencies_forward[task_name] = node.get("dependencies", [])for dep in dependencies_forward[task_name]:# 校验是否存在不在dag中的任务if dep not in tasks:raise Exception(f"任务:{dep} 不在dag中，dag数据错误")if dep not in dependencies_resolve:dependencies_resolve[dep] = []dependencies_resolve[dep].append(task_name)# 4. 计算入度task_degree = {node.get("name"): len(node.get("dependencies", [])) for node in dag}# 5. 入度为0，进入待执行队里exec_queue = deque()for task, degree in task_degree.items():if degree == 0:exec_queue.append(task)# 6. 出待执行队列，执行任务，当前任务执行完成，意味着后置任务可以执行，更新当前任务关联的入度减去1sorted_task = []while exec_queue:current_task = exec_queue.popleft()# todo: 拿出具体任务执行的，这里是具体业务处理步骤# 简单的将当前入度为0立即执行的任务添加到执行顺序列表sorted_task.append(current_task)# 执行完成，减少任务的入度for dep in dependencies_resolve.get(current_task, []):task_degree[dep] -= 1if task_degree[dep] == 0:exec_queue.append(dep)# 判断是否存在环if len(sorted_task) != len(tasks):raise Exception(f"dag存在环")# 校验：# 1. 是否出现不在dag中的任务，dag图错误# 2. 是否存在dag环，存在dag环的话，入度始终不会为0return sorted_taskpassif __name__ == '__main__':print(topology_sort(task_dag))