如何在Python类函数中最简并行化百万最短路径循环？

针对你在类函数内并行处理百万条最短路径的需求，核心解决思路是：多进程无法直接共享字典，因此让每个进程独立处理部分路径并生成子字典，最后在主进程合并所有子字典。以下是两种最简清晰的实现方式：

方法一：使用concurrent.futures.ProcessPoolExecutor（推荐） #

这个模块语法更简洁，支持在类函数内直接调用，无需额外的全局函数或__main__特殊处理：

from concurrent.futures import ProcessPoolExecutor

class YourClass:
    def your_function(self, shortest_paths, demand):
        some_dict = {}
        
        # 定义局部函数，处理单条路径的节点对统计
        def process_single_path(args):
            j, paths = args
            sub_dict = {}
            if len(paths) > 1:
                for i in range(len(paths) - 1):
                    edge = (paths[i], paths[i + 1])
                    sub_dict[edge] = sub_dict.get(edge, 0) + demand[j]
            return sub_dict
        
        # 并行处理所有路径，收集子字典结果
        with ProcessPoolExecutor() as executor:
            results = executor.map(process_single_path, enumerate(shortest_paths))
        
        # 合并所有子字典到主字典
        for sub_dict in results:
            for edge, val in sub_dict.items():
                some_dict[edge] = some_dict.get(edge, 0) + val
        
        return some_dict

方法二：使用multiprocessing.Pool配合局部函数 #

若更习惯multiprocessing模块，可通过局部函数+参数打包的方式实现，无需全局函数：

from multiprocessing import Pool

class YourClass:
    def your_function(self, shortest_paths, demand):
        some_dict = {}
        
        def process_single_path(args):
            j, paths = args
            sub_dict = {}
            if len(paths) > 1:
                for i in range(len(paths) - 1):
                    edge = (paths[i], paths[i + 1])
                    sub_dict[edge] = sub_dict.get(edge, 0) + demand[j]
            return sub_dict
        
        # 创建进程池并处理任务
        with Pool() as pool:
            results = pool.map(process_single_path, enumerate(shortest_paths))
        
        # 合并子字典
        for sub_dict in results:
            for edge, val in sub_dict.items():
                some_dict[edge] = some_dict.get(edge, 0) + val
        
        return some_dict

优化建议（针对百万级路径） #

如果单条路径的处理逻辑很轻量，可将路径批量分组处理（比如每1000条为一组），减少进程间通信的开销：

def process_batch(batch):
    sub_dict = {}
    for j, paths in batch:
        if len(paths) > 1:
            for i in range(len(paths) - 1):
                edge = (paths[i], paths[i + 1])
                sub_dict[edge] = sub_dict.get(edge, 0) + demand[j]
    return sub_dict

# 拆分路径为批量
batch_size = 1000
batches = []
for i in range(0, len(shortest_paths), batch_size):
    batches.append(list(enumerate(shortest_paths[i:i+batch_size])))

# 并行处理批量任务
with ProcessPoolExecutor() as executor:
    results = executor.map(process_batch, batches)

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