Python多进程下跨目录动态加载模块报错：ModuleNotFoundError

阿华AIGC实验室

2026-5-22

解决Python多进程下动态加载不同目录模块的问题

你碰到的问题核心是多进程环境中子进程的模块搜索路径不包含目标目录，而且主进程动态加载的模块不会自动被子进程复用。我来拆解原因和对应的解决方案：

问题根源

当你用imp.load_source在主进程里加载D:\pool\test\user.py时，这个模块只存在于主进程的内存中。Python的多进程是通过创建独立的子进程解释器实现的，子进程不会继承主进程中动态加载的模块，而且默认的sys.path里没有D:\pool\test这个目录——这就是子进程尝试导入user模块时抛出ModuleNotFoundError的原因。

另外提一句：imp模块在Python 3.4之后已经被官方废弃，推荐用importlib来做动态模块加载，下面的方案都会优先用更现代的写法。

解决方案

方案1：将模块目录添加到全局`sys.path`（最简便）

直接把user.py所在的目录加入sys.path，这样主进程和子进程都能通过常规import找到模块：

import multiprocessing as mp
import sys
from pathlib import Path

# 用原始字符串避免Windows路径的转义问题
module_dir = r"D:\pool\test"
sys.path.append(module_dir)

# 现在可以直接导入目标函数
from user import userfun

class Test():
    def __init__(self, pool):
        pool.processes = 1

vec = []
for i in range(10):
    vec.append([userfun, i])

# 后续的多进程池操作示例
# def worker(args):
#     func, num = args
#     return func() + num
# results = pool.map(worker, vec)

方案2：用`importlib`动态加载并注册模块

如果必须动态加载（比如路径是运行时确定的），可以用importlib加载模块并注册到sys.modules，同时确保子进程能找到路径：

import multiprocessing as mp
import importlib.util
import sys
from pathlib import Path

module_path = Path(r"D:\pool\test\user.py")
# 创建模块规格
spec = importlib.util.spec_from_file_location("user", module_path)
# 创建空模块对象
user_module = importlib.util.module_from_spec(spec)
# 注册到sys.modules，让解释器识别这个模块
sys.modules["user"] = user_module
# 执行模块内容
spec.loader.exec_module(user_module)
# 获取目标函数
userfun = user_module.userfun

# 关键：把模块所在目录加入sys.path，确保子进程能找到
sys.path.append(module_path.parent)

class Test():
    def __init__(self, pool):
        pool.processes = 1

vec = []
for i in range(10):
    vec.append([userfun, i])

# 后续多进程操作

方案3：通过进程池初始化函数配置子进程

如果需要统一配置所有子进程的环境，可以用Pool的initializer参数，在每个子进程启动时自动添加模块路径：

import multiprocessing as mp
import sys

# 子进程启动时执行的初始化函数
def init_worker():
    # 给子进程的sys.path添加模块目录
    sys.path.append(r"D:\pool\test")

class Test():
    def __init__(self, pool):
        pool.processes = 1

# 创建进程池时指定初始化函数
pool = mp.Pool(processes=1, initializer=init_worker)

# 现在可以在主进程和子进程中正常导入
from user import userfun

vec = []
for i in range(10):
    vec.append([userfun, i])

# 后续多进程操作