拓扑排序实战：用Python手把手解决课程安排问题（附LeetCode例题）

拓扑排序实战：用Python手把手解决课程安排问题（附LeetCode例题）

最近在帮一个做在线教育平台的朋友优化他们的课程推荐系统，遇到了一个挺有意思的问题：用户报了一堆有先修要求的课程，系统怎么才能自动生成一个合理的学习顺序，保证不会出现“还没学微积分就去学高等数学”的尴尬？这让我想起了算法里一个经典的工具——拓扑排序。它不是什么高深莫测的魔法，而是处理这类“依赖关系”问题的瑞士军刀。无论是课程安排、任务调度、还是软件构建时的依赖解析，背后都可能藏着它的身影。如果你正在准备技术面试，或者想在实际项目中优雅地解决这类先后顺序问题，掌握拓扑排序的Python实现，绝对能让你事半功倍。今天，我们就抛开枯燥的理论，直接从LeetCode上的经典例题入手，一步步拆解，看看如何用Python把拓扑排序“玩”起来。

1. 理解核心：当任务有了“先后”约束

我们生活在一个充满依赖的世界里。穿袜子必须在穿鞋之前，学习函数是学习微积分的基础，编译代码前需要先解决所有依赖库。这些关系，在计算机科学里，可以用一种叫做有向无环图的结构来完美建模。

想象一下，你要学习一系列课程，有些课需要先修其他课。我们可以把每门课看作图中的一个“节点”，如果课程A是课程B的先修课，我们就画一条从A指向B的“边”。这样一来，所有课程和先修关系就构成了一张图。一个至关重要的前提是：这张图里不能有环。什么叫环？就是A依赖B，B依赖C，C又回头依赖A，形成了一个死循环。在有环的情况下，你永远找不到一个合理的开始顺序，就像“先有鸡还是先有蛋”的悖论。因此，我们处理的对象特指有向无环图。

拓扑排序，就是给这张DAG中的所有节点安排一个线性的顺序，保证对于任何一条从节点U到节点V的边，U在顺序中都排在V的前面。它输出的不是一个唯一的序列，可能有很多种合法的排序结果。

注意：拓扑排序只适用于有向无环图。如果图中存在环，则无法进行拓扑排序。在实际应用中，检测图中是否存在环往往是解决问题的第一步。

为了进行拓扑排序，我们需要关注每个节点的“入度”，即有多少条边指向它。入度为0的节点，意味着没有任何先决条件，可以立刻开始执行。整个算法的核心思想，就是不断地“移除”这些入度为0的节点，并更新剩余节点的入度。

2. 算法骨架：Kahn算法与BFS的默契配合

最经典、最直观的拓扑排序实现方法是Kahn算法，它本质上是一种广度优先搜索。我们用一个具体的课程安排场景来模拟这个过程。

假设我们有4门课程，编号为0到3，先修关系如下：

• 要学习课程1，必须先完成课程0。
• 要学习课程3，必须先完成课程1和课程2。
• 课程2没有先修要求。

我们可以用邻接表来表示这个图，并计算每个课程的入度。

# 课程先修关系: prerequisites[i] = [ai, bi] 表示要学习课程 ai，必须先完成课程 bi prerequisites = [[1,0], [3,1], [3,2]] numCourses = 4 # 构建邻接表和入度数组 from collections import defaultdict, deque adj_list = defaultdict(list) # 邻接表，存储每个节点的后继节点 in_degree = [0] * numCourses # 入度数组 for course, prereq in prerequisites: adj_list[prereq].append(course) # prereq -> course in_degree[course] += 1 print("邻接表:", dict(adj_list)) print("入度数组:", in_degree)

运行上面的代码，你会得到：

邻接表: {0: [1], 1: [3], 2: [3]} 入度数组: [0, 1, 0, 2]

解释：课程0指向课程1，课程1指向课程3，课程2指向课程3。课程0和2的入度为0，课程1的入度为1（来自课程0），课程3的入度为2（来自课程1和2）。

接下来是Kahn算法的步骤：

1. 初始化队列：将所有入度为0的节点加入队列。
2. 处理队列： a. 从队列中取出一个节点，将其加入拓扑排序的结果序列。 b. 遍历该节点的所有后继节点，将它们的入度减1（相当于“移除”当前节点及其发出的边）。 c. 如果某个后继节点的入度因此变为0，则将其加入队列。
3. 检查结果：如果结果序列包含了图中所有的节点，则拓扑排序成功；否则，说明图中存在环。

