Dijkstra算法实战：用Python手把手教你实现最短路径规划（附完整代码）
# 从零到一：用Python彻底掌握Dijkstra最短路径算法
如果你曾经用过手机地图导航，或者玩过需要寻找最优路线的策略游戏，那你其实已经在不知不觉中接触过我们今天要深入探讨的核心技术了。想象一下，你站在一个复杂的交通网络中心，面前有无数条道路通向不同的目的地，每条路都有不同的通行时间。如何快速找到到达每个地方的最快路线？这就是Dijkstra算法要解决的经典问题。
我第一次接触这个算法是在大学的数据结构课上，当时觉得它既神奇又有些抽象。直到后来参与一个物流配送系统的项目，需要为数百个配送点规划最优路线时，才真正体会到这个算法的强大之处。那天晚上，我盯着满屏的节点和边，突然意识到：这不就是现实世界中的路径规划问题吗？从那一刻起，Dijkstra算法从一个课堂概念变成了我工具箱中不可或缺的利器。
## 1. 最短路径问题的现实意义与数学基础
在我们深入代码之前，先来理解一下这个算法到底在解决什么问题。**单源最短路径**这个术语听起来可能有些学术化，但它的应用场景却无处不在。
### 1.1 从生活场景到数学模型
考虑这样一个实际场景：你所在的城市有一个快递配送中心，每天需要向全市50个网点送货。每个网点之间的行驶时间已知，如何为每辆配送车规划最高效的路线？这就是典型的最短路径问题。
从数学角度，我们可以用**加权有向图**来建模这个问题：
- **顶点（Vertex）**：代表各个配送网点
- **边（Edge）**：代表网点之间的道路
- **权重（Weight）**：代表行驶时间或距离
用数学符号表示就是：设图G=(V,E)，其中V是顶点集合，E是边集合，每条边e∈E都有一个权重w(e)。对于给定的源点s∈V，我们需要找到从s到每个顶点v∈V的最短路径长度。
### 1.2 Dijkstra算法的核心思想
Edsger W. Dijkstra在1956年提出这个算法时，采用了一种**贪心策略**。算法的基本思路可以用一个简单的比喻来理解：
> 想象你在一个漆黑的迷宫里，手里只有一支手电筒。你从起点开始，每次只照亮离你最近的那个出口，然后从那里继续探索。虽然你不能一眼看到整个迷宫，但通过这种"最近优先"的策略，你最终能找到到达每个地方的最短路径。
算法的正式描述如下：
1. **初始化**：设置起点距离为0，其他所有顶点距离为无穷大
2. **选择**：从未处理的顶点中选择距离起点最近的顶点
3. **松弛**：更新该顶点所有邻居的距离
4. **标记**：将该顶点标记为已处理
5. **重复**：重复步骤2-4，直到所有顶点都被处理
这个过程中最关键的洞察是：**一旦一个顶点被标记为已处理，从起点到该顶点的最短距离就确定了**。这是因为所有边的权重都是非负的，所以不可能通过其他未处理顶点找到更短的路径。
## 2. 基础实现：从理论到代码的第一次飞跃
理解了算法思想后，让我们用Python来实现最基础的版本。这个版本虽然效率不高，但能清晰地展示算法的每一步。
### 2.1 图的表示方法
在编程中，我们首先需要决定如何表示图。对于Dijkstra算法，最常用的有两种表示方法：
```python
# 邻接矩阵表示法
graph_matrix = [
[0, 4, 0, 0, 0, 0, 0, 8, 0],
[4, 0, 8, 0, 0, 0, 0, 11, 0],
[0, 8, 0, 7, 0, 4, 0, 0, 2],
[0, 0, 7, 0, 9, 14, 0, 0, 0],
[0, 0, 0, 9, 0, 10, 0, 0, 0],
[0, 0, 4, 14, 10, 0, 2, 0, 0],
[0, 0, 0, 0, 0, 2, 0, 1, 6],
[8, 11, 0, 0, 0, 0, 1, 0, 7],
[0, 0, 2, 0, 0, 0, 6, 7, 0]
]
# 邻接表表示法（更节省空间）
graph_list = {
0: {1: 4, 7: 8},
1: {0: 4, 2: 8, 7: 11},
2: {1: 8, 3: 7, 5: 4, 8: 2},
3: {2: 7, 4: 9, 5: 14},
4: {3: 9, 5: 10},
5: {2: 4, 3: 14, 4: 10, 6: 2},
6: {5: 2, 7: 1, 8: 6},
7: {0: 8, 1: 11, 6: 1, 8: 7},
8: {2: 2, 6: 6, 7: 7}
}
```
> **选择建议**：对于稀疏图（边数远小于顶点数的平方），使用邻接表更节省内存；对于稠密图，邻接矩阵可能更合适。在实际项目中，我通常根据数据特点来选择，大多数情况下邻接表是更好的选择。
### 2.2 基础实现的完整代码
下面是Dijkstra算法的基础实现，我添加了详细的注释和调试输出，方便你理解每一步的执行过程：
```python
def dijkstra_basic(graph, start):
"""
Dijkstra算法的基础实现
时间复杂度：O(V²)，适合小规模图
参数：
graph: 邻接表表示的图，格式为 {顶点: {邻居: 权重}}
start: 起始顶点
返回：
distances: 从起点到各顶点的最短距离
previous: 最短路径中每个顶点的前驱顶点
"""
# 初始化距离字典，所有顶点距离设为无穷大
distances = {vertex: float('inf') for vertex in graph}
distances[start] = 0
# 记录每个顶点的前驱顶点，用于重建路径
previous = {vertex: None for vertex in graph}
# 未处理的顶点集合
unvisited = set(graph.keys())
# 记录处理顺序，用于调试
processed_order = []
print(f"开始执行Dijkstra算法，起点: {start}")
print(f"初始距离: {distances}")
print("-" * 50)
while unvisited:
# 步骤1：选择未处理顶点中距离最小的顶点
current = min(unvisited, key=lambda vertex: distances[vertex])
# 如果最小距离是无穷大，说明剩下的顶点不可达
if distances[current] == float('inf'):
print(f"顶点 {current} 不可达，停止算法")
break
# 从未处理集合中移除当前顶点
unvisited.remove(current)
processed_order.append(current)
print(f"处理顶点: {current}, 当前距离: {distances[current]}")
print(f"未处理顶点: {unvisited}")
# 步骤2：松弛操作 - 更新邻居的距离
for neighbor, weight in graph[current].items():
if neighbor in unvisited: # 只处理未标记的邻居
new_distance = distances[current] + weight
print(f" 检查边 {current}->{neighbor} (权重: {weight})")
print(f" 当前距离到 {neighbor}: {distances[neighbor]}")
print(f" 新计算距离: {new_distance}")
if new_distance < distances[neighbor]:
distances[neighbor] = new_distance
previous[neighbor] = current
print(f" ✓ 更新 {neighbor} 的距离为 {new_distance}")
else:
print(f" ✗ 保持原距离")
print(f"更新后距离: {distances}")
print("-" * 30)
print(f"\n处理顺序: {processed_order}")
print(f"最终距离: {distances}")
return distances, previous
def reconstruct_path(previous, start, target):
"""根据前驱字典重建从起点到目标点的路径"""
path = []
current = target
while current is not None:
path.append(current)
current = previous[current]
path.reverse()
# 检查路径是否有效（起点到目标点是否可达）
if path[0] == start:
return path
else:
return [] # 不可达
# 测试基础实现
if __name__ == "__main__":
# 使用之前定义的邻接表图
test_graph = graph_list
print("=" * 60)
print("Dijkstra算法基础实现测试")
print("=" * 60)
start_vertex = 0
distances, previous = dijkstra_basic(test_graph, start_vertex)
print("\n" + "=" * 60)
print("最短路径结果汇总")
print("=" * 60)
for vertex in test_graph:
if vertex != start_vertex:
path = reconstruct_path(previous, start_vertex, vertex)
if path:
print(f"从 {start_vertex} 到 {vertex}: 距离={distances[vertex]}, 路径={path}")
else:
print(f"从 {start_vertex} 到 {vertex}: 不可达")
```
运行这段代码，你会看到算法如何一步步"探索"整个图。这种基础实现的时间复杂度是O(V²)，其中V是顶点数。对于小规模图（比如几十个顶点），这个实现完全够用。但在实际应用中，我们经常需要处理成千上万个顶点，这时候就需要优化了。
## 3. 优先队列优化：将复杂度从O(V²)降到O((V+E)logV)
基础实现的主要瓶颈在于每次都要遍历所有未处理顶点来找到距离最小的那个。当顶点数很多时，这个操作会变得非常慢。解决方案是使用**优先队列（Priority Queue）**。
### 3.1 优先队列的工作原理
优先队列是一种特殊的数据结构，它总是能快速返回最小（或最大）的元素。在Python中，我们可以使用`heapq`模块来实现最小堆作为优先队列。
让我用一个实际项目的经验来说明为什么优先队列如此重要。曾经我需要为一个有5000多个交叉口的城市规划紧急救援路线，使用基础实现需要近10分钟才能计算出结果，而使用优先队列优化后，只需要不到1秒。
### 3.2 优化实现的完整代码
```python
import heapq
from typing import Dict, List, Tuple, Optional
def dijkstra_optimized(graph: Dict[int, Dict[int, float]], start: int) -> Tuple[Dict[int, float], Dict[int, Optional[int]]]:
"""
使用优先队列优化的Dijkstra算法
时间复杂度：O((V+E)logV)，适合大规模图
参数：
graph: 邻接表表示的加权有向图
start: 起始顶点
返回：
distances: 从起点到各顶点的最短距离
predecessors: 最短路径中每个顶点的前驱顶点
"""
# 初始化距离和前驱
distances = {vertex: float('inf') for vertex in graph}
predecessors = {vertex: None for vertex in graph}
distances[start] = 0
# 使用优先队列（最小堆）
# 堆中元素为 (距离, 顶点)
priority_queue = [(0, start)]
# 记录顶点是否已确定最短路径
visited = set()
# 性能统计
operations_count = 0
relaxations_count = 0
print(f"开始优化版Dijkstra算法，起点: {start}")
print(f"图大小: {len(graph)}个顶点")
print("-" * 50)
while priority_queue:
operations_count += 1
# 从堆中取出距离最小的顶点
current_distance, current_vertex = heapq.heappop(priority_queue)
# 如果这个顶点已经处理过，跳过
if current_vertex in visited:
continue
# 标记为已访问
visited.add(current_vertex)
if operations_count % 10 == 0: # 每10次操作输出一次进度
print(f"已处理 {len(visited)}/{len(graph)} 个顶点...")
# 遍历所有邻居
for neighbor, weight in graph[current_vertex].items():
relaxations_count += 1
# 计算通过当前顶点到邻居的新距离
new_distance = current_distance + weight
# 如果找到更短的路径，更新距离并加入优先队列
if new_distance < distances[neighbor]:
distances[neighbor] = new_distance
predecessors[neighbor] = current_vertex
heapq.heappush(priority_queue, (new_distance, neighbor))
print(f"\n算法完成统计:")
print(f"总操作次数: {operations_count}")
print(f"松弛操作次数: {relaxations_count}")
print(f"已处理顶点数: {len(visited)}")
return distances, predecessors
def find_shortest_path(predecessors: Dict[int, Optional[int]], start: int, target: int) -> List[int]:
"""重建从起点到目标点的最短路径"""
path = []
current = target
# 从目标点反向追踪到起点
while current is not None:
path.append(current)
current = predecessors[current]
path.reverse()
# 检查路径是否有效
return path if path and path[0] == start else []
def benchmark_dijkstra():
"""对比基础实现和优化实现的性能"""
import time
import random
# 生成测试图
def generate_test_graph(num_vertices: int, edge_probability: float = 0.3) -> Dict[int, Dict[int, float]]:
"""生成随机测试图"""
graph = {i: {} for i in range(num_vertices)}
for i in range(num_vertices):
for j in range(num_vertices):
if i != j and random.random() < edge_probability:
# 随机权重，范围1-100
weight = random.randint(1, 100)
graph[i][j] = weight
return graph
# 测试不同规模的图
test_sizes = [10, 50, 100, 200, 500]
print("性能对比测试")
print("=" * 60)
print(f"{'顶点数':<10} {'基础实现(ms)':<15} {'优化实现(ms)':<15} {'加速比':<10}")
print("-" * 60)
for size in test_sizes:
test_graph = generate_test_graph(size, 0.1) # 10%的边概率
# 测试基础实现
start_time = time.time()
distances1, _ = dijkstra_basic(test_graph, 0)
basic_time = (time.time() - start_time) * 1000
# 测试优化实现
start_time = time.time()
distances2, _ = dijkstra_optimized(test_graph, 0)
optimized_time = (time.time() - start_time) * 1000
# 验证结果一致性
assert distances1 == distances2, "两种实现结果不一致！"
speedup = basic_time / optimized_time if optimized_time > 0 else float('inf')
print(f"{size:<10} {basic_time:<15.2f} {optimized_time:<15.2f} {speedup:<10.1f}x")
# 运行优化算法
if __name__ == "__main__":
# 使用更复杂的测试图
complex_graph = {
0: {1: 4, 7: 8},
1: {0: 4, 2: 8, 7: 11},
2: {1: 8, 3: 7, 5: 4, 8: 2},
3: {2: 7, 4: 9, 5: 14},
4: {3: 9, 5: 10},
5: {2: 4, 3: 14, 4: 10, 6: 2},
6: {5: 2, 7: 1, 8: 6},
7: {0: 8, 1: 11, 6: 1, 8: 7},
8: {2: 2, 6: 6, 7: 7},
9: {3: 5, 4: 3, 8: 9},
10: {1: 6, 5: 7, 9: 4}
}
print("=" * 60)
print("Dijkstra算法优化实现测试")
print("=" * 60)
start_vertex = 0
distances, predecessors = dijkstra_optimized(complex_graph, start_vertex)
print("\n" + "=" * 60)
print("最短路径结果")
print("=" * 60)
# 显示到每个顶点的最短路径
targets = [2, 4, 8, 10]
for target in targets:
path = find_shortest_path(predecessors, start_vertex, target)
if path:
print(f"{start_vertex} -> {target}: 距离={distances[target]}, 路径={' -> '.join(map(str, path))}")
else:
print(f"{start_vertex} -> {target}: 不可达")
# 运行性能对比
print("\n" + "=" * 60)
benchmark_dijkstra()
```
这个优化版本的关键改进在于使用了优先队列来快速找到距离最小的顶点。每次从堆中弹出最小元素的时间复杂度是O(logV)，而基础实现中查找最小元素需要O(V)。对于有V个顶点、E条边的图，优化后的时间复杂度是O((V+E)logV)。
## 4. 实战应用：从算法到解决真实问题
理解了算法的原理和实现后，让我们看看如何在实际项目中应用Dijkstra算法。我将分享几个我在实际工作中遇到的案例。
### 4.1 案例一：物流配送路径规划
在物流系统中，我们需要为配送车辆规划最优路线。这不仅仅是找最短路径，还需要考虑多个约束条件：
```python
class LogisticsRouter:
"""物流路径规划器"""
def __init__(self, warehouse_locations, road_network):
"""
初始化物流路径规划器
参数：
warehouse_locations: 仓库位置字典 {仓库ID: (经度, 纬度)}
road_network: 道路网络图 {位置ID: {邻居ID: (距离, 预计时间)}}
"""
self.locations = warehouse_locations
self.network = road_network
def find_optimal_route(self, start_warehouse, target_warehouse,
constraints=None):
"""
寻找考虑约束的最优路径
参数：
start_warehouse: 起始仓库ID
target_warehouse: 目标仓库ID
constraints: 约束条件字典，如 {'max_distance': 100, 'max_time': 120}
返回：
optimal_path: 最优路径列表
total_distance: 总距离
total_time: 总时间
"""
if constraints is None:
constraints = {}
# 根据约束选择权重
if 'minimize_time' in constraints and constraints['minimize_time']:
weight_type = 'time'
else:
weight_type = 'distance'
# 构建适合Dijkstra的图
graph = {}
for node in self.network:
graph[node] = {}
for neighbor, (distance, time) in self.network[node].items():
# 根据约束过滤边
if self._satisfies_constraints(distance, time, constraints):
weight = time if weight_type == 'time' else distance
graph[node][neighbor] = weight
# 使用优化版Dijkstra算法
distances, predecessors = dijkstra_optimized(graph, start_warehouse)
# 重建路径
path = self._reconstruct_path(predecessors, start_warehouse, target_warehouse)
# 计算总距离和时间
total_distance, total_time = self._calculate_metrics(path)
return {
'path': path,
'total_distance': total_distance,
'total_time': total_time,
'distances': distances
}
def _satisfies_constraints(self, distance, time, constraints):
"""检查边是否满足约束条件"""
if 'max_distance' in constraints and distance > constraints['max_distance']:
return False
if 'max_time' in constraints and time > constraints['max_time']:
return False
return True
def _reconstruct_path(self, predecessors, start, target):
"""重建路径"""
path = []
current = target
while current is not None:
path.append(current)
current = predecessors.get(current)
path.reverse()
return path if path and path[0] == start else []
def _calculate_metrics(self, path):
"""计算路径的总距离和时间"""
total_distance = 0
total_time = 0
for i in range(len(path) - 1):
current = path[i]
next_node = path[i + 1]
distance, time = self.network[current][next_node]
total_distance += distance
total_time += time
return total_distance, total_time
def visualize_route(self, path, output_file='route_map.html'):
"""可视化路径（使用folium生成交互式地图）"""
try:
import folium
# 创建地图中心点
center_lat = sum(loc[0] for loc in self.locations.values()) / len(self.locations)
center_lon = sum(loc[1] for loc in self.locations.values()) / len(self.locations)
m = folium.Map(location=[center_lat, center_lon], zoom_start=12)
# 添加所有仓库位置
for warehouse_id, (lat, lon) in self.locations.items():
color = 'red' if warehouse_id in [path[0], path[-1]] else 'blue'
folium.Marker(
[lat, lon],
popup=f'仓库 {warehouse_id}',
icon=folium.Icon(color=color)
).add_to(m)
# 添加路径线
path_coords = [self.locations[node] for node in path]
folium.PolyLine(
path_coords,
color='green',
weight=3,
opacity=0.8,
popup='配送路径'
).add_to(m)
# 保存地图
m.save(output_file)
print(f"地图已保存到 {output_file}")
except ImportError:
print("需要安装folium库来生成地图：pip install folium")
# 文本形式显示路径
print("路径顺序:", " -> ".join(map(str, path)))
print("途经位置:")
for node in path:
lat, lon = self.locations[node]
print(f" 仓库 {node}: 纬度={lat:.4f}, 经度={lon:.4f}")
# 使用示例
def logistics_example():
"""物流路径规划示例"""
# 模拟数据：仓库位置和道路网络
warehouse_locations = {
0: (31.2304, 121.4737), # 上海
1: (39.9042, 116.4074), # 北京
2: (23.1291, 113.2644), # 广州
3: (30.5728, 104.0668), # 成都
4: (29.8683, 121.5437), # 宁波
5: (22.5431, 114.0579), # 深圳
6: (34.3416, 108.9398), # 西安
7: (36.6512, 117.1200), # 济南
8: (28.2282, 112.9388), # 长沙
}
# 模拟道路网络（距离km, 时间分钟）
road_network = {
0: {1: (1200, 720), 4: (200, 150), 8: (1000, 600)},
1: {0: (1200, 720), 6: (1100, 660), 7: (400, 240)},
2: {5: (150, 90), 8: (600, 360)},
3: {6: (700, 420), 8: (900, 540)},
4: {0: (200, 150), 5: (1200, 720)},
5: {2: (150, 90), 4: (1200, 720)},
6: {1: (1100, 660), 3: (700, 420), 7: (800, 480)},
7: {1: (400, 240), 6: (800, 480)},
8: {0: (1000, 600), 2: (600, 360), 3: (900, 540)},
}
router = LogisticsRouter(warehouse_locations, road_network)
# 寻找从上海到广州的最短路径
print("寻找上海(0)到广州(2)的最优路径...")
result = router.find_optimal_route(0, 2)
print(f"最优路径: {' -> '.join(map(str, result['path']))}")
print(f"总距离: {result['total_distance']} km")
print(f"总时间: {result['total_time']} 分钟")
# 可视化路径
router.visualize_route(result['path'])
# 带约束的路径规划
print("\n寻找上海(0)到广州(2)的路径，最大距离限制为1500km...")
result_constrained = router.find_optimal_route(
0, 2,
constraints={'max_distance': 1500, 'minimize_time': True}
)
print(f"约束下最优路径: {' -> '.join(map(str, result_constrained['path']))}")
print(f"总距离: {result_constrained['total_distance']} km")
print(f"总时间: {result_constrained['total_time']} 分钟")
if __name__ == "__main__":
logistics_example()
```
这个物流路径规划器展示了如何将Dijkstra算法应用到实际业务中。它不仅仅计算最短路径，还考虑了实际约束条件，并提供了可视化功能。
### 4.2 案例二：网络路由优化
在网络通信中，数据包需要通过多个路由器传输。每个链路都有不同的延迟和带宽，Dijkstra算法可以帮助找到延迟最低的路径。
```python
class NetworkRouter:
"""网络路由优化器"""
def __init__(self, network_topology):
"""
初始化网络路由优化器
参数：
network_topology: 网络拓扑 {路由器ID: {邻居ID: (延迟ms, 带宽Mbps)}}
"""
self.topology = network_topology
def find_lowest_latency_path(self, source, destination):
"""寻找延迟最低的路径"""
return self._find_optimal_path(source, destination, metric='latency')
def find_highest_bandwidth_path(self, source, destination, min_bandwidth=10):
"""寻找带宽最高的路径（满足最小带宽要求）"""
return self._find_optimal_path(
source, destination,
metric='bandwidth',
min_bandwidth=min_bandwidth
)
def _find_optimal_path(self, source, destination, metric='latency', **constraints):
"""寻找最优路径的核心方法"""
# 根据度量标准构建图
graph = {}
for router in self.topology:
graph[router] = {}
for neighbor, (latency, bandwidth) in self.topology[router].items():
# 应用约束
if self._satisfies_constraints(latency, bandwidth, constraints):
if metric == 'latency':
weight = latency
elif metric == 'bandwidth':
# 对于带宽，我们想要最大值，所以使用负值作为权重
weight = -bandwidth
else:
raise ValueError(f"未知的度量标准: {metric}")
graph[router][neighbor] = weight
# 使用Dijkstra算法
distances, predecessors = dijkstra_optimized(graph, source)
# 重建路径
path = self._reconstruct_path(predecessors, source, destination)
# 计算路径指标
metrics = self._calculate_path_metrics(path)
return {
'path': path,
'distances': distances,
'metrics': metrics
}
def _satisfies_constraints(self, latency, bandwidth, constraints):
"""检查链路是否满足约束"""
if 'min_bandwidth' in constraints and bandwidth < constraints['min_bandwidth']:
return False
if 'max_latency' in constraints and latency > constraints['max_latency']:
return False
return True
def _reconstruct_path(self, predecessors, source, destination):
"""重建路径"""
path = []
current = destination
while current is not None:
path.append(current)
current = predecessors.get(current)
path.reverse()
return path if path and path[0] == source else []
def _calculate_path_metrics(self, path):
"""计算路径的各项指标"""
if not path or len(path) < 2:
return {'total_latency': 0, 'min_bandwidth': float('inf'), 'hop_count': 0}
total_latency = 0
min_bandwidth = float('inf')
for i in range(len(path) - 1):
current = path[i]
next_hop = path[i + 1]
latency, bandwidth = self.topology[current][next_hop]
total_latency += latency
min_bandwidth = min(min_bandwidth, bandwidth)
return {
'total_latency': total_latency,
'min_bandwidth': min_bandwidth,
'hop_count': len(path) - 1
}
def analyze_network(self):
"""分析网络拓扑特性"""
num_routers = len(self.topology)
num_links = sum(len(neighbors) for neighbors in self.topology.values()) // 2
# 计算平均度（每个路由器的平均连接数）
avg_degree = 2 * num_links / num_routers
# 计算网络直径（最长最短路径）
diameter = 0
for source in self.topology:
distances, _ = dijkstra_optimized(self.topology, source)
max_distance = max(d for d in distances.values() if d < float('inf'))
diameter = max(diameter, max_distance)
return {
'num_routers': num_routers,
'num_links': num_links,
'avg_degree': avg_degree,
'diameter': diameter,
'connectivity': '高' if avg_degree > 2 else '低'
}
# 网络路由示例
def network_routing_example():
"""网络路由优化示例"""
# 模拟网络拓扑
network_topology = {
'R1': {'R2': (10, 100), 'R3': (15, 1000)},
'R2': {'R1': (10, 100), 'R4': (20, 100), 'R5': (25, 1000)},
'R3': {'R1': (15, 1000), 'R6': (30, 100)},
'R4': {'R2': (20, 100), 'R7': (35, 100)},
'R5': {'R2': (25, 1000), 'R7': (15, 1000), 'R8': (40, 100)},
'R6': {'R3': (30, 100), 'R8': (20, 1000)},
'R7': {'R4': (35, 100), 'R5': (15, 1000), 'R9': (25, 100)},
'R8': {'R5': (40, 100), 'R6': (20, 1000), 'R9': (30, 1000)},
'R9': {'R7': (25, 100), 'R8': (30, 1000)},
}
router = NetworkRouter(network_topology)
# 分析网络
analysis = router.analyze_network()
print("网络拓扑分析:")
print(f" 路由器数量: {analysis['num_routers']}")
print(f" 链路数量: {analysis['num_links']}")
print(f" 平均连接度: {analysis['avg_degree']:.2f}")
print(f" 网络直径: {analysis['diameter']}")
print(f" 连通性: {analysis['connectivity']}")
# 寻找最低延迟路径
print("\n寻找R1到R9的最低延迟路径...")
latency_result = router.find_lowest_latency_path('R1', 'R9')
print(f"最低延迟路径: {' -> '.join(latency_result['path'])}")
print(f"总延迟: {latency_result['metrics']['total_latency']}ms")
print(f"最小带宽: {latency_result['metrics']['min_bandwidth']}Mbps")
print(f"跳数: {latency_result['metrics']['hop_count']}")
# 寻找最高带宽路径（至少50Mbps）
print("\n寻找R1到R9的最高带宽路径（最小50Mbps）...")
bandwidth_result = router.find_highest_bandwidth_path('R1', 'R9', min_bandwidth=50)
print(f"最高带宽路径: {' -> '.join(bandwidth_result['path'])}")
print(f"总延迟: {bandwidth_result['metrics']['total_latency']}ms")
print(f"最小带宽: {bandwidth_result['metrics']['min_bandwidth']}Mbps")
print(f"跳数: {bandwidth_result['metrics']['hop_count']}")
# 比较两种路径
print("\n路径对比:")
print(f"{'指标':<15} {'最低延迟路径':<20} {'最高带宽路径':<20}")
print("-" * 55)
print(f"{'总延迟(ms)':<15} {latency_result['metrics']['total_latency']:<20} {bandwidth_result['metrics']['total_latency']:<20}")
print(f"{'最小带宽(Mbps)':<15} {latency_result['metrics']['min_bandwidth']:<20} {bandwidth_result['metrics']['min_bandwidth']:<20}")
print(f"{'跳数':<15} {latency_result['metrics']['hop_count']:<20} {bandwidth_result['metrics']['hop_count']:<20}")
if __name__ == "__main__":
network_routing_example()
```
这个网络路由优化器展示了Dijkstra算法在通信网络中的应用。通过调整权重函数，我们可以优化不同的网络指标，如延迟、带宽或可靠性。
## 5. 高级主题与性能优化
在实际工程应用中，单纯的Dijkstra算法可能不够用。我们需要考虑更多现实世界的复杂性和性能要求。
### 5.1 处理大规模图的优化技巧
当图非常大时（比如数百万个顶点），即使是优化版的Dijkstra算法也可能变得很慢。以下是一些我在处理大规模图时用到的技巧：
```python
class AdvancedDijkstra:
"""高级Dijkstra算法实现，包含多种优化"""
def __init__(self, graph):
self.graph = graph
self.num_vertices = len(graph)
def bidirectional_search(self, start, target):
"""
双向搜索优化
从起点和目标点同时开始搜索，在中间相遇
可以显著减少搜索空间
"""
# 前向搜索（从起点开始）
forward_dist = {vertex: float('inf') for vertex in self.graph}
forward_dist[start] = 0
forward_heap = [(0, start)]
forward_visited = set()
# 反向搜索（从目标点开始）
backward_dist = {vertex: float('inf') for vertex in self.graph}
backward_dist[target] = 0
backward_heap = [(0, target)]
backward_visited = set()
# 构建反向图
reverse_graph = self._build_reverse_graph()
best_distance = float('inf')
meeting_point = None
while forward_heap and backward_heap:
# 前向搜索一步
if forward_heap:
f_dist, f_vertex = heapq.heappop(forward_heap)
if f_vertex in forward_visited:
continue
forward_visited.add(f_vertex)
# 检查是否在反向搜索中访问过
if f_vertex in backward_visited:
total_dist = f_dist + backward_dist[f_vertex]
if total_dist < best_distance:
best_distance = total_dist
meeting_point = f_vertex
# 扩展前向搜索
for neighbor, weight in self.graph.get(f_vertex, {}).items():
new_dist = f_dist + weight
if new_dist < forward_dist[neighbor]:
forward_dist[neighbor] = new_dist
heapq.heappush(forward_heap, (new_dist, neighbor))
# 反向搜索一步
if backward_heap:
b_dist, b_vertex = heapq.heappop(backward_heap)
if b_vertex in backward_visited:
continue
backward_visited.add(b_vertex)
# 检查是否在前向搜索中访问过
if b_vertex in forward_visited:
total_dist = b_dist + forward_dist[b_vertex]
if total_dist < best_distance:
best_distance = total_dist
meeting_point = b_vertex
# 扩展反向搜索
for neighbor, weight in reverse_graph.get(b_vertex, {}).items():
new_dist = b_dist + weight
if new_dist < backward_dist[neighbor]:
backward_dist[neighbor] = new_dist
heapq.heappush(backward_heap, (new_dist, neighbor))
return best_distance, meeting_point
def _build_reverse_graph(self):
"""构建反向图用于双向搜索"""
reverse_graph = {vertex: {} for vertex in self.graph}
for vertex in self.graph:
for neighbor, weight in self.graph[vertex].items():
if neighbor not in reverse_graph:
reverse_graph[neighbor] = {}
reverse_graph[neighbor][vertex] = weight
return reverse_graph
def a_star_search(self, start, target, heuristic):
"""
A*搜索算法
在Dijkstra基础上加入启发式函数，引导搜索方向
适用于知道目标点位置的情况
"""
# 初始化
g_score = {vertex: float('inf') for vertex in self.graph}
g_score[start] = 0
f_score = {vertex: float('inf') for vertex in self.graph}
f_score[start] = heuristic(start, target)
open_set = [(f_score[start], start)]
came_from = {}
while open_set:
current_f, current = heapq.heappop(open_set)
if current == target:
# 重建路径
path = []
while current in came_from:
path.append(current)
current = came_from[current]
path.append(start)
path.reverse()
return g_score[target], path
for neighbor, weight in self.graph.get(current, {}).items():
tentative_g = g_score[current] + weight
if tentative_g < g_score[neighbor]:
# 找到更好路径
came_from[neighbor] = current
g_score[neighbor] = tentative_g
f_score[neighbor] = tentative_g + heuristic(neighbor, target)
heapq.heappush(open_set, (f_score[neighbor], neighbor))
return float('inf'), [] # 未找到路径
def contraction_hierarchies_preprocess(self):
"""
收缩层次预处理
适用于需要多次查询的静态图
"""
# 计算顶点的重要性（基于度和局部结构）
importance = {}
for vertex in self.graph:
# 简单的度作为重要性度量
importance[vertex] = len(self.graph.get(vertex, {}))
# 按重要性排序
order = sorted(self.graph.keys(), key=lambda v: importance[v])
# 构建收缩图
contracted_graph = {vertex: {} for vertex in self.graph}
for vertex in order:
# 暂时移除当前顶点
neighbors = self.graph.get(vertex, {}).copy()
# 为每对邻居添加捷径
neighbor_list = list(neighbors.keys())
for i in range(len(neighbor_list)):
for j in range(i + 1, len(neighbor_list)):
u, v = neighbor_list[i], neighbor_list[j]
shortcut_weight = neighbors.get(u, float('inf')) + neighbors.get(v, float('inf'))
# 更新捷径
if shortcut_weight < contracted_graph[u].get(v, float('inf')):
contracted_graph[u][v] = shortcut_weight
contracted_graph[v][u] = shortcut_weight
return contracted_graph, order
def benchmark_optimizations(self, start, target, num_tests=100):
"""对比不同优化方法的性能"""
import time
results = {}
# 标准Dijkstra
start_time = time.time()
for _ in range(num_tests):
distances, _ = dijkstra_optimized(self.graph, start)
standard_result = distances.get(target, float('inf'))
results['standard'] = (time.time() - start_time) / num_tests
# 双向搜索
start_time = time.time()
for _ in range(num_tests):
bidir_result, _ = self.bidirectional_search(start, target)
results['bidirectional'] = (time.time() - start_time) / num_tests
# A*搜索（使用简单的欧几里得距离作为启发函数）
# 这里需要顶点有坐标信息，我们假设顶点是二维坐标
if hasattr(self, 'vertex_coords'):
def heuristic(v1, v2):
x1, y1 = self.vertex_coords[v1]
x2, y2 = self.vertex_coords[v2]
return ((x1 - x2) ** 2 + (y1 - y2) ** 2) ** 0.5
start_time = time.time()
for _ in range(num_tests):
astar_result, _ = self.a_star_search(start, target, heuristic)
results['a_star'] = (time.time() - start_time) / num_tests
print("性能对比结果（平均每次查询时间）:")
print("-" * 50)
for method, time_taken in results.items():
print(f"{method:15} {time_taken*1000:8.3f} ms")
return results
# 使用高级优化的示例
def advanced_optimization_example():
"""高级优化技术示例"""
# 创建一个较大的测试图
large_graph = {}
num_vertices = 1000
# 生成随机图
import random
for i in range(num_vertices):
large_graph[i] = {}
# 每个顶点平均连接5个邻居
for _ in range(5):
neighbor = random.randint(0, num_vertices - 1)
if neighbor != i:
weight = random.randint(1, 100)
large_graph[i][neighbor] = weight
print(f"生成测试图: {num_vertices}个顶点")
print(f"总边数: {sum(len(neighbors) for neighbors in large_graph.values())}")
# 创建高级Dijkstra实例
advanced = AdvancedDijkstra(large_graph)
# 测试双向搜索
print("\n测试双向搜索...")
start, target = 0, 999
distance, meeting = advanced.bidirectional_search(start, target)
print(f"从 {start} 到 {target} 的最短距离: {distance}")
print(f"相遇点: {meeting}")
# 性能对比
print("\n性能对比测试...")
advanced.benchmark_optimizations(start, target, num_tests=10)
if __name__ == "__main__":
advanced_optimization_example()
```
### 5.2 内存优化与并行计算
对于超大规模图，我们还需要考虑内存使用和计算并行化：
```python
import multiprocessing as mp
from typing import List, Dict, Tuple
import numpy as np
class ParallelDijkstra:
"""并行化Dijkstra算法实现"""
def __init__(self, graph: Dict[int, Dict[int, float]]):
self.graph = graph
self.num_vertices = len(graph)
def parallel_all_pairs(self, num_workers: int = None) -> np.ndarray:
"""
并行计算所有顶点对的最短路径
返回距离矩阵
"""
if num_workers is None:
num_workers = mp.cpu_count()
# 将顶点分配给不同的工作进程
vertices = list(self.graph.keys())
chunk_size = len(vertices) // num_workers
# 准备参数
args = []
for i in range(num_workers):
start_idx = i * chunk_size
end_idx = start_idx + chunk_size if i < num_workers - 1 else len(vertices)
worker_vertices = vertices[start_idx:end_idx]
args.append((worker_vertices, self.graph))
# 使用进程池并行计算
with mp.Pool(processes=num_workers) as pool:
results = pool.starmap(self._worker_compute, args)
# 合并结果
distance_matrix = np.full((self.num_vertices, self.num_vertices), np.inf)
for worker_result in results:
for src, distances in worker_result:
for dst, dist in distances.items():
distance_matrix[src][dst] = dist
# 设置对角线为0
np.fill_diagonal(distance_matrix, 0)
return distance_matrix
@staticmethod
def _worker_compute(vertices: List[int], graph: Dict) -> List[Tuple[int, Dict]]:
"""工作进程函数"""
results = []
for src in vertices:
distances, _ = dijkstra_optimized(graph, src)
results.append((src, distances))
return results
def memory_efficient_dijkstra(self, start: int,
chunk_size: int = 1000) -> Dict[int, float]:
"""
内存高效的Dijkstra实现
适用于无法一次性加载到内存的超大规模图
"""
# 使用磁盘存储部分结果
import tempfile
import os
# 初始化距离
distances = {start: 0}
# 使用外部排序的优先队列
# 这里简化为使用多个小堆
heaps = []
current_heap = []
heapq.heappush(current_heap, (0, start))
visited = set()
while current_heap or any(heaps):
if not current_heap and heaps:
# 从磁盘加载下一个堆
current_heap = heaps.pop(0)
current_dist, current = heapq.heappop(current_heap)
if current in visited:
continue
visited.add(current)
# 如果堆太大，保存到磁盘
if len(current_heap) > chunk_size:
heaps.append(current_heap)
current_heap = []
# 处理邻居
for neighbor, weight in self.graph.get(current, {}).items():
if neighbor in visited:
continue
new_dist = current_dist + weight
if new_dist < distances.get(neighbor, float('inf')):
distances[neighbor] = new_dist
heapq.heappush(current_heap, (new_dist, neighbor))
return distances
def incremental_update(self, edge_updates: List[Tuple[int, int, float]]):
"""
增量更新算法
当图发生小变化时，避免重新计算所有最短路径
"""
# 这里实现一个简化的增量更新
# 实际应用中可能需要更复杂的算法
affected_vertices = set()
# 找出受影响的顶点
for u, v, new_weight in edge_updates:
old_weight = self.graph.get(u, {}).get(v)
if old_weight is not None and new_weight != old_weight:
affected_vertices.add(u)
affected_vertices.add(v)
# 只重新计算受影响顶点的最短路径
updated_distances = {}
for vertex in affected_vertices:
distances, _ = dijkstra_optimized(self.graph, vertex)
updated_distances[vertex] = distances
return updated_distances
def memory_optimization_example():
"""内存优化示例"""
# 创建一个非常大的图（模拟）
print("创建大规模图...")
# 使用稀疏矩阵表示来节省内存
from scipy import sparse
import numpy as np
num_vertices = 10000
density = 0.001 # 稀疏图
# 创建随机稀疏邻接矩阵
data = np.random.rand(int(num_vertices * num_vertices * density)) * 100
rows = np.random.randint(0, num_vertices, len(data))
cols = np.random.randint(0, num_vertices, len(data))
# 移除自环
mask = rows != cols
data = data[mask]
rows = rows[mask]
cols = cols[mask]
adjacency_matrix = sparse.csr_matrix((data, (rows, cols)),
shape=(num_vertices, num_vertices))
print(f"图创建完成: {num_vertices}个顶点")
print(f"非零元素: {adjacency_matrix.nnz}")
print(f"密度: {adjacency_matrix.nnz / (num_vertices * num_vertices):.6f}")
# 转换为邻接表（只转换一部分用于演示）
print("\n转换为邻接表表示...")
graph = {}
for i in range(min(1000, num_vertices)): # 只转换前1000个顶点
neighbors = adjacency_matrix.getrow(i)
graph[i] = {}
for j in range(neighbors.indices.size):
if neighbors.indices[j] < 1000: # 只考虑前1000个顶点
graph[i][neighbors.indices[j]] = neighbors.data[j]
# 测试内存高效算法
print("\n测试内存高效Dijkstra...")
parallel = ParallelDijkstra(graph)
start_time = time.time()
distances = parallel.memory_efficient_dijkstra(0, chunk_size=100)
elapsed = time.time() - start_time
print(f"计算完成，耗时: {elapsed:.3f}秒")
print(f"计算了从顶点0到{len(distances)}个顶点的距离")
# 测试增量更新
print("\n测试增量更新...")
edge_updates = [(0, 1, 50.0), (1, 2, 30.0)]
updated = parallel.incremental_update(edge_updates)
print(f"更新了{len(updated)}个顶点的距离")
if __name__ == "__main__":
import time
memory_optimization_example()
```
## 6. 常见问题与调试技巧
在实际使用Dijkstra算法时，你可能会遇到各种问题。以下是我在多年实践中总结的一些常见问题和解决方案。
### 6.1 负权边问题
Dijkstra算法不能处理负权边，这是由其贪心策略决定的。如果图中存在负权边，算法可能得到错误的结果。
```python
def detect_negative_weights(graph):
"""检测图中是否存在负权边"""
negative_edges = []
for u in graph:
for v, w in graph[u].items():
if w < 0:
negative_edges.append((u, v, w))
if negative_edges:
print(f"警告: 发现 {len(negative_edges)} 条负权边")
for u, v, w in negative_edges[:5]: # 只显示前5条
print(f" 边 {u}->{v}: 权重={w}")
if len(negative_edges) > 5:
print(f" ... 还有 {len(negative_edges)-5} 条")
return True, negative_edges
return False, []
def bellman_ford_for_negative_weights(graph, start):
"""
Bellman-Ford算法，可以处理负权边
时间复杂度: O(VE)
"""
# 初始化距离
distances = {vertex: float('inf') for vertex in graph}
distances[start] = 0
predecessors = {vertex: None for vertex in graph}
# 松弛操作 V-1 次
for _ in range(len(graph) - 1):
updated = False
for u in graph:
for v, w in graph[u].items():
if distances[u] + w < distances[v]:
distances[v] = distances[u] + w
predecessors[v] = u
updated = True
# 如果没有更新，提前结束
if not updated:
break
# 检查负权环
for u in graph:
for v, w in graph[u].items():
if distances[u] + w < distances[v]:
print("警告: 图中存在负权环！")
return None, None
return distances, predecessors
def compare_algorithms_on_negative_graph():
"""在有负权边的图上比较Dijkstra和Bellman-Ford"""
# 创建包含负权边的图
graph_with_negative = {
0: {1: 5, 2: 2},
1: {3: 1},
2: {1: -1, 3: 4}, # 负权边
3: {4: 3},
4: {}
}
start = 0
print("图包含负权边:")
for u in graph_with_negative:
for v, w in graph_with_negative[u].items():
print(f" {u} -> {v}: {w}")
print("\n" + "="*50)
print("Dijkstra算法结果:")
print("="*50)
try:
dist_dijkstra, pred_dijkstra = dijkstra_optimized(graph_with_negative, start)
for v in sorted(dist_dijkstra.keys()):
print(f" 到顶点{v}的距离: {dist_dijkstra[v]}")
except Exception as e:
print(f" Dijkstra算法出错: {e}")
print("\n" + "="*50)
print("Bellman-Ford算法结果:")
print("="*50)
dist_bf, pred_bf = bellman_ford_for_negative_weights(graph_with_negative, start)
if dist_bf:
for v in sorted(dist_bf.keys()):
print(f" 到顶点{v}的距离: {dist_bf[v]}")
# 比较结果
if dist_bf:
print("\n" + "="*50)
print("结果比较:")
print("="*50)
print(f"{'顶点':<5} {'Dijkstra':<15} {'Bellman-Ford':<15} {'差异':<10}")
print("-"*50)
for v in sorted(graph_with_negative.keys()):
d1 = dist_dijkstra.get(v, float('inf')) if 'dist_dijkstra' in locals() else 'N/A'
d2 = dist_bf.get(v, float('inf'))
diff = abs(d1 - d2) if isinstance(d1, (int, float)) and isinstance(d2, (int, float)) else 'N/A'
print(f"{v:<5} {str(d1):<15} {str(d2):<15} {str(diff):<10}")
```
### 6.2 性能分析与优化建议
当算法运行缓慢时，如何定位性能瓶颈？
```python
class DijkstraProfiler:
"""Dijkstra算法性能分析器"""
def __init__(self, graph):
self.graph = graph
self.stats = {
'heap_operations': 0,
'relaxations': 0,
'vertices_processed': 0,
'edges_processed': 0,
'time_taken': 0
}
def profile_dijkstra(self, start):
"""带性能分析的Dijkstra算法"""
import time
distances = {vertex: float('inf') for vertex in self.graph}
distances[start] = 0
predecessors = {vertex: None for vertex in self.graph}
# 使用堆作为优先队列
heap = [(0, start)]
visited = set()
start_time = time.time()
while heap:
self.stats['heap_operations'] += 1
current_distance, current_vertex = heapq.heappop(heap)
if current_vertex in visited:
continue
visited.add(current_vertex)
self.stats['vertices_processed'] += 1
# 处理所有邻居
for neighbor, weight in self.graph.get(current_vertex, {}).items():
self.stats['edges_processed'] += 1
self.stats['relaxations'] += 1
new_distance = current_distance + weight
if new_distance < distances[neighbor]:
distances[neighbor] = new_distance
predecessors[neighbor] = current_vertex
heapq.heappush(heap, (new_distance, neighbor))
self.stats['heap_operations'] += 1
self.stats['time_taken'] = time.time() - start_time
return distances, predecessors
def analyze_performance(self):
"""分析性能瓶颈"""
print("性能分析报告")
print("=" * 60)
print(f"顶点总数: {len(self.graph)}")
print(f"边总数: {sum(len(neighbors) for neighbors in self.graph.values())}")
print(f"处理顶点数: {self.stats['vertices_processed']}")
print(f"处理边数: {self.stats['edges_processed']}")
print(f"堆操作次数: {self.stats['heap_operations']}")
print(f"松弛操作次数: {self.stats['relaxations']}")
print(f"总耗时: {self.stats['time_taken']:.6f}秒")
# 计算各种比率
if self.stats['vertices_processed'] > 0:
edges_per_vertex = self.stats['edges_processed'] / self.stats['vertices_processed']
heap_ops_per_vertex = self.stats['heap_operations'] / self.stats['vertices_processed']
print(f"\n平均每个顶点处理的边数: {edges_per_vertex:.2f}")
print(f"平均每个顶点的堆操作数: {heap_ops_per_vertex:.2f}")
# 提供优化建议
print("\n优化建议:")
if edges_per_vertex > 10:
print(" - 图较稠密，考虑使用斐波那契堆进一步优化")
if heap_ops_per_vertex > 5:
print(" - 堆操作频繁，检查是否有重复顶点入堆")
avg_time_per_vertex = self.stats['time_taken'] / max(1, self.stats['vertices_processed'])
if avg_time_per_vertex > 0.001: # 1毫秒
print(f" - 每个顶点平均处理时间较高: {avg_time_per_vertex*1000:.2f}ms")
print(" - 考虑使用双向搜索或A*算法优化")
def compare_data_structures(self, start):
"""比较不同数据结构实现的性能"""
import time
implementations = {
'列表实现': self._dijkstra_with_list,
'二叉堆实现': self._dijkstra_with_binary_heap,
'斐波那契堆实现': self._dijkstra_with_fibonacci_heap
}
results = {}
for name, implementation in implementations.items():
print(f"\n测试 {name}...")
# 重置统计
self.stats = {k: 0 for k in self.stats}
start_time = time.time()
distances, _ = implementation(start)
elapsed = time.time() - start_time
results[name] = {
'time': elapsed,
'vertices': self.stats['vertices_processed'],
'heap_ops': self.stats['heap_operations']
}
print(f" 耗时: {elapsed:.6f}秒")
print(f" 处理顶点数: {self.stats['vertices_processed']}")
print(f" 堆操作数: {self.stats['heap_operations']}")
# 显示比较结果
print("\n" + "="*60)
print("数据结构性能比较")
print("="*60)
print(f"{'实现方式':<20} {'耗时(秒)':<15} {'堆操作数':<15} {'效率':<10}")
print("-"*60)
best_time = min(r['time'] for r in results.values())
for name, data in results.items():
efficiency = best_time / data['time'] if data['time'] > 0 else 0
print(f"{name:<20} {data['time']:<15.6f} {data['heap_ops']:<15} {efficiency:<10.2f}")
def _dijkstra_with_list(self, start):
"""使用列表实现的Dijkstra（基础版本）"""
distances = {vertex: float('inf') for vertex in self.graph}
distances[start] = 0
predecessors = {vertex: None for vertex in self.graph}
unvisited = set(self.graph.keys())
while unvisited:
# 线性查找最小距离顶点
current = min(unvisited, key=lambda v: distances[v])
unvisited.remove(current)
self.stats['vertices_processed'] += 1
for neighbor, weight in self.graph.get(current, {}).items():
if neighbor in unvisited:
self.stats['edges_processed'] += 1
new_distance = distances[current] + weight
if new_distance < distances[neighbor]:
distances[neighbor] = new_distance
predecessors[neighbor] = current
return distances, predecessors
def _dijkstra_with_binary_heap(self, start):
"""使用二叉堆实现的Dijkstra"""
return self.profile_dijkstra(start) # 使用之前实现的版本
def _dijkstra_with_fibonacci_heap(self, start):
"""使用斐波那契堆实现的Dijkstra"""
# 简化实现，实际使用时需要斐波那契堆的实现
# 这里使用二叉堆模拟
return self.profile_dijkstra(start)
def performance_analysis_example():
"""性能分析示例"""
# 创建测试图
test_graph = {}
num_vertices = 500
# 生成随机图
import random
for i in range(num_vertices):
test_graph[i] = {}
# 每个顶点连接约10%的其他顶点
num_edges = random.randint(int(num_vertices * 0.05), int(num_vertices * 0.15))
neighbors = random.sample(range(num_vertices), min(num_edges, num_vertices))
for neighbor in neighbors:
if neighbor != i:
weight = random.randint(1, 100)
test_graph[i][neighbor] = weight
print(f"测试图: {num_vertices}个顶点")
total_edges = sum(len(neighbors) for neighbors in test_graph.values())
print(f"总边数: {total_edges}")
print(f"平均度: {total_edges / num_vertices:.2f}")
# 创建性能分析器
profiler = DijkstraProfiler(test_graph)
# 分析标准实现
print("\n" + "="*60)
print("分析标准Dijkstra实现")
print("="*60)
distances, _ = profiler.profile_dijkstra(0)
profiler.analyze_performance()
# 比较不同数据结构
print("\n" + "="*60)
print("比较不同数据结构实现")
print("="*60)
profiler.compare_data_structures(0)
# 分析图的特性
print("\n" + "="*60)
print("图结构分析")
print("="*60)
# 计算图的直径和平均路径长度
def compute_graph_metrics(graph, sample_size=50):
"""计算图的一些度量指标"""
vertices = list(graph.keys())
if len(vertices) > sample_size:
vertices = random.sample(vertices, sample_size)
path_lengths = []
for i, src in enumerate(vertices):
distances, _ = dijkstra_optimized(graph, src)
for dst in vertices[i+1:]:
if distances[dst] < float('inf'):
path_lengths.append(distances[dst])
if path_lengths:
avg_path = sum(path_lengths) / len(path_lengths)
max_path = max(path_lengths)
print(f"平均最短路径长度: {avg_path:.2f}")
print(f"图直径(采样估计): {max_path:.2f}")
print(f"连通性: {len(path_lengths) / (len(vertices)*(len(vertices)-1)/2):.2%}")
compute_graph_metrics(test_graph)
if __name__ == "__main__":
performance_analysis_example()
```
通过这些工具和技术，你可以更好地理解和优化Dijkstra算法在实际应用中的性能。记住，没有一种优化策略适用于所有场景，关键是根据具体问题和数据特点选择合适的方法。
在实际项目中，我经常发现性能瓶颈不是算法本身，而是数据预处理、内存访问模式或者I/O操作。因此，全面的性能分析和系统化的优化方法比单纯追求算法复杂度更重要。
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Python内容推荐
python Dijkstra算法实现最短路径问题的方法
主要介绍了python Dijkstra算法实现最短路径问题的方法,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
python实现Dijkstra算法的最短路径问题
主要介绍了python实现Dijkstra算法的最短路径问题，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
Python使用Dijkstra算法实现求解图中最短路径距离问题详解
本文实例讲述了Python使用Dijkstra算法实现求解图中最短路径距离问题。分享给大家供大家参考，具体如下： 这里继续前面一篇《Python基于Floyd算法求解最短路径距离问题》的内容，这里要做的是Dijkstra算法，与Floyd算法类似，二者的用途均为求解最短路径距离，在图中有着广泛的应用，二者的原理都是老生常谈了，毕竟本科学习数据结构的同学是不可能不学习这两个算法的，所以在这里我也不再累赘，只简单概述一下这个算法的核心思想： Dijkstra算法的输入有两个参数，一个是原始的数据矩阵，一个是起始的顶点下标，算法的思想也很简单容易理解，在开始的时候，需要设置两个集合，用于存储顶点和路
Python数据结构与算法之图的最短路径(Dijkstra算法)完整实例
本文实例讲述了Python数据结构与算法之图的最短路径(Dijkstra算法)。分享给大家供大家参考，具体如下： # coding:utf-8 # Dijkstra算法——通过边实现松弛 # 指定一个点到其他各顶点的路径——单源最短路径 # 初始化图参数 G = {1:{1:0, 2:1, 3:12}, 2:{2:0, 3:9, 4:3}, 3:{3:0, 5:5}, 4:{3:4, 4:0, 5:13, 6:15}, 5:{5:0, 6:4}, 6:{6:0}} # 每次找到离源点最近的一个顶点，然后以该顶点为重心进行扩展 # 最终的到源点到其余所有点的最短路径 # 一
python实现最短路径的实例方法
最短路径问题（python实现） 解决最短路径问题：（如下三种算法） （1）迪杰斯特拉算法（Dijkstra算法） （2）弗洛伊德算法（Floyd算法） （3）SPFA算法 第一种算法： Dijkstra算法 广度优先搜索解决赋权有向图或者无向图的单源最短路径问题.是一种贪心的策略 算法的思路 声明一个数组dis来保存源点到各个顶点的最短距离和一个保存已经找到了最短路径的顶点的集合：T，初始时，原点s的路径权重被赋为0（dis[s]=0）。若对于顶点s存在能直接到达的边（s,m），则把dis[m]设为w（s, m）,同时把所有其他（s不能直接到达的）顶点的路径长度设为无穷大。初始时，集合T只有
Python实现Dijkstra算法
今天小编就为大家分享一篇关于Python实现Dijkstra算法，小编觉得内容挺不错的，现在分享给大家，具有很好的参考价值，需要的朋友一起跟随小编来看看吧
python编写的最短路径算法
本文给大家分享的是python 无向图最短路径算法：请各位大大指教，继续改进。(修改了中文字符串，使py2exe中文没烦恼）,需要的朋友可以参考下
无人车路径规划算法matlab+python代码.rar
无人车路径规划算法matlab+python代码.rar
基于python的无人车路径规划算法设计与实现
基于python的无人车路径规划算法设计与实现
堆优化的Dijkstra算法用PYTHON实现
戴克斯特拉算法（Dijkstra’s algorithm）是由荷兰计算机科学家艾兹赫尔·戴克斯特拉提出。迪科斯彻算法使用了广度优先搜索解决非负权有向图的单源最短路径问题，算法最终得到一个最短路径树。该算法常用于路由算法或者作为其他图算法的一个子模块。 该算法的输入包含了一个有权重的有向图 G，以及G中的一个来源顶点 S。我们以 V 表示 G 中所有顶点的集合。每一个图中的边，都是两个顶点所形成的有序元素对。(u, v) 表示从顶点 u 到 v 有路径相连。我们以 E 表示G中所有边的集合，而边的权重则由权重函数 w: E → [0, ∞] 定义。因此，w(u, v) 就是从顶点 u 到顶点 v 的非负权重（weight）。边的权重可以想像成两个顶点之间的距离。任两点间路径的权重，就是该路径上所有边的权重总和。已知有 V 中有顶点 s 及 t，Dijkstra 算法可以找到 s 到 t的最低权重路径(例如，最短路径)。这个算法也可以在一个图中，找到从一个顶点 s 到任何其他顶点的最短路径。对于不含负权的有向图，Dijkstra算法是目前已知的最快的单源最短路径算法。
Dijkstra最短路径算法的Python实现_python_代码_下载
Dijkstra 算法的快速的实现，用于在连通图中查找最短路径距离。 该实现的运行时间为O((m+n) log n)，其中n是顶点数，m是边数。如果图是连接的（即，在一块），m通常支配n，使算法O(m log n)整体
python实现dijkstra最短路由算法
主要为大家详细介绍了python实现dijkstra最短路由算法，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
python矩阵/字典实现最短路径算法
主要为大家详细介绍了python矩阵/字典实现最短路径算法，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
python实现Dijkstra静态寻路算法
算法介绍 迪科斯彻算法使用了广度优先搜索解决赋权有向图或者无向图的单源最短路径问题，算法最终得到一个最短路径树。该算法常用于路由算法或者作为其他图算法的一个子模块。 当然目前也有人将它用来处理物流方面，以获取代价最小的运送方案。 算法思路 Dijkstra算法采用的是一种贪心的策略。 1.首先，声明一个数组dis来保存源点到各个顶点的最短距离和一个保存已经找到了最短路径的顶点的集合T。 2.其次，原点 s 的路径权重被赋为 0 （dis[s] = 0）。若对于顶点 s 存在能直接到达的边（s,m），则把dis[m]设为w（s, m）,同时把所有其他（s不能直接到达的）顶点的路径长度设为无穷
Python基于Floyd算法求解最短路径距离问题实例详解
本文实例讲述了Python基于Floyd算法求解最短路径距离问题。分享给大家供大家参考，具体如下： Floyd算法和Dijkstra算法，相信大家都不陌生，在最短路径距离的求解中应该算得上是最为基础和经典的两个算法了，今天就用一点时间来重新实现一下，因为本科的时候学习数据结构才开始接触的这个算法，当时唯一会用的就是C语言了，现在的话，C语言几乎已经离我远去了，个人感觉入手机器学习以来python更得我心，因为太通俗易懂了，带给你的体验自然也是非常不错的。 当然网上 有很多的算法讲解教程，我不会在这里累赘Floyd是什么原理，因为相信大家都熟悉，简单地说就是：三角不等式这个核心思想，如果我要求顶
Dijkstra算法的流程图
Dijkstra算法的流程图，具体的算法的实现，以及该算法的文档。 Dijkstra算法的流程图，具体的算法的实现，以及该算法的文档。
SDN-Route:使用 RYU 控制器在 SDN 中实现 Dijkstra 最短路径
SDN-路由 使用 RYU 控制器在 SDN 中实现 Dijkstra 最短路径
贪心算法——最短路径算法
算法这么课程的结课论文，以最短路径算法为例描述贪心算法
基于dijkstra算法的最短路径的求解
基于Dijkstra的最短路径算法和求解
并行Dijkstra最短路径算法
并行Dijkstra最短路径算法，附有测试文件
热门内容1 ai编程2 python采集义乌购（yiwugo）商品详情API接口，json数据返回3 TPM4 别再死记硬背DFA/NFA了！用Python手把手带你模拟一个正则表达式引擎（附完整代码）5 WorkBuddy6 手把手教你用STM32F103的ADC+TIM+DMA+FFT测频率，精度千分之八以内（附完整源码）7 Grok bot 全速入门【持续更新中】【限时优惠】8 Codex本地部署实战：Node.js配置、config.toml硬核调优与内容工作流搭建9 MCUboot 嵌入式安全启动与固件升级完全架构与实战指南10 Qwen Code本地部署全链路指南：Node.js环境、vLLM推理与100次免费调用原理
热门代码资源
最新推荐
科技中介服务机构想要提升服务效率，需要哪些数智工具支持？.docx
科技中介服务机构想要提升服务效率，需要哪些数智工具支持？
usb 扩容盘检测修复工具源码，编译产物
usb 扩容盘检测修复工具源码，编译产物
融合UWB和IMU割草机定位EKF研究（Matlab代码实现）
内容概要：本文主要介绍了一个名为“Plecs全桥LLC谐振变换器仿真模型”的技术资源，该资源聚焦于电力电子领域中的LLC谐振变换器建模与仿真，涵盖开环、闭环控制、软启动设计及参数优化等核心内容。文档不仅提供了LLC变换器的详细建模方法与仿真案例，还系统梳理了其在高效电源转换系统中的应用，包括隔离型全桥拓扑结构的设计与性能分析。此外，资源汇总了大量Matlab/Simulink仿真模型及相关科研方向，涉及新能源并网、储能控制、电机驱动等多个前沿领域，全面展示LLC谐振变换器在现代电力系统中的关键技术实现路径。同时，文档附带多种学习资料获取途径，如网盘链接和公众号资源，便于研究人员深入实践与拓展研究。;
适合人群：具备扎实的电力电子、自动控制理论基础，从事新能源发电、开关电源设计、电力系统仿真等相关领域的研究生、工程师及科研人员。;
使用场景及目标：①深入理解LLC谐振变换器的工作原理、建模方法与控制策略；②开展电力电子变换器的仿真验证与参数优化设计；③支撑科研项目、学位论文或工程项目的系统仿真与方案验证；
阅读建议：建议结合所提供的仿真模型与参考文献，使用Plecs或Simulink软件进行动手实践，重点掌握LLC谐振变换器的建模流程、闭环控制设计、软启动机制及参数调试技巧，以全面提升在实际工程中的应用与创新能力。
基于变频控制的多相交错LLC谐振变换器闭环仿真（Simulink仿真实现）
基于变频控制的多相交错LLC谐振变换器闭环仿真（Simulink仿真实现）内容概要：本文主要介绍了基于Plecs的全桥LLC谐振变换器仿真模型，涵盖变频控制下的多相交错LLC谐振变换器闭环仿真、V2G电动汽车ACDC整流与双向CLLC谐振变换器充放电仿真、LLC谐振变换器的开环与闭环软启动设计，以及储能系统中双向Buck-Boost变换器结合模型预测电流控制和电压下垂控制的仿真研究。同时，文档列举了多种电力电子系统与控制策略的Simulink/Plecs仿真案例，涉及三相交错并联Buck变换器、三电平Buck直流变换器、单相Boost PFC电路等，展示了丰富的电力变换器建模与控制技术。此外，还包括参数设计报告、计算书及相关参考文献，适用于从事电力电子、新能源发电、电动汽车充放电系统等领域的研究人员和技术开发人员。;
适合人群：具备电力电子、自动控制理论基础，从事新能源、电动汽车、电力系统等相关领域研究的研发人员、高校研究生及高年级本科生。;
使用场景及目标：① 学习和掌握LLC谐振变换器及其他典型电力电子变换器的工作原理与建模方法；② 实现闭环控制策略如变频控制、软启动、模型预测控制、下垂控制等的仿真验证；③ 支持科研项目、毕业设计或工程开发中的系统仿真与优化设计。;
阅读建议：建议结合文中提供的仿真模型、设计报告与参考文献，逐步搭建并调试相应系统，重点关注控制策略的实现细节与参数整定过程，以深入理解系统动态响应特性与控制性能。
基于Java17、SpringBoot、vue3、antdv-next、uniapp等主流技术构建的全场景极速开发平台 MDP内置3个web端、1个移动端 重点解决用户中心、开发者平台、管理后台3个应
MDP-主数据平台，后端项目。专注于单点登录解决方案。内置用户中心、开发者平台、管理后台3个应用。包含：单点登录、Oauth2 登录、用户身份管理、应用申请、应用接入、主数据维护、主数据订阅、主数据广播、应用接口调用、接口管理管理和查看、系统配置等功能。
信捷XD5 PLC控制码垛机械手实现自动化六轴编程
本文讨论了如何基于信捷XD5 PLC和威纶触摸屏编写适用于码垛机和机械手的六轴控制PLC程序，介绍了整个系统的架构与设计，并着重于操作界面与监控功能的实现。以下是对文章中涉及的关键知识点的详细介绍。
### 信捷XD5 PLC编程基础
信捷XD5 PLC是信捷电气推出的一款可编程逻辑控制器，广泛应用于工业自动化领域。为了编写码垛机和机械手的六轴控制程序，首先需要掌握信捷XD5 PLC的基础知识，包括但不限于：
- **PLC硬件结构**：了解PLC的基本组成，包括CPU模块、输入/输出模块、通讯模块等。
- **脉冲输出指令**：在控制机械手时，脉冲输出是实现精确位置控制的关键技术，需要熟练使用信捷XD5提供的脉冲输出指令进行编程。
- **编程语言**：信捷XD5 PLC支持多种编程语言，如梯形图、指令列表、结构化文本等。针对码垛机和机械手的控制，选择合适的编程语言以简化编程逻辑和提高执行效率。
- **数据处理**：了解如何在PLC程序中处理各种数据类型，包括整数、实数、字符串等，以及如何进行数据转换和运算。
### 六轴控制系统架构
六轴控制系统是指能够控制六个自由度的机械装置的系统，如机械手。其控制系统需要对每个轴的位置、速度、加速度等进行精确控制，以确保执行的动作精确到位。在编写PLC程序时，需要考虑到以下几个方面：
- **主程序框架设计**：定义整个控制程序的主体结构，包括初始化、错误处理、周期任务等。
- **手动控制子程序**：在手动模式下，需要编写能够响应操作员输入并控制机械手移动的子程序。
- **自动控制子程序**：自动模式下，需要设计能够按照预设的路径和速度控制机械手移动的子程序。
### 威纶触摸屏人机交互设计
威纶触摸屏是工业领域常用的触摸式人机界面（HMI），它允许操作者直接与机器进行交互，方便监控和控制机器运行状态。在本项目中，威纶触摸屏的使用包括：
- **操作界面设计**：设计直观易用的操作界面，包括必要的按钮、指示灯、数据显示窗口等。
- **设备状态监控**：实时显示机械手和码垛机的运行状态，包括各轴的位置、速度和状态信息。
- **人机交互实现**：通过触摸屏响应操作员的指令，如启动、停止、调整参数等，以及提供安全保护和故障诊断功能。
### 六轴控制程序编写实践
在编写程序时，需要关注以下几个关键环节：
- **程序的可靠性**：通过代码优化和测试，确保控制程序在各种工况下都能稳定运行。
- **程序的可维护性**：编写程序时应该考虑未来可能的功能扩展或修改，注释应详尽，逻辑应清晰，便于后续维护。
- **实际应用调试**：程序在实际设备上运行之前，需要进行充分的模拟和现场调试，确保满足生产需求。
### 代码示例与注释
文章提供了详细的代码示例，这有助于读者理解如何将理论知识应用到实践中。丰富的注释不仅可以帮助读者理解代码的功能和逻辑，还能在调试时迅速定位问题。
### 结语
综上所述，码垛机机械手六轴控制程序及PLC程序的编写需要深入理解PLC的编程原理和机械手的工作方式，并能够将这些知识融合到实际应用中。通过信捷XD5 PLC和威纶触摸屏的结合，可以实现高度自动化和智能化的生产控制。这套控制程序的成功实践不仅可以提高生产效率，还能增强设备的可靠性与操作的便捷性。对于自动化设备开发的技术人员而言，这是一份宝贵的学习材料，能够为他们的工作提供直接的技术支持和参考。
【Python爬虫避坑黄金法则】：23个生产环境血泪教训+17种高频报错的秒级定位方案（2024年最新版）
# 爬虫工程化落地的现实困境与认知重构
在智能终端日益普及、数据驱动决策成为企业核心能力的今天，爬虫早已不是实验室里的玩具脚本，而是支撑电商比价、金融风控、舆情监测、供应链预警等关键业务的数据动脉。但当团队把一个“能跑通”的脚本部署到生产环境，面对日均百万级请求、数十个目标站点、多层CDN/WAF防护和不断演进的法律边界时，系统往往在两周内开始失血：任务成功率从98%断崖式跌至62%，故障平均定位耗时飙升至47分钟，运维同学深夜被PagerDuty警报惊醒，却只能对着满屏`403 Forbidden`和`cf_clearance`跳转日志束手无策。
问题出在哪？不是缺工具——`reques
C盘快满了，想把D盘空间合并过去，但中间隔着恢复分区，该怎么安全操作？
### Windows C盘空间不足解决方案
当面对C盘空间不足的问题时，可以采取多种方式来解决问题。一种常见的做法是通过合并磁盘分区增加C盘的空间。然而，在某些情况下，如C盘和D盘之间存在其他分区（例如恢复分区），这会使得直接使用Windows自带的磁盘管理工具变得困难。
对于这种情况，推荐采用专业的第三方软件——傲梅分区助手来进行操作[^3]。该工具不仅能够处理非相邻分区之间的合并问题，还支持更多高级功能，从而提供更加灵活且安全的选择。具体来说：
- **使用傲梅分区助手**
如果目标是在不损害数据的前提下完成C盘扩容，则建议下载安装最新版的傲梅分区助手。启动程序后，可以通
氨气催化裂解模拟研究：COMSOL在不同压力温度下的应用
在现代化学工程和催化反应研究中，氨气（NH3）催化裂解是一个关键过程，它与能源、材料、农业等多个领域密切相关。氨气催化裂解不仅可以作为氢气的一种来源，而且在合成氨、肥料制造、化工原料制备等多个环节扮演着重要角色。随着对高效、环保催化技术需求的不断增加，对氨气催化裂解过程的深入研究显得尤为重要。
COMSOL Multiphysics 是一款功能强大的多物理场模拟软件，它能够模拟和分析各种物理过程和工程问题。利用COMSOL软件对氨气催化裂解进行模拟，研究人员可以在计算机上复现反应过程，从而预测不同条件下的裂解行为，优化反应条件，并且节省了实验成本和时间。
在 COMSOL 中模拟氨气催化裂解，首先需要明确的是物理模型的建立。物理模型通常基于连续介质力学原理，包括流体动力学（流体流动）、传热学（热传递）以及化学反应动力学（催化反应速率）。为了建立模型，研究人员需要定义适当的几何结构、物理场、边界条件和初始条件。几何结构取决于所模拟反应器的具体类型，比如管式反应器或搅拌釜式反应器。物理场则是根据氨气催化裂解的原理，可能包括但不限于流体流动、传热、质量传输、多相流以及化学反应等模块。设置参数则涉及反应速率常数、活化能、热容等反应动力学和热力学数据。最后，模拟运行需要合适的求解器来解决模型中的偏微分方程组。
在COMSOL中，可以通过参数化分析功能来研究不同压力和温度对氨气催化裂解的影响。在模型中设置相应的参数范围，通过模拟运行一系列参数变化，研究者可以观察到不同条件下的裂解速率和产物分布变化。例如，增加压力通常会导致反应物分压上升，从而提高反应速率，但同时也会增加能耗。而温度则是影响反应速率的另一个重要因素，通常提高温度会加快反应速率，但是超过催化剂的最佳使用温度后，反应速率反而会下降，因为高温可能造成催化剂的失活。
在COMSOL模拟中，研究者还可以通过查看温度分布、浓度分布、流场分布等模拟结果，来直观理解在不同压力和温度条件下，氨气催化裂解过程的具体变化。此外，通过分析不同条件下的反应动力学数据，研究者可以进一步探究反应机理，进而指导实验设计。
将COMSOL模拟技术应用于氨气催化裂解的研究，不仅可以为实验提供理论指导，还能帮助研究人员在优化工艺条件方面取得突破，提高氨气催化裂解的效率和产物的纯度。随着模拟技术的发展和计算能力的提升，基于COMSOL的氨气催化裂解模拟研究将会在化学工程领域发挥越来越大的作用。
STOP2模式RTC唤醒失准根源锁定：电源域切换时序偏差+LSI频率漂移+备份域寄存器锁存延迟——通过CubeMX配置反推+HAL源码注释溯源+示波器捕获LSE起振全过程
# STOP2模式RTC唤醒失准问题的深度解构与鲁棒性保障体系
在工业物联网边缘节点、电池供电传感器、智能穿戴设备等对功耗与时间精度双敏感的应用场景中，STM32L4/L5系列微控制器凭借其亚毫安级STOP2待机电流（典型1.2 µA @ 3.3 V, 25°C）成为主流选择。然而，一个长期被低估却反复在量产阶段爆发的问题正悄然侵蚀着系统可靠性：**STOP2模式下RTC唤醒定时存在稳定、可复现、且与环境强相关的系统性偏差——实测达±8.7ms（标称1秒周期），低温下甚至突破±45ms阈值，直接导致边缘AI推理任务漏触发、数据采集断点、安全看门狗超时，最终引发客户批量退货。**
这不是偶发