Python multiprocessing多输入多输出并行场景适用什么函数？

问题对应解决思路 #

• 多输入参数问题：使用Pool.starmap()替代Pool.map()，该接口支持传入由多参数元组组成的可迭代对象，自动解包元组作为目标函数的入参，无需将输入合并为单个可迭代变量。
• 多输出问题：不存在mp.Process()仅支持单变量输出的限制，若使用进程池方案，目标函数直接返回包含4个geodataframe列表的元组即可，进程池会自动处理进程间的返回值传递。

代码示例 #

1. 依赖导入与任务函数定义 #

import multiprocessing as mp
import geopandas as gpd
from typing import List, Tuple

# 自定义并行任务函数，支持多入参，直接返回多值元组
def your_parallel_task(arg1: str, arg2: int, arg3: gpd.GeoDataFrame) -> Tuple[List[gpd.GeoDataFrame], List[gpd.GeoDataFrame], List[gpd.GeoDataFrame], List[gpd.GeoDataFrame]]:
    # 此处写入你的业务处理逻辑
    gdf_list_1 = []
    gdf_list_2 = []
    gdf_list_3 = []
    gdf_list_4 = []
    # 业务处理逻辑省略
    return (gdf_list_1, gdf_list_2, gdf_list_3, gdf_list_4)

2. 主进程调用逻辑 #

if __name__ == "__main__":
    # 组装多组任务参数，每组参数为一个元组，对应一次任务的所有入参
    task_param_list = [
        (arg1_val1, arg2_val1, arg3_val1),
        (arg1_val2, arg2_val2, arg3_val2),
        # 剩余所有任务的参数依次添加
    ]

    # 初始化进程池，processes参数可自定义并发数，默认等于CPU核心数
    with mp.Pool(processes=4) as pool:
        # starmap自动解包每个参数元组传入目标函数，返回所有任务的结果列表
        task_results = pool.starmap(your_parallel_task, task_param_list)
    
    # 拆分汇总所有返回结果
    total_gdf_list1 = []
    total_gdf_list2 = []
    total_gdf_list3 = []
    total_gdf_list4 = []
    for res in task_results:
        g1, g2, g3, g4 = res
        total_gdf_list1.extend(g1)
        total_gdf_list2.extend(g2)
        total_gdf_list3.extend(g3)
        total_gdf_list4.extend(g4)

补充说明 #

geopandas对象支持pickle序列化，可直接在进程间传递，无需额外做序列化转换。如果存在部分固定参数不需要随任务变化，可结合functools.partial绑定固定参数，进一步简化参数组装逻辑。

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