动态加载模块下的Python多进程调用问题求助

这个问题我之前也碰到过，核心矛盾就是多进程间传递已加载模块/函数会因为pickle序列化失败，而重复加载脚本又太浪费资源——不过好在你的脚本数量很少（2-10个），完全可以通过让子进程提前初始化加载所有脚本的方式解决，既避开序列化问题，又避免重复加载的开销。

核心思路 #

因为脚本总数很少，我们可以让每个子进程在启动时一次性加载所有需要的脚本模块并缓存起来，后续处理任务时只传递脚本名称和目录路径（这两个都是可序列化的基础类型），子进程直接从缓存中取出对应的模块调用函数即可。这样每个子进程只加载一次脚本，复用性拉满，也不会触发PicklingError。

具体实现代码 #

import importlib.util
import multiprocessing

# 子进程全局变量，用于缓存已加载的脚本模块
loaded_modules = {}

def init_worker(scripts_list):
    """子进程初始化函数：一次性加载所有脚本模块"""
    global loaded_modules
    for script_name, script_path in scripts_list:
        spec = importlib.util.spec_from_file_location(f"{script_name}.script", script_path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        loaded_modules[script_name] = module

def worker(task):
    """子进程任务函数：根据脚本名称调用对应模块的函数"""
    dir_path, script_name = task
    target_module = loaded_modules[script_name]
    return target_module.get_features(dir_path)

if __name__ == "__main__":
    # ----------------------
    # 替换成你自己的实际数据
    # ----------------------
    # 你的脚本列表：[(脚本名称, 脚本路径), ...]
    scripts = [("example", "./example.py"), ("another", "./another_script.py")]
    # 任务列表：把原来的(目录路径, 已加载模块)改成(目录路径, 脚本名称)
    work_list = [("/path/to/dir1", "example"), ("/path/to/dir2", "another"), ...]

    # 创建进程池，传入初始化函数和脚本列表参数
    with multiprocessing.Pool(6, initializer=init_worker, initargs=(scripts,)) as p:
        all_results = p.map(worker, work_list)

    # 合并结果到集合
    results_set = set(item for res_list in all_results for item in res_list)

为什么这个方案有效？ #

1. 避开序列化问题：传递给worker的只有字符串（脚本名称）和路径，都是Python原生支持pickle的类型，不会触发PicklingError。
2. 避免重复加载：每个子进程在启动时只加载一次所有2-10个脚本，后续处理所有对应任务时直接复用缓存的模块，彻底消除重复加载的开销。
3. 跨平台兼容：不管是Windows（spawn模式）还是Unix/Linux/macOS（fork模式），这个方案都能稳定工作，不会因为进程创建模式的差异出问题。

额外补充（针对Unix/Linux/macOS用户） #

如果你只在类Unix系统上运行，也可以利用默认的fork进程创建模式——fork出来的子进程会直接继承父进程已加载的模块，这时候你可以直接传递函数对象，不需要初始化函数：

import multiprocessing

def worker(task):
    dir_path, func = task
    return func(dir_path)

if __name__ == "__main__":
    # 这里work_list可以保留原来的(目录路径, 已加载模块.get_features)格式
    with multiprocessing.Pool(6) as p:
        all_results = p.map(worker, work_list)
    
    results_set = set(item for res_list in all_results for item in res_list)

不过这个方案只适用于fork模式，Windows系统不支持，所以还是推荐前面的通用初始化方案更稳妥。

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