如何使用Python multiprocessing在共享容器类内返回多个类实例

前置注意事项 #

• 首先将类的私有方法前缀从双下划线__改为单下划线_，双下划线的名称修饰机制会导致多进程序列化时无法找到对应方法，比如把__populatePatient改为_populatePatient、__buildPatientFromPatid改为_buildPatientFromPatid
• 不需要在当前类的方法内写__main__判断，只需要保证你整个程序最上层调用buildAllPatients的入口代码被if __name__ == "__main__":块包裹即可，跨平台（尤其是Windows）必须满足这个要求
• 确保你的Patient类、Patients容器类是可序列化（pickle）的，否则子进程无法将处理完的对象返回给主进程

改造后的代码 #

首先是buildPatientsFromPatid方法的多进程实现：

import multiprocessing
from functools import partial

def buildPatientsFromPatid(self, patidList, featureList) -> Patients:
    patients = Patients()
    # 固定方法的featureList参数，不需要每次迭代重复传
    process_func = partial(self._buildPatientFromPatid, featureList=featureList)
    
    # 开启36核进程池，用上下文管理器自动回收资源
    with multiprocessing.Pool(processes=36) as pool:
        # 50万数据量建议设置chunksize减少IPC开销，值可以根据实际运行效率调整
        chunksize = max(1, len(patidList) // (36 * 4))
        # 如果不需要保持patidList的顺序，用imap_unordered会比imap/map快30%以上
        for patient in pool.imap_unordered(process_func, patidList, chunksize=chunksize):
            patients.addPatient(patient)
    
    return patients

优化建议 #

• 如果你的类实例self包含非常大的数据集（比如完整的ioDF），序列化传递给子进程的开销很高，可以把只读的公共数据（比如ioDF）设置为全局变量，Linux/macOS默认的fork模式下子进程会共享这部分只读内存，不需要额外复制，大幅降低内存开销和启动耗时
• 如果_populatePatient内部需要查询每个patid对应的子DF，可以提前把patid和对应的indDF绑定成元组再传给进程池，避免子进程重复查询全量DF
• 如果单条patid的处理速度很快，建议调大chunksize，减少进程间通信的频率；如果单条处理速度很慢，可以调小chunksize，避免进程负载不均

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