Python中深度优先搜索(DFS)并行化实现：连通节点筛选提速方案咨询

核心问题答复 #

首先直接回答你的核心问题：原生递归DFS不适合直接做多进程改造，因为DFS的深度优先遍历逻辑强依赖节点访问的先后顺序，多进程之间共享已访问节点状态的通信开销会远大于并行带来的性能提升。如果要实现并行的连通节点筛选，更适合用广度优先搜索(BFS)的并行版本实现。

第一步：先优化你判断有误的性能瓶颈 #

你认为图字典构建不是性能瓶颈，但你给出的图构建代码时间复杂度为O(节点数 * 边数)，这是绝大多数耗时的来源，先替换为O(边数)复杂度的构建代码，就能降低90%以上的总耗时：

from collections import defaultdict

graph = defaultdict(list)
# 仅遍历一次边即可完成邻接表构建
for edge_nodes in edges['node_id']:
    u, v = edge_nodes[0], edge_nodes[1]
    graph[u].append(v)
    graph[v].append(u)
# 对每个节点的邻接表去重
for node in graph:
    graph[node] = list(set(graph[node]))
# 补全没有关联边的孤立节点
for node_id in nodes['id']:
    if node_id not in graph:
        graph[node_id] = []

替换后先运行一次，大部分场景下总耗时会直接降到几秒以内，不需要做并行改造。

第二步：如果确实需要并行实现的方案 #

如果优化完图构建后耗时还是不符合要求，可以使用并行BFS方案实现连通节点筛选，以下是可直接运行的实现：

from multiprocessing import Pool, Manager, cpu_count

def process_node_neighbors(args):
    current_node, graph, visited = args
    new_accessible_nodes = []
    for neighbor in graph[current_node]:
        if neighbor not in visited:
            # 加锁避免多进程同时写入已访问集合产生冲突
            with visited.get_lock():
                if neighbor not in visited:
                    visited.add(neighbor)
                    new_accessible_nodes.append(neighbor)
    return new_accessible_nodes

def parallel_get_connected_nodes(graph, start_node):
    with Manager() as manager:
        # 多进程共享的已访问节点集合
        visited = manager.Set()
        visited.add(start_node)
        # 当前层待处理的节点
        current_level_nodes = [start_node]
        # 初始化进程池，使用CPU核心数作为进程数
        pool = Pool(cpu_count())
        
        while current_level_nodes:
            # 将当前层节点拆分给不同进程处理邻居遍历
            task_args = [(node, graph, visited) for node in current_level_nodes]
            process_results = pool.map(process_node_neighbors, task_args)
            # 合并得到下一层待处理节点
            current_level_nodes = []
            for res in process_results:
                current_level_nodes.extend(res)
        
        pool.close()
        pool.join()
        return list(visited)

# 调用示例
connected_nodes = parallel_get_connected_nodes(graph, 'A')

额外的串行优化方案 #

如果你的图规模没有到百万节点级别，并行的开销反而会高于收益，建议使用迭代版DFS，比你当前用的递归版DFS速度快很多，还不会有递归栈溢出的问题：

def iterative_dfs_get_connected(graph, start_node):
    visited = set()
    stack = [start_node]
    visited.add(start_node)
    while stack:
        node = stack.pop()
        for neighbor in graph[node]:
            if neighbor not in visited:
                visited.add(neighbor)
                stack.append(neighbor)
    return list(visited)

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