3天搞定分类图手写实现，拒绝文档焦虑

3天搞定分类图手写实现，拒绝文档焦虑 官方文档太长，翻两页就忘，根本抓不住重点。与其对着枯燥的 API 列表发呆，不如直接上手，用 20 行代码跑通一个极简分类图原型。这里不堆砌术语，我们直接切入核心，通过手写实现的方式，把“分类图”这个听起来很高大上的数据结构，拆解成你能看懂的 Python 代码。 项目目标：我们要造一个什么样的轮子 很多开发者听到“分类图”或者“有向无环图（DAG）”就头大，觉得这是编译器或者复杂调度系统才需要关心的东西。其实不然。在数据清洗、任务依赖管理、甚至前端组件树渲染中，分类图的思想无处不在。 我们的目标很明确：从零搭建一个轻量级的分类图工具。它不需要支持百万级节点，不需要复杂的并发锁机制，但必须具备以下三个核心能力：节点增删：能动态添加任务节点和依赖关系。 拓扑排序：能输出合法的执行顺序，这是分类图最核心的价值。 环检测：如果依赖关系形成了死循环，必须能准确报错，而不是让程序卡死。为什么强调手写实现？因为库（如 networkx）虽然强大，但当你需要嵌入到特定业务逻辑中，或者面试被问到“请简述拓扑排序的底层原理”时，依赖库的黑盒会让你哑口无言。自己写一遍，才能把内存占用、时间复杂度这些指标刻在脑子里。 目录结构：极简主义，拒绝过度设计 作为一个实战项目，我们要保持工程化的整洁，但不搞形式主义。整个项目只需要三个文件，放在同一个文件夹下即可运行。 category_graph/ ├── graph_core.py # 核心算法实现：节点、边、拓扑排序 ├── demo.py # 演示脚本：构建具体业务场景 └── tests.py # 单元测试：验证边界情况这种结构的好处是，你可以随时复制 graph_core.py 到任何项目中复用，而不需要安装任何第三方依赖。这就是纯 Python 标准库的威力。 核心代码实现：逐行拆解拓扑排序 分类图的核心在于拓扑排序。通俗点说，就是“先完成前置任务，再执行后续任务”。最常用的算法是 Kahn 算法，它基于 BFS（广度优先搜索），利用入度（In-degree）来判断节点是否可以被处理。 下面是 graph_core.py 的完整代码。我会把关键逻辑拆解开，告诉你每一行代码背后的意图。 import collections from typing import List, Dict, Set, Optionalclass CategoryGraph:一个基于有向无环图（DAG）的分类图实现用于管理任务依赖和执行顺序def __init__(self):# 邻接表：存储每个节点指向哪些后继节点# 例如：A - [B, C] 表示 A 完成后，B 和 C 可以开始self.graph: Dict[str, List[str]] = {}# 入度表：记录每个节点有多少个前驱节点# 入度为 0 的节点，就是可以立即执行的节点self.in_degree: Dict[str, int] = {}def add_node(self, node: str):添加单个节点，初始化入度为0if node not in self.graph:self.graph[node] = []self.in_degree[node] = 0def add_edge(self, source: str, target: str):添加依赖关系：source - target意味着 target 依赖于 source，source 必须先执行# 确保源节点和目标节点都已初始化self.add_node(source)self.add_node(target)# 检查是否已存在这条边，防止重复添加导致入度错误if target not in self.graph[source]:self.graph[source].append(target)self.in_degree[target] += 1# 如果形成了环，这里暂时不检测，留到拓扑排序时处理def topological_sort(self) - Optional[List[str]]:执行拓扑排序返回：合法的任务执行顺序列表，如果存在环则返回 None# 1. 找出所有入度为 0 的节点，放入队列queue = collections.deque()for node, degree in self.in_degree.items():if degree == 0:queue.append(node)result = []processed_count = 0# 2. BFS 遍历过程while queue:current = queue.popleft()result.append(current)processed_count += 1# 遍历当前节点的所有后继节点for neighbor in self.graph[current]:# 后继节点的入度减 1，因为前驱节点已经处理完毕self.in_degree[neighbor] -= 1# 如果后继节点入度变为 0，说明它的所有前置依赖都满足了if self.in_degree[neighbor] == 0:queue.append(neighbor)# 3. 判断是否存在环# 如果处理的节点数少于总节点数，说明有节点永远无法入队（入度不为0），即存在环if processed_count len(self.in_degree):return Nonereturn result代码深度解析数据结构选择： 我们使用了两个字典：self.graph 和 self.in_degree。self.graph 是邻接表，Dict[str, List[str]]。为什么不用列表？因为节点名称可能是字符串，用字典查找邻居是 O(1) 的，而列表遍历是 O(N)。在大规模图中，这点差异会被放大。 self.in_degree 记录依赖数。这是 Kahn 算法的灵魂。只有入度为 0，节点才是“自由”的，才能被调度。为什么用 collections.deque？ 在 Python 中，list.pop(0) 的时间复杂度是 O(N)，因为它需要移动所有后续元素。而 deque.popleft() 是 O(1)。在拓扑排序中，队列操作频繁，使用 deque 是性能优化的关键细节。这一点在 MDN Web Docs 关于 JavaScript 数据结构的文章中也有提及，虽然语言不同，但底层逻辑在高性能计算中是通用的。环检测逻辑： 代码最后有一个判断：if processed_count len(self.in_degree)。 想象一下，如果有 A-B, B-C, C-A 这样的循环。A 的入度是 1（来自 C），B 是 1（来自 A），C 是 1（来自 B）。初始队列是空的！程序直接结束，processed_count 为 0，小于总节点数 3，于是返回 None。这就是最简单的环检测，不需要额外的 DFS 标记栈。运行与测试：用业务场景验证逻辑 光有算法不够，得跑起来。我们在 demo.py 中模拟一个“网站部署流水线”的场景。 场景描述：install_deps：安装依赖（无依赖） lint_code：代码检查（依赖 install_deps） unit_test：单元测试（依赖 install_deps） build_docker：构建镜像（依赖 lint_code 和 unit_test） deploy_prod：生产部署（依赖 build_docker）from graph_core import CategoryGraphdef main():print(=== 开始构建部署流水线分类图 ===)g = CategoryGraph()# 添加节点和依赖关系g.add_edge(install_deps, lint_code)g.add_edge(install_deps, unit_test)g.add_edge(lint_code, build_docker)g.add_edge(unit_test, build_docker)g.add_edge(build_docker, deploy_prod)# 执行拓扑排序order = g.topological_sort()if order:print(合法的执行顺序:)for i, step in enumerate(order, 1):print(f {i}. {step})else:print(错误：检测到循环依赖！)print(\n=== 测试循环依赖 ===)g2 = CategoryGraph()g2.add_edge(A, B)g2.add_edge(B, C)g2.add_edge(C, A) # 形成环 A-B-C-Aorder2 = g2.topological_sort()print(f检测结果: {order2}) # 应该输出 Noneif __name__ == __main__:main()运行结果： === 开始构建部署流水线分类图 === 合法的执行顺序:1. install_deps2. lint_code3. unit_test4. build_docker5. deploy_prod=== 测试循环依赖 === 检测结果: None注意看输出，lint_code 和 unit_test 的顺序可能互换，这取决于字典的遍历顺序。在 Python 3.7+ 中，字典是有序的，但在逻辑上，这两个任务是可以并行的。如果业务要求严格串行，你需要在应用层加锁；如果允许并行，这个顺序就是完美的。 优化扩展：从玩具到生产级 上面的代码能跑，但离“生产级”还有距离。以下是几个在实际项目中必须考虑的优化点：并行执行支持： 当前的 topological_sort 返回的是一个线性列表。但在实际部署中，lint_code 和 unit_test 可以同时进行。 改进方案：修改算法，返回“层级列表”（List of List）。每一层的节点可以并行执行。 # 伪代码思路 levels = [] current_level = [node for node in in_degree if in_degree[node] == 0] while current_level:levels.append(current_level)next_level = []for node in current_level:for neighbor in graph[node]:in_degree[neighbor] -= 1if in_degree[neighbor] == 0:next_level.append(neighbor)current_level = next_level内存优化： 如果节点数量达到百万级，Dict[str, List[str]] 的开销会很大。 改进方案：将节点名称映射为整数 ID。使用 List[List[int]] 存储邻接表。整数的内存占用远小于字符串，且 CPU 缓存友好度更高。持久化存储： 分类图结构经常需要保存和加载。 改进方案：实现 to_dict 和 from_dict 方法，将图结构序列化为 JSON。 def to_dict(self):return {graph: self.graph,in_degree: self.in_degree}异常处理增强： 当前 add_edge 没有检查 source 或 target 是否为空。在生产环境中，输入验证是防止脏数据进入系统的最后一道防线。建议添加 assert source and target。小结 通过这篇实战，我们完成了一个手写实现的分类图工具。核心收获：你不再被“拓扑排序”这个词吓倒，你知道了它其实就是“不断挑出没有依赖的任务执行”。 关键技巧：使用 in_degree 数组追踪依赖状态，使用 deque 保证队列操作效率。 避坑指南：一定要处理环检测，否则程序会静默失败或死循环。分类图不仅仅是一个算法题，它是解决依赖管理问题的通用范式。无论是 CI/CD 流水线、大数据任务调度，还是前端微前端的加载顺序，背后都是这套逻辑。 现在，你手里有了这个轮子。你可以把它扔进你的下一个项目里，或者在此基础上扩展成支持并行调度的任务管理器。 还有什么不懂的？评论区留言挨个回。