Python嵌套循环下函数并行处理的实现问题咨询

并行化你的路径生成函数：问题分析与解决方案 #

你说得对，现在的写法完全没实现并行——pool.apply()本身是阻塞式的单任务调用，你把整个source_nodes和sink_nodes丢进去，其实就是让一个单独的进程跑完所有双重循环，和不用多进程的效率没差，甚至还多了进程启动的额外开销。

为什么当前代码没并行？ #

apply(func, args)的逻辑是：把args传给func，然后在一个进程里执行完整个函数再返回结果。它本来就不是用来批量处理并行任务的，你需要的是能把任务拆分后分配给多个进程的方法，比如map()或者starmap()。

正确的并行实现方式 #

我们可以把大任务拆成多个独立的小任务，让进程池里的多个进程同时处理。这里有两种常见的拆分思路：

思路1：拆分每个(source, sink)节点对为独立任务

把原来的双重循环拆成单个节点对的处理，这样每个节点对的路径生成都能并行执行：

首先改写函数，让它只处理一组source和sink：

def create_single_path(source, sink):
    # 直接返回单个节点对的路径列表
    return list(nx.all_simple_paths(Directed_G, source, sink))

然后生成所有节点对的组合，用starmap()来并行处理（starmap支持传入带多个参数的任务元组）：

import multiprocessing as mp
from itertools import product

source_nodes = [10413173, 10414530, 10414530, 10437199]
sink_nodes = [10420346, 10438770, 10438711, 10414530, 10436258]

# 生成所有source和sink的组合对
task_pairs = product(source_nodes, sink_nodes)

# 用with语句自动管理进程池
with mp.Pool() as pool:
    # 每个任务元组的元素会被作为参数传入create_single_path
    results = pool.starmap(create_single_path, task_pairs)

# 把所有进程返回的结果合并成一个大列表
path = [item for sublist in results for item in sublist]

思路2：按source节点拆分任务

如果不想拆成单个节点对，也可以让每个进程处理一个source节点对应的所有sink节点：

先改写函数处理单个source和所有sink：

def create_paths_for_source(source, sink_list):
    paths = []
    for sink in sink_list:
        paths.extend(list(nx.all_simple_paths(Directed_G, source, sink)))
    return paths

然后用functools.partial固定sink参数，再用map()并行处理每个source节点：

import multiprocessing as mp
from functools import partial

source_nodes = [10413173, 10414530, 10414530, 10437199]
sink_nodes = [10420346, 10438770, 10438711, 10414530, 10436258]

# 固定sink_list参数，让函数只接受source作为输入
partial_func = partial(create_paths_for_source, sink_list=sink_nodes)

with mp.Pool() as pool:
    # 每个source节点会被分配到不同进程处理
    results = pool.map(partial_func, source_nodes)

# 合并结果
path = [item for sublist in results for item in sublist]

额外注意事项 #

• 确保Directed_G在子进程中可正常访问：如果是大型图，尽量在子进程内重新加载图数据，避免进程间拷贝大对象拖慢性能。
• 进程池大小：默认mp.Pool()会使用CPU核心数，你也可以手动指定（比如mp.Pool(processes=4)）。
• 内存优化：如果路径结果很大，可以用imap()或imap_unordered()渐进式获取结果，避免一次性占用过多内存。

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