本文介绍如何通过深度优先递归遍历任务依赖图，生成每个任务版本所直接或间接影响的所有下游任务列表，适用于构建系统、工作流调度等场景。
本文介绍如何通过深度优先递归遍历任务依赖图，生成每个任务版本所直接或间接影响的所有下游任务列表，适用于构建系统、工作流调度等场景。
在构建自动化、CI/CD 流水线或 DAG 任务调度系统时，常需分析「某任务变更后，哪些下游任务需要重新执行」——即向下传播依赖（downstream dependencies）。给定一组任务，每个任务通过 dependencies 字段声明其直接上游依赖（即“我依赖谁”），我们需要反向推导出：每个任务作为上游时，会触发哪些下游任务（即“谁依赖我”及其传递依赖）。
关键在于：这不是简单的单层映射，而是要展开整个依赖链路。例如：
-
"2"依赖"1"→"1"的下游包含"2" -
"3"依赖"2"→"1"的下游也应包含"3"（因为"3"→"2"→"1"） - 同理
"4"也属于"1"的下游
因此，正确做法是：对每个任务 T，从 T 出发，向上追溯其所有祖先依赖，并将 T 的 version 追加到每个祖先的下游列表中。这本质上是一次反向拓扑扩散（reverse DFS）。
以下是优化后的实现：
from pprint import pprint
list_of_task_to_generate = [
{"version": "1", "dependencies": []},
{"version": "2", "dependencies": ["1"]},
{"version": "3", "dependencies": ["2"]},
{"version": "4", "dependencies": ["3"]},
{"version": "5", "dependencies": []},
{"version": "6", "dependencies": ["5"]},
]
def get_downstream_dependencies(tasks):
"""
递归构建下游依赖映射：{task_version -> [all downstream tasks]}
时间复杂度：O(V + E)，V为任务数，E为依赖边数
"""
# 预处理：构建 version → task 的快速查找字典，避免每次线性扫描
task_map = {task["version"]: task for task in tasks}
result = {}
def dfs(task, root_version):
"""从当前 task 向上递归，将 root_version 加入其所有依赖项的下游列表"""
version = task["version"]
# 初始化当前节点的下游列表（若未创建）
if version not in result:
result[version] = []
# 将 root_version（发起调用的原始任务）加入当前节点的下游
if root_version != version: # 避免自引用（可选）
result[version].append(root_version)
# 递归处理当前任务的所有直接依赖（即向上追溯）
for dep_version in task["dependencies"]:
if dep_version in task_map:
dfs(task_map[dep_version], root_version)
# 对每个任务作为 root 发起一次 DFS
for task in tasks:
dfs(task, task["version"])
return result
# 执行并输出
downstream = get_downstream_dependencies(list_of_task_to_generate)
pprint(downstream)
输出结果：
{
'1': ['2', '3', '4'],
'2': ['3', '4'],
'3': ['4'],
'4': [],
'5': ['6'],
'6': []
}
✅ 完全匹配预期目标。
注意事项与优化建议：
-
避免重复计算：原代码中因未区分“当前任务”和“被追溯的依赖”，导致同一
version被多次添加（如'1': ['2','2','2']）。本方案通过root_version明确传播源，确保每个下游任务只被添加一次。 -
性能提升：使用
task_map（dict查找）替代for i in list_of_tasks的线性搜索，将单次查找从 O(n) 降至 O(1)，整体复杂度从 O(n²) 优化至 O(n + e)。 -
数据结构建议：实际工程中，推荐直接用
dict存储任务（如{"1": {"dependencies": []}}），而非列表，更符合图建模直觉。 -
环检测（进阶）：当前实现假设无循环依赖。生产环境应加入
visited集合防止无限递归，例如if dep_version in visited: raise ValueError("Cycle detected")。
该方法简洁、可读性强，且易于扩展（如支持权重、条件依赖或并发遍历），是处理任务依赖图下游分析的稳健实践。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