我们用代码来实现这个流程：

def topological_sort_kahn(numCourses, prerequisites): adj_list = defaultdict(list) in_degree = [0] * numCourses # 1. 构建图并计算入度 for course, prereq in prerequisites: adj_list[prereq].append(course) in_degree[course] += 1 # 2. 初始化队列，放入所有入度为0的节点 queue = deque([i for i in range(numCourses) if in_degree[i] == 0]) topo_order = [] # 3. BFS过程 while queue: node = queue.popleft() topo_order.append(node) # 遍历后继节点 for neighbor in adj_list[node]: in_degree[neighbor] -= 1 if in_degree[neighbor] == 0: queue.append(neighbor) # 4. 判断是否有环 if len(topo_order) == numCourses: return topo_order # 成功，返回拓扑序 else: return [] # 失败，存在环，无法拓扑排序 # 测试我们之前的例子 result = topological_sort_kahn(4, [[1,0], [3,1], [3,2]]) print("拓扑排序结果:", result) # 输出可能是 [0, 2, 1, 3] 或 [2, 0, 1, 3]

这个算法的时间复杂度是 O(V + E)，其中V是节点数，E是边数，对于稀疏图来说非常高效。空间复杂度主要是存储邻接表和队列，也是 O(V + E)。

3. 实战演练：LeetCode课程表问题深度剖析

理论讲完了，是时候真刀真枪地解决问题了。LeetCode上有两道经典的“课程表”问题，完美契合拓扑排序的应用场景。

3.1 LeetCode 207. 课程表

问题描述：你这个学期必须选修numCourses门课程，记为0到numCourses - 1。在选修某些课程之前需要一些先修课程。先修课程按数组prerequisites给出，其中prerequisites[i] = [ai, bi]，表示如果要学习课程ai则必须先学习课程bi。请你判断是否可能完成所有课程的学习？如果可以，返回true；否则，返回false。

本质：这就是判断课程依赖关系图（一个有向图）中是否存在环。如果存在环，则无法完成所有课程；如果是DAG，则可以完成。

思路：直接使用我们上面实现的Kahn算法。如果最终拓扑序列包含了所有课程，说明无环，返回True；否则，返回False。

def canFinish(numCourses, prerequisites): from collections import defaultdict, deque adj = defaultdict(list) indeg = [0] * numCourses # 建图 for course, pre in prerequisites: adj[pre].append(course) indeg[course] += 1 # 初始化队列 q = deque([i for i in range(numCourses) if indeg[i] == 0]) count = 0 # 记录已排序的课程数 # BFS拓扑排序 while q: node = q.popleft() count += 1 for neighbor in adj[node]: indeg[neighbor] -= 1 if indeg[neighbor] == 0: q.append(neighbor) # 如果排序的课程数等于总课程数，说明无环 return count == numCourses # 测试用例 print(canFinish(2, [[1,0]])) # True: 0->1，可以学完 print(canFinish(2, [[1,0], [0,1]])) # False: 0<->1形成环，无法学完

关键点：这里我们并不需要保存完整的拓扑序列，只需要计数。因为题目只关心能否完成，而不关心具体顺序。

3.2 LeetCode 210. 课程表 II

问题描述：在上一题的基础上，要求返回你为了学完所有课程所安排的学习顺序。如果不可能完成所有课程，返回一个空数组。

本质：在判断是否有环的同时，输出一个可能的拓扑排序序列。

思路：几乎和上一题一样，只是我们需要在从队列中取出节点时，将其记录到结果列表中。

def findOrder(numCourses, prerequisites): from collections import defaultdict, deque adj = defaultdict(list) indeg = [0] * numCourses for course, pre in prerequisites: adj[pre].append(course) indeg[course] += 1 q = deque([i for i in range(numCourses) if indeg[i] == 0]) order = [] while q: node = q.popleft() order.append(node) for neighbor in adj[node]: indeg[neighbor] -= 1 if indeg[neighbor] == 0: q.append(neighbor) # 如果顺序长度等于课程总数，返回顺序；否则返回空数组 return order if len(order) == numCourses else [] # 测试用例 print(findOrder(4, [[1,0], [2,0], [3,1], [3,2]])) # 可能输出: [0, 1, 2, 3] 或 [0, 2, 1, 3] print(findOrder(2, [[0,1], [1,0]])) # 输出: []

