拓扑排序
拓扑排序算法:Kahn 算法(BFS)与 DFS 后序逆序、环检测与关键路径。
1. 拓扑排序原理
1.1 定义
对有向无环图(DAG)的顶点进行线性排序,使得每条边 中 排在 前面。
1.2 应用场景
- 编译依赖分析
- 课程安排
- 任务调度
- 包管理器依赖解析
2. Kahn 算法(BFS)
2.1 算法步骤
1. 计算所有顶点的入度
2. 将入度为0的顶点入队
3. 循环:
a. 出队顶点 v,加入结果
b. 将 v 的所有邻居入度减1
c. 如果邻居入度变为0,入队
4. 如果结果包含所有顶点 → 成功
否则 → 图中有环
2.2 实现
from collections import deque
def kahn_topo_sort(n, edges):
graph = [[] for _ in range(n)]
in_degree = [0] * n
for u, v in edges:
graph[u].append(v)
in_degree[v] += 1
queue = deque([i for i in range(n) if in_degree[i] == 0])
result = []
while queue:
v = queue.popleft()
result.append(v)
for neighbor in graph[v]:
in_degree[neighbor] -= 1
if in_degree[neighbor] == 0:
queue.append(neighbor)
if len(result) == n:
return result # 拓扑排序
return None # 有环
3. DFS 算法
3.1 算法步骤
1. 对每个未访问顶点执行 DFS
2. 在 DFS 完成时(后序)将顶点压入栈
3. 最终栈的逆序即为拓扑排序
4. 如果 DFS 遇到正在访问的顶点 → 有环
3.2 实现
def dfs_topo_sort(n, edges):
graph = [[] for _ in range(n)]
for u, v in edges:
graph[u].append(v)
visited = [0] * n # 0:未访问, 1:正在访问, 2:已完成
result = []
has_cycle = False
def dfs(v):
nonlocal has_cycle
visited[v] = 1
for neighbor in graph[v]:
if visited[neighbor] == 1:
has_cycle = True
return
if visited[neighbor] == 0:
dfs(neighbor)
visited[v] = 2
result.append(v)
for v in range(n):
if visited[v] == 0:
dfs(v)
if has_cycle:
return None
return result[::-1] # 逆序
4. 环检测
4.1 Kahn 算法检测
if len(result) < n:
# 图中有环,未排序的顶点构成环
cycle_nodes = [i for i in range(n) if in_degree[i] > 0]
4.2 DFS 三色标记检测
白色(0): 未访问
灰色(1): 正在访问(在递归栈中)
黑色(2): 已完成
遇到灰色节点 → 发现后向边 → 有环
5. 关键路径(AOE 网)
5.1 最早完成时间
def earliest_time(n, edges, durations):
topo = kahn_topo_sort(n, edges)
if topo is None:
return None
earliest = [0] * n
for v in topo:
for u, v_edge in [(u, v) for u, v in edges if v == v_edge]:
earliest[v] = max(earliest[v], earliest[u] + durations[u])
return earliest
5.2 最晚完成时间
def latest_time(n, edges, durations, project_time):
topo = kahn_topo_sort(n, edges)
latest = [project_time] * n
for v in reversed(topo):
successors = [w for u, w in edges if u == v]
if not successors:
latest[v] = project_time - durations[v]
else:
latest[v] = min(latest[w] for w in successors) - durations[v]
return latest
6. 算法对比
| 维度 | Kahn (BFS) | DFS |
|---|---|---|
| 时间 | ||
| 空间 | ||
| 环检测 | 结果数 < V | 三色标记 |
| 字典序最小 | 优先队列 | 不保证 |
| 并行友好 | 是 | 否 |
字典序最小拓扑排序:用优先队列替代普通队列。
import heapq
def kahn_lexicographic(n, edges):
graph = [[] for _ in range(n)]
in_degree = [0] * n
for u, v in edges:
graph[u].append(v)
in_degree[v] += 1
heap = [i for i in range(n) if in_degree[i] == 0]
heapq.heapify(heap)
result = []
while heap:
v = heapq.heappop(heap)
result.append(v)
for neighbor in graph[v]:
in_degree[neighbor] -= 1
if in_degree[neighbor] == 0:
heapq.heappush(heap, neighbor)
return result if len(result) == n else None