本文目录导读：
我来详细讲解如何使用NetworkX进行拓扑排序,包含多个实际案例。
基础拓扑排序
import networkx as nx
import matplotlib.pyplot as plt
# 创建一个有向无环图(DAG)
G = nx.DiGraph()
# 添加边（任务依赖关系）
edges = [
('A', 'B'), # A → B：A完成后才能开始B
('A', 'C'), # A → C：A完成后才能开始C
('B', 'D'), # B → D：B完成后才能开始D
('C', 'D'), # C → D：C完成后才能开始D
('D', 'E') # D → E：D完成后才能开始E
]
G.add_edges_from(edges)
# 执行拓扑排序
try:
topological_order = list(nx.topological_sort(G))
print("拓扑排序结果:", topological_order)
# 输出: ['A', 'B', 'C', 'D', 'E'] 或 ['A', 'C', 'B', 'D', 'E']
except nx.NetworkXUnfeasible:
print("图中存在环，无法进行拓扑排序")
课程安排系统案例
import networkx as nx
class CourseScheduler:
def __init__(self):
self.graph = nx.DiGraph()
def add_course_prerequisites(self, course, prerequisites):
"""添加课程及其先修课程"""
for prereq in prerequisites:
self.graph.add_edge(prereq, course) # 先修课程 → 当前课程
def get_study_order(self):
"""获取学习顺序"""
try:
schedule = list(nx.topological_sort(self.graph))
return schedule
except nx.NetworkXUnfeasible:
print("课程依赖关系存在循环，无法安排学习顺序")
return []
def check_cycle(self):
"""检查是否有循环依赖"""
try:
nx.find_cycle(self.graph)
return True
except nx.NetworkXNoCycle:
return False
def visualize_schedule(self):
"""可视化课程依赖关系"""
pos = nx.spring_layout(self.graph)
nx.draw(self.graph, pos, with_labels=True,
node_color='lightblue',
node_size=2000,
arrows=True,
arrowstyle='->')
plt.title("课程依赖关系图")
plt.show()
# 使用案例
scheduler = CourseScheduler()
# 添加课程依赖
scheduler.add_course_prerequisites("Algorithms", ["Data Structures", "Mathematics"])
scheduler.add_course_prerequisites("Data Structures", ["Programming Basics"])
scheduler.add_course_prerequisites("Machine Learning", ["Algorithms", "Statistics"])
scheduler.add_course_prerequisites("Statistics", ["Mathematics"])
scheduler.add_course_prerequisites("Mathematics", ["Programming Basics"])
# 获取学习顺序
study_order = scheduler.get_study_order()
print("推荐学习顺序:", study_order)
# 输出类似: ['Programming Basics', 'Mathematics', 'Data Structures', 'Statistics', 'Algorithms', 'Machine Learning']
项目管理依赖排序
import networkx as nx
from datetime import datetime, timedelta
class ProjectTaskScheduler:
def __init__(self):
self.graph = nx.DiGraph()
self.task_durations = {}
def add_task(self, task_name, duration_days, dependencies=None):
"""添加任务及其依赖"""
self.task_durations[task_name] = duration_days
if dependencies:
for dep in dependencies:
self.graph.add_edge(dep, task_name)
else:
self.graph.add_node(task_name)
def get_task_order(self):
"""获取任务执行顺序"""
try:
return list(nx.topological_sort(self.graph))
except nx.NetworkXUnfeasible:
print("任务依赖存在循环")
return []
def get_critical_path(self):
"""获取关键路径（最长路径）"""
try:
# 找到所有拓扑排序
task_order = list(nx.topological_sort(self.graph))
# 计算每个任务的最早开始时间
earliest_start = {task: 0 for task in task_order}
for task in task_order:
for pred in self.graph.predecessors(task):
earliest_start[task] = max(
earliest_start[task],
earliest_start[pred] + self.task_durations[pred]
)
# 计算每个任务的最晚开始时间
latest_start = {task: float('inf') for task in task_order}
for task in reversed(task_order):
if self.graph.out_degree(task) == 0:
latest_start[task] = earliest_start[task]
else:
for succ in self.graph.successors(task):
latest_start[task] = min(
latest_start[task],
latest_start[succ] - self.task_durations[task]
)
# 找出关键路径上的任务（最早开始时间等于最晚开始时间）
critical_tasks = [task for task in task_order
if earliest_start[task] == latest_start[task]]
return critical_tasks, earliest_start, latest_start
except Exception as e:
print(f"计算关键路径时出错: {e}")
return [], {}, {}
def get_project_duration(self):
"""获取项目总持续时间"""
critical_tasks, _, _ = self.get_critical_path()
return sum(self.task_durations[task] for task in critical_tasks)
# 使用案例
project = ProjectTaskScheduler()
# 添加项目任务
project.add_task("需求分析", 5)
project.add_task("系统设计", 3, ["需求分析"])
project.add_task("数据库设计", 2, ["系统设计"])
project.add_task("前端开发", 8, ["系统设计"])
project.add_task("后端开发", 10, ["系统设计", "数据库设计"])
project.add_task("测试", 4, ["前端开发", "后端开发"])
project.add_task("部署", 1, ["测试"])
project.add_task("用户培训", 2, ["部署"])
# 获取任务顺序
task_order = project.get_task_order()
print("任务执行顺序:", task_order)
# 获取关键路径
critical_tasks, earliest, latest = project.get_critical_path()
print("\n关键路径上的任务:", critical_tasks)
print("项目总工期:", project.get_project_duration(), "天")
# 显示任务时间安排
print("\n任务时间安排:")
for task in task_order:
print(f"{task}: 最早开始={earliest[task]}天, 最晚开始={latest[task]}天, "
f"工期={project.task_durations[task]}天")
并行任务拓扑排序
import networkx as nx
from collections import defaultdict
class ParallelTaskScheduler:
def __init__(self):
self.graph = nx.DiGraph()
def add_dependency(self, from_task, to_task):
"""添加任务依赖"""
self.graph.add_edge(from_task, to_task)
def get_parallel_levels(self):
"""获取可以并行执行的任务层级"""
try:
topological_order = list(nx.topological_sort(self.graph))
# 计算每个任务的层级
levels = {}
for task in topological_order:
predecessors = list(self.graph.predecessors(task))
if not predecessors:
levels[task] = 0
else:
levels[task] = max(levels[pred] for pred in predecessors) + 1
# 按层级分组
level_groups = defaultdict(list)
for task, level in levels.items():
level_groups[level].append(task)
return dict(level_groups)
except nx.NetworkXUnfeasible:
print("存在环，无法并行调度")
return {}
def visualize_parallel_tasks(self):
"""可视化并行任务"""
levels = self.get_parallel_levels()
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 6))
# 子图1：显示DAG
pos = nx.spring_layout(self.graph)
nx.draw(self.graph, pos, with_labels=True,
node_color='lightgreen',
node_size=2000,
arrows=True,
ax=ax1)
ax1.set_title("任务依赖图")
# 子图2：显示层级
for level, tasks in levels.items():
y_pos = [level] * len(tasks)
x_pos = range(len(tasks))
ax2.scatter(x_pos, y_pos, s=2000, alpha=0.6)
for i, task in enumerate(tasks):
ax2.annotate(task, (i, level), ha='center', va='center')
ax2.set_xlabel("并行任务")
ax2.set_ylabel("执行层级")
ax2.set_title("并行任务层级")
ax2.grid(True)
plt.tight_layout()
plt.show()
# 使用案例
parallel_scheduler = ParallelTaskScheduler()
# 添加依赖关系
parallel_scheduler.add_dependency("Task1", "Task4")
parallel_scheduler.add_dependency("Task1", "Task5")
parallel_scheduler.add_dependency("Task2", "Task5")
parallel_scheduler.add_dependency("Task3", "Task6")
parallel_scheduler.add_dependency("Task4", "Task7")
parallel_scheduler.add_dependency("Task5", "Task7")
parallel_scheduler.add_dependency("Task6", "Task8")
# 获取并行层级
parallel_levels = parallel_scheduler.get_parallel_levels()
print("并行执行计划:")
for level, tasks in sorted(parallel_levels.items()):
print(f"层级 {level}: {', '.join(tasks)}")
高级功能：多条件拓扑排序
import networkx as nx
class AdvancedTopologicalScheduler:
def __init__(self):
self.graph = nx.DiGraph()
self.priorities = {} # 任务优先级
def add_task(self, task, priority=0, dependencies=None):
"""添加带优先级的任务"""
self.priorities[task] = priority
if dependencies:
for dep in dependencies:
self.graph.add_edge(dep, task)
else:
self.graph.add_node(task)
def topological_sort_with_priority(self):
"""按优先级排序的拓扑排序"""
try:
# 获取所有拓扑排序（生成器）
all_topsorts = nx.all_topological_sorts(self.graph)
# 根据优先级评分选择最优排序
best_score = -1
best_order = None
for topsort in all_topsorts:
# 计算当前排序的评分
score = self._calculate_order_score(topsort)
if score > best_score:
best_score = score
best_order = topsort
return best_order, best_score
except nx.NetworkXUnfeasible:
print("存在环")
return [], 0
def _calculate_order_score(self, order):
"""计算排序评分"""
score = 0
for i, task in enumerate(order):
# 优先级越高的任务越早完成得分越高
priority = self.priorities.get(task, 0)
score += priority * (len(order) - i)
return score
def get_possible_orders(self, max_results=5):
"""获取多种可能的排序"""
try:
all_topsorts = list(nx.all_topological_sorts(self.graph))
return all_topsorts[:max_results]
except nx.NetworkXUnfeasible:
return []
# 使用案例
advanced_scheduler = AdvancedTopologicalScheduler()
# 添加带优先级的任务
advanced_scheduler.add_task("紧急Bug修复", priority=10)
advanced_scheduler.add_task("功能开发A", priority=5, dependencies=["数据库优化"])
advanced_scheduler.add_task("功能开发B", priority=3, dependencies=["功能开发A"])
advanced_scheduler.add_task("数据库优化", priority=8)
advanced_scheduler.add_task("代码审查", priority=6, dependencies=["功能开发A", "功能开发B"])
advanced_scheduler.add_task("测试", priority=4, dependencies=["代码审查"])
advanced_scheduler.add_task("文档更新", priority=1, dependencies=["功能开发A"])
# 获取最优排序
best_order, score = advanced_scheduler.topological_sort_with_priority()
print(f"最优排序 (评分: {score}): {best_order}")
# 获取所有可能的排序
print("\n所有可能的排序:")
for i, order in enumerate(advanced_scheduler.get_possible_orders(3), 1):
print(f"方案{i}: {order}")
实用工具函数
def check_dag_health(graph):
"""检查DAG图健康状况"""
results = {
'is_dag': nx.is_directed_acyclic_graph(graph),
'node_count': graph.number_of_nodes(),
'edge_count': graph.number_of_edges(),
'has_cycle': False,
'roots': [],
'leaves': []
}
if not results['is_dag']:
results['has_cycle'] = True
try:
cycle = nx.find_cycle(graph)
results['cycle'] = cycle
except nx.NetworkXNoCycle:
pass
else:
# 找出根节点（没有入边的节点）
results['roots'] = [n for n in graph.nodes() if graph.in_degree(n) == 0]
# 找出叶子节点（没有出边的节点）
results['leaves'] = [n for n in graph.nodes() if graph.out_degree(n) == 0]
return results
def layered_topological_sort(graph):
"""分层拓扑排序（用于并行执行）"""
if not nx.is_directed_acyclic_graph(graph):
return None
# 复制图
G = graph.copy()
layers = []
while G.nodes():
# 找出当前层（入度为0的节点）
current_layer = [n for n in G.nodes() if G.in_degree(n) == 0]
if not current_layer:
break
layers.append(current_layer)
# 移除当前层的节点
G.remove_nodes_from(current_layer)
return layers
# 完整的综合示例
def main():
print("=== 拓扑排序综合示例 ===\n")
# 1. 创建任务依赖图
G = nx.DiGraph()
tasks = [
('A', 'B'),
('A', 'C'),
('B', 'D'),
('C', 'D'),
('C', 'E'),
('D', 'F'),
('E', 'F'),
]
G.add_edges_from(tasks)
# 2. 检查DAG健康
health = check_dag_health(G)
print(f"图状态: {'正常' if health['is_dag'] else '有环'}")
print(f"节点数: {health['node_count']}")
print(f"边数: {health['edge_count']}")
print(f"根节点: {health['roots']}")
print(f"叶子节点: {health['leaves']}")
# 3. 标准拓扑排序
if health['is_dag']:
standard_order = list(nx.topological_sort(G))
print(f"\n标准拓扑排序: {standard_order}")
# 4. 分层排序
layers = layered_topological_sort(G)
print("\n分层排序 (并行层级):")
for i, layer in enumerate(layers):
print(f" 层级 {i+1}: {layer}")
# 5. 可视化
pos = nx.spring_layout(G)
plt.figure(figsize=(10, 6))
# 为不同层级着色
colors = ['red', 'blue', 'green', 'orange', 'purple']
node_colors = []
node_to_layer = {}
for i, layer in enumerate(layers):
for node in layer:
node_to_layer[node] = i
for node in G.nodes():
if node in node_to_layer:
node_colors.append(colors[node_to_layer[node] % len(colors)])
else:
node_colors.append('gray')
nx.draw(G, pos, with_labels=True,
node_color=node_colors,
node_size=2000,
arrows=True,
font_size=12,
font_weight='bold')
plt.title("任务依赖图 (颜色表示执行层级)")
plt.show()
if __name__ == "__main__":
main()
关键要点
- 核心函数:
nx.topological_sort()是主要函数 - 错误处理: 始终使用 try-except 处理环检测
- 并行执行: 可以使用分层排序识别可并行任务
- 优先级: 可以通过自定义评分函数实现优先级排序
- 可视化: NetworkX 提供了丰富的可视化选项
这些案例覆盖了拓扑排序的常见应用场景,你可以根据具体需求选择合适的实现方式。