这两道题是理解拓扑排序应用的绝佳起点。它们清晰地展示了从问题建模（构建有向图）到算法选择（Kahn算法/BFS），再到代码实现的完整链路。

4. 不止于排序：拓扑排序的进阶应用与变体

拓扑排序不仅仅能给出一个顺序，它作为遍历DAG的一种框架，可以融合其他算法思想，解决更复杂的问题。下面我们看两个常见的变体。

4.1 结合动态规划：求解DAG中的最长路径

在很多任务调度场景中，我们不仅需要顺序，还需要知道完成所有任务的最短或最长时间。如果每条边都有权重（例如，任务耗时），那么在DAG中，求从一个起点到所有其他点的最长路径是一个经典问题。这可以通过在拓扑排序的过程中进行动态规划来实现。

思路：设dp[node]表示从任意入度为0的起点（或多个）到达node节点的最长路径长度。在进行拓扑排序时，当我们处理完一个节点u并准备更新其后继节点v时，我们可以松弛v的最长路径值：dp[v] = max(dp[v], dp[u] + weight(u, v))。由于拓扑序保证了在处理v时，所有可能的前驱节点u都已经被处理过，所以这个DP是正确且高效的。

假设我们有带权重的课程依赖图，权重代表学习该先修课所需的时间，我们想估算学完每门课的最早完成时间。

def longest_path_in_dag(numCourses, prerequisites_with_weight): """ prerequisites_with_weight: List[(course, prereq, weight)] 返回一个列表，表示到达每个课程节点的最长路径长度（时间）。 """ from collections import defaultdict, deque adj = defaultdict(list) # 存储 (neighbor, weight) indeg = [0] * numCourses dp = [0] * numCourses # dp[i] 表示到达课程i的最长路径长度 for course, pre, w in prerequisites_with_weight: adj[pre].append((course, w)) indeg[course] += 1 q = deque([i for i in range(numCourses) if indeg[i] == 0]) while q: u = q.popleft() for v, w in adj[u]: # 动态规划状态转移 dp[v] = max(dp[v], dp[u] + w) indeg[v] -= 1 if indeg[v] == 0: q.append(v) # 检查是否所有节点都被处理（无环） if sum(indeg) > 0: raise ValueError("图中存在环，无法计算最长路径。") return dp # 示例：课程0和2无先修，耗时分别为5和3。课程1依赖0，耗时2。课程3依赖1和2，从1来耗时4，从2来耗时1。 # 边: (1,0,2), (3,1,4), (3,2,1) result = longest_path_in_dag(4, [(1,0,2), (3,1,4), (3,2,1)]) print("到达每门课的最早完成时间:", result) # 解释： # dp[0]=0, dp[2]=0 (起点) # dp[1] = max(dp[1], dp[0]+2) = 2 # dp[3] 有两个前驱： # 从1来: dp[1]+4 = 6 # 从2来: dp[2]+1 = 1 # 取最大值，dp[3] = 6 # 输出: [0, 2, 0, 6]

4.2 检测环与处理非法输入

在实际系统中，依赖关系可能由用户输入或外部配置定义，存在出现环的风险。一个健壮的拓扑排序实现必须能检测并报告环的存在。我们的Kahn算法天然具备这个功能——如果算法结束后，拓扑序列中的节点数少于总节点数，就说明有环，并且剩下的节点都是环的一部分或者依赖于环。

有时候，我们不仅想知道有没有环，还想知道环具体在哪里，以便给出更友好的错误提示。这可以在Kahn算法的基础上稍作修改，最后那些入度不为0的节点，就是与环相关的节点。

def find_cycle_nodes(numCourses, prerequisites): from collections import defaultdict, deque adj = defaultdict(list) indeg = [0] * numCourses for course, pre in prerequisites: adj[pre].append(course) indeg[course] += 1 q = deque([i for i in range(numCourses) if indeg[i] == 0]) topo_order = [] while q: u = q.popleft() topo_order.append(u) for v in adj[u]: indeg[v] -= 1 if indeg[v] == 0: q.append(v) if len(topo_order) == numCourses: print("图中无环。") return [] else: # 找到所有入度仍大于0的节点，它们位于环中或受环影响 cycle_related_nodes = [i for i in range(numCourses) if indeg[i] > 0] print(f"检测到环，相关节点有: {cycle_related_nodes}") return cycle_related_nodes # 测试有环的情况: 0->1, 1->2, 2->0 find_cycle_nodes(3, [[1,0], [2,1], [0,2]])

