如何优化Python可并行代码运行时长？多线程/进程提速无效问题

问题根源分析 #

• 多线程无效原因：Python的*全局解释器锁（GIL）*导致CPU密集型任务无法通过多线程实现真正并行，线程切换反而会增加额外开销，导致运行时间变长。
• 多进程未提速原因：
  1. 手动创建Process的启动开销较高，若单个分区的计算量不足以抵消进程启动和通信的开销，整体效率反而下降。
  2. 共享数据occlusions在多进程模式下是进程私有拷贝，修改无法同步到主进程，且进程间共享数据的同步操作会带来额外开销。
  3. 全局变量（如array_width、observer_z等）在多进程中会被重复拷贝，增加内存开销和初始化时间。

具体优化方案 #

1. 使用进程池替代手动创建进程 #

进程池可以复用已创建的进程，大幅减少进程启动和销毁的开销，推荐使用concurrent.futures.ProcessPoolExecutor或multiprocessing.Pool。

2. 避免共享数据，采用"计算+返回结果+合并"模式 #

让每个进程独立计算自己分区的结果，最后在主进程合并所有结果，彻底避免进程间数据同步的开销。

3. 优化任务粒度 #

确保每个进程处理的任务量足够大，抵消进程调度的开销。建议将进程数设置为CPU核心数（可通过os.cpu_count()获取），最大化利用硬件资源。

4. 传递必要参数，减少全局依赖 #

将全局变量（如array_width、observer_z等）作为参数传递给子进程函数，避免多进程下的全局变量拷贝问题。

5. 数据结构优化 #

为每个分区创建独立的结果存储，避免频繁的列表append操作，最后合并结果时直接赋值到目标数组，提升数据处理效率。

修改后的代码示例 #

from spatialmath import SE3
import numpy as np
from math import sqrt
import time
from concurrent.futures import ProcessPoolExecutor
import os

def dummy():
    '''模拟原函数运行时间'''
    time.sleep(0.5)
    return True

def transform_pose_to_a_different_coordinate_frame(pose):
    '''实际坐标转换函数，此处简化返回输入'''
    return pose

def check_occlusions_for_division(params):
    # 接收所有必要参数，避免全局依赖
    x_range, y_range, array_width, array_height, observer_z, observee_z = params
    
    # 为当前分区创建独立的结果存储
    partition_occlusions = {}
    for observer_x in x_range:
        for observer_y in y_range:
            observer_pose = SE3((observer_x, observer_y, observer_z))
            observer_pose = transform_pose_to_a_different_coordinate_frame(observer_pose)
            
            occluded_list = []
            for observee_x in range(array_width):
                for observee_y in range(array_height):
                    observee_pose = SE3(observee_x, observee_y, observee_z)
                    observee_pose = transform_pose_to_a_different_coordinate_frame(observee_pose)
                    occluded = dummy()
                    
                    if occluded:
                        occluded_list.append([observee_x, observee_y])
            
            partition_occlusions[(observer_x, observer_y)] = np.array(occluded_list)
    return partition_occlusions


if __name__ == "__main__":
    omap_res = 0.1
    array_width = 61
    array_height = 61
    observer_z = 3
    observee_z = 0.1

    # 进程数设置为CPU核心数，最大化利用硬件
    num_processes = os.cpu_count() or 4
    assert sqrt(num_processes) == int(sqrt(num_processes))

    # 拆分数组为多个分区
    num_x_divisions = int(sqrt(num_processes))
    num_y_divisions = int(sqrt(num_processes))
    x_div_size = array_width // num_x_divisions
    y_div_size = array_height // num_y_divisions
    task_params = []

    for i in range(num_x_divisions):
        for j in range(num_y_divisions):
            div_x_range = range(i*x_div_size, (i+1)*x_div_size if i!=num_x_divisions-1 else array_width)
            div_y_range = range(j*y_div_size, (j+1)*y_div_size if j!=num_y_divisions-1 else array_height)
            # 将所有必要参数打包成元组
            task_params.append((div_x_range, div_y_range, array_width, array_height, observer_z, observee_z))

    # 使用进程池执行并行任务
    start_time = time.time()
    occlusions = np.empty((array_width, array_height, 0)).tolist()
    
    with ProcessPoolExecutor(max_workers=num_processes) as executor:
        results = executor.map(check_occlusions_for_division, task_params)
    
    # 合并所有进程的结果
    for result in results:
        for (x, y), occluded_arr in result.items():
            occlusions[x][y] = occluded_arr
    
    end_time = time.time()
    print(f"并行运行时长: {end_time - start_time:.2f}秒")

额外优化建议 #

• 如果实际场景中的check_occlusion函数可以进行向量化改造（比如用numpy批量处理坐标计算），结合并行计算能获得更大的性能提升。
• 若任务规模极大，可考虑使用Dask等分布式并行框架处理分块数据，进一步优化资源利用。

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