Python中cProfile累计时间与查询执行时间差异大的原因及优化建议

你的cProfile统计范围没有覆盖查询执行的全部耗时逻辑，这是二者差距显著的核心原因：

从代码来看，pr.enable()是在以下耗时操作之后才调用的：

• 解析用户传入的查询时间窗口并转换为时间戳
• 遍历全部2万条航班，为每条航班计算并添加start_timestamp和end_timestamp字段
• 过滤出时间窗口内的航班生成flights_filtered

这些步骤的CPU时间完全没被cProfile统计到，但它们都包含在start_time = time.time()和end_time = time.time()的计时范围内。

另外，cProfile统计的是进程占用的CPU时间，而time.time()获取的是墙钟时间（真实流逝的时间）：如果程序运行期间有其他进程抢占CPU，或者存在IO等待（比如后续从数据库读取数据时），墙钟时间也会比CPU时间更长，但当前场景下主要原因还是统计范围不全。

要验证这一点，只需把pr.enable()移到find_routes函数的开头（或者至少移到start_time = time.time()之前），就能看到cProfile的累计时间会接近实际执行时间。

针对当前的BFS路线查询逻辑，结合你的数据规模（2万条航班，日均4000条），可以从以下几个方向优化：

一、预处理数据，避免重复计算 #

• 提前计算时间戳：在CSV读取阶段（而非每次查询时）就把start_timestamp和end_timestamp计算好并存入航班字典，避免每次查询都遍历2万条航班重复计算。
• 预先生成分组排序的航班映射：提前构建origin->destination->sorted_flights的映射，按start_timestamp排序好，每次查询直接复用，不用每次都重新分组排序。
• 预存机场连通性：当前的connectivity_map已经是预处理的，但可以考虑把它和航班数据绑定，比如在预处理时就确保连通性包含所有有效航线。

二、优化BFS过程的效率 #

1. 优化已访问机场的计算
   当前每次循环都通过{flight['origin'] for flight in path}重新生成已访问机场集合，对于长路径会重复计算。可以在路径中携带已访问集合的副本，避免重复生成：

# 初始化队列时携带已访问集合
to_explore = deque(
    (
        [flight], 
        {flight['origin'], flight['destination']}
    ) 
    for dest_flights in next_flights_map[origin].values() 
    for flight in dest_flights
)
# 循环时直接复用已访问集合
while to_explore:
    path, visited = to_explore.popleft()
    last_flight = path[-1]
    # 后续逻辑中使用visited，新增机场时直接复制集合
    new_visited = visited.copy()
    new_visited.add(next_flight['destination'])

2. 二分查找跳过不符合时间条件的航班
   由于每个origin->destination的航班列表已经按start_timestamp排序，可以用二分查找找到第一个满足next_flight['start_timestamp'] > last_flight['end_timestamp']的航班，之后的航班都符合条件，无需逐个遍历检查：

import bisect

# 预处理时：
next_flights_map = defaultdict(lambda: defaultdict(dict))
for flight in flights:
    origin = flight['origin']
    dest = flight['destination']
    next_flights_map[origin][dest].setdefault('flights', []).append(flight)
    next_flights_map[origin][dest].setdefault('timestamps', []).append(flight['start_timestamp'])
# 对每个航线的航班和时间戳同步排序
for origin_data in next_flights_map.values():
    for dest_data in origin_data.values():
        sorted_pairs = sorted(zip(dest_data['timestamps'], dest_data['flights']))
        dest_data['timestamps'], dest_data['flights'] = zip(*sorted_pairs)

# 查询时使用二分查找：
dest_data = next_flights_map[last_flight['destination']].get(dest, None)
if dest_data:
    timestamps = dest_data['timestamps']
    flights_list = dest_data['flights']
    # 找到第一个大于last_flight['end_timestamp']的索引
    idx = bisect.bisect_right(timestamps, last_flight['end_timestamp'])
    # 直接遍历idx之后的所有符合时间条件的航班
    for next_flight in flights_list[idx:]:
        new_path = path + [next_flight]
        to_explore.append(new_path)

3. 减少路径拷贝开销
   当前new_path = path + [next_flight]会创建新的列表副本，路径越长拷贝开销越大。可以改用链表结构，或者只存储航班的索引（而非完整字典）来减少内存拷贝。

三、数据库迁移后的优化 #

当后续从数据库读取数据时：

• 利用数据库的查询能力过滤时间窗口内的航班，只查询符合条件的航班，避免在Python中处理全部2万条数据。
• 对origin、destination、start_datetime字段建立索引，加速数据库查询。
• 考虑用数据库的存储过程或窗口函数预处理部分逻辑，减少Python端的计算压力。

四、其他优化点 #

• 保留当前路径长度不超过4段的限制，避免无限遍历。
• 如果valid_paths会生成大量结果，考虑用生成器逐步返回结果，而非一次性存储所有路径，减少内存占用。
• 若有多个查询请求，可考虑用线程池或进程池并行处理，注意数据共享的线程安全问题。

内容的提问来源于stack exchange，提问作者terrabl