如何在Python多进程中为函数传入多个参数

问题场景 #

开发项目时需实现多进程功能，目标函数processDFStandardCurve需要多个参数，其中仅siteID为每个进程的变量，所有siteID存储在siteIDList中。使用p.map()时程序无响应，改用p.starmap()则抛出错误：

TypeError: starmap() takes from 3 to 4 positional arguments but 9 were given

用户代码如下：

multiprocess函数 #

def multiprocess (cursor, testsDict, optionsDict, outputPath, calculated_pdf, stationToPriority, result):
    siteIDList = []
    for line in result:
        siteID = line[3]
        nbsNum = line[2]
        if siteID != "":
            siteIDList.append(siteID)
        else:
            siteIDList.append(nbsNum)
    p = multiprocessing.Pool()
    p.starmap(processDFStandardCurve, siteIDList,  cursor, testsDict, optionsDict, outputPath, calculated_pdf, stationToPriority)

processDFStandardCurve函数定义 #

def processDFStandardCurve(cursor, siteID, testsDict, optionsDict, outputPath, calculated_pdf, stationToPriority):
    # 函数实现
    pass

错误原因 #

starmap()的参数格式使用错误：该方法仅接受3个必填参数（池对象、目标函数、参数元组的可迭代对象），可选参数chunksize。你直接将siteIDList和其他固定参数作为独立参数传入，导致参数数量超出方法定义，触发类型错误。

同时，map()无响应是因为map()要求目标函数仅接受单个参数，而你的函数需要多个参数，参数不匹配导致程序异常阻塞。

解决方案 #

方法1：使用starmap+参数元组列表 #

将固定参数与每个siteID打包成元组，组成参数元组列表后传入starmap()：

def multiprocess (cursor, testsDict, optionsDict, outputPath, calculated_pdf, stationToPriority, result):
    siteIDList = []
    for line in result:
        siteID = line[3]
        nbsNum = line[2]
        if siteID != "":
            siteIDList.append(siteID)
        else:
            siteIDList.append(nbsNum)
    
    # 构建每个进程的参数元组，顺序与目标函数参数一致
    task_args = [
        (cursor, site_id, testsDict, optionsDict, outputPath, calculated_pdf, stationToPriority)
        for site_id in siteIDList
    ]
    
    p = multiprocessing.Pool()
    p.starmap(processDFStandardCurve, task_args)
    # 关闭进程池并等待所有任务完成
    p.close()
    p.join()

starmap()会自动将每个元组内的元素按顺序解包，作为目标函数的参数传入，完美匹配processDFStandardCurve的参数需求。

方法2：使用functools.partial绑定固定参数 #

通过partial将固定参数绑定到目标函数，生成仅需siteID作为参数的新函数，再用map()执行：

from functools import partial

def multiprocess (cursor, testsDict, optionsDict, outputPath, calculated_pdf, stationToPriority, result):
    siteIDList = []
    for line in result:
        siteID = line[3]
        nbsNum = line[2]
        if siteID != "":
            siteIDList.append(siteID)
        else:
            siteIDList.append(nbsNum)
    
    # 绑定固定参数，生成单参数函数
    partial_func = partial(
        processDFStandardCurve,
        cursor=cursor,
        testsDict=testsDict,
        optionsDict=optionsDict,
        outputPath=outputPath,
        calculated_pdf=calculated_pdf,
        stationToPriority=stationToPriority
    )
    
    p = multiprocessing.Pool()
    p.map(partial_func, siteIDList)
    p.close()
    p.join()

两种方法均可解决问题，可根据个人习惯选择。

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