5. 从理论到工程：实际项目中的考量与优化

在真实的软件开发中，应用拓扑排序远比解算法题复杂。我们需要考虑性能、扩展性、错误处理以及如何与现有系统集成。

1. 图的存储选择： 对于大规模稀疏图，邻接表是首选，空间复杂度O(V+E)。如果节点数量非常大且需要频繁查询两个节点间是否有边，可以考虑使用邻接矩阵，但空间开销为O(V²)。在Python中，defaultdict(list)或listoflist是常见的邻接表实现方式。对于超大规模图，可能需要使用数据库或分布式图存储。

2. 增量更新与动态图： 很多系统的依赖关系不是一成不变的。例如，一个持续集成系统，任务依赖会随着代码提交而改变。这时，每次发生变更都重新进行全图的拓扑排序成本太高。我们可以考虑增量算法：

• 正向影响：当添加一条边A->B时，只有B及其后继节点的拓扑顺序可能需要后移。
• 反向影响：当删除一条边时，情况更复杂一些。 维护每个节点的“优先级”或“等级”，并在更新时局部调整，是处理动态DAG的常见思路。不过，实现一个完全正确且高效的动态拓扑排序算法颇具挑战性，很多时候，如果更新不频繁，重新计算全图排序可能是更简单可靠的选择。

3. 并行化潜力： 拓扑排序的BFS过程在每一轮中，所有入度为0的节点是相互独立的，可以并行处理。这在处理海量任务调度时非常有价值。我们可以用一个线程池或分布式工作队列来并发执行这些可运行的任务。

# 伪代码示意并行处理思想 def parallel_topo_sort(tasks, dependencies): # 初始化入度表和已完成任务集合 indeg = compute_indegree(tasks, dependencies) completed = set() result_order = [] while len(completed) < len(tasks): # 找出当前所有入度为0且未完成的任务 ready_tasks = [t for t in tasks if indeg[t] == 0 and t not in completed] if not ready_tasks: raise CycleError("存在循环依赖") # 并行执行这些ready_tasks (例如使用线程池) with ThreadPoolExecutor() as executor: futures = {executor.submit(execute_task, task): task for task in ready_tasks} for future in as_completed(futures): task = futures[future] try: future.result() # 等待任务执行完成 except Exception as e: handle_error(task, e) finally: # 任务完成，更新状态 completed.add(task) result_order.append(task) # 更新后继节点的入度 for successor in get_successors(task, dependencies): indeg[successor] -= 1 return result_order

4. 处理“等价类”或“强连通分量”： 有时，一组任务相互依赖，形成一个强连通分量（SCC），在DAG中它们应该被压缩成一个“超级节点”。这就需要先使用Tarjan或Kosaraju算法找出所有SCC，将每个SCC缩点，得到一个新的DAG，再对这个DAG进行拓扑排序。这在分析复杂系统模块依赖时非常有用。

5. 可视化与调试： 对于复杂的依赖关系，一张图胜过千言万语。在开发调试阶段，可以将构建的DAG和计算出的拓扑序用Graphviz等工具生成图片，直观地验证算法的正确性，并向非技术人员解释调度逻辑。

# 使用graphviz简单可视化（需要安装graphviz库） from graphviz import Digraph def visualize_dag(numCourses, prerequisites, order=None): dot = Digraph(comment='Course DAG') for i in range(numCourses): dot.node(str(i), f'Course {i}') for course, pre in prerequisites: dot.edge(str(pre), str(course)) if order: # 可以用颜色或顺序标记拓扑序 pass dot.render('course_dag.gv', view=True) # 生成并打开PDF

拓扑排序是一个原理简单但应用深广的算法。从LeetCode上的几行代码，到支撑起整个分布式任务调度引擎，其核心思想始终如一：理清依赖，逐步推进。下次当你面对任何带有“先后”关系的问题时，不妨先画个图，想想它是不是个DAG，也许拓扑排序就是那把隐藏的钥匙。我在处理一个微服务启动顺序的问题时，就是靠它避免了服务间因依赖未就绪而启动失败的混乱局面。记住，判断环的存在永远是第一步，一个清晰的依赖图是后续一切操作的基础。