Python嵌套并行多进程实现问题求助（Windows环境）

需要处理大型数据集，目标是实现funB实例并行运行，且每个funB内部的funA实例也能并行执行。原始代码结构如下：

def funA(var1, var2, var3):
    # 业务代码

def funB(varB):
    # 业务代码
    
    funA(x1, x2, x3)
    funA(y1, y2, y3)

    # 业务代码

# 其他代码

funB(z1)
funB(z2)

# 其他代码

失败的尝试方案 #

尝试1：multiprocessing.Pool #

使用multiprocessing.Pool运行后出现Windows系统错误，代码如下：

from multiprocessing import Pool

def funA(var1, var2, var3):
    # 业务代码

def funB(varB):
    # 业务代码
    
    pool1 = Pool()
    pool1.apply_async(funA, [x1, x2, x3])
    pool1.apply_async(funA, [y1, y2, y3])

    # 业务代码

# 其他代码

pool0 = Pool()
pool0.map(funB, [z1, z2])

# 其他代码

尝试2：pathos的ProcessingPool和ThreadingPool #

同样出现Windows系统错误，代码如下：

from pathos.multiprocessing import ProcessingPool, ThreadingPool
amap = ProcessingPool().amap
tmap = ThreadingPool().map

def funA(var1, var2, var3):
    # 业务代码

def funB(varB):
    # 业务代码
    
    tmap(funA, [x1, x2, x3])
    tmap(funA, [y1, y2, y3])

    # 业务代码

# 其他代码

amap(funB, [z1, z2])

# 其他代码

尝试3：concurrent.futures.ThreadPoolExecutor #

程序不崩溃，但目标函数未执行，即使添加shutdown(wait=True)也无效，直接跳转到后续代码，代码如下：

from concurrent.futures import ThreadPoolExecutor
Pool1 = ThreadPoolExecutor(max_workers=2)
Pool2 = ThreadPoolExecutor(max_workers=2)

def funA(var1, var2, var3):
    # 业务代码

def funB(varB):
    # 业务代码
    
    Pool2.submit(funA, x1, x2, x3)
    Pool2.submit(funA, y1, y2, y3)
    Pool2.shutdown(wait=True)

    # 业务代码

# 其他代码

Pool1.submit(funB, [z1, z2])
Pool1.shutdown(wait=True)

# 其他代码

环境信息 #

• 系统：Windows 11
• Python版本：3.11（Anaconda环境）

正确实现方式 #

针对Windows系统的多进程嵌套问题，核心要注意两点：

1. Windows下多进程必须通过if __name__ == '__main__':包裹主逻辑，避免子进程重复初始化代码
2. 嵌套多进程/线程时，用上下文管理器自动管理池的生命周期，避免资源泄漏或死锁

推荐使用concurrent.futures的ProcessPoolExecutor实现外层funB的并行，内层funA根据任务类型选择ThreadPoolExecutor（IO密集型）或ProcessPoolExecutor（CPU密集型）。可行代码如下：

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

def funA(var1, var2, var3):
    # 业务代码示例
    print(f"Running funA with params: {var1}, {var2}, {var3}")
    # 模拟耗时操作
    import time
    time.sleep(1)
    return f"Result from funA: {var1}"

def funB(varB):
    # 业务代码示例
    print(f"Running funB with param: {varB}")
    
    # 内层用ThreadPoolExecutor（IO密集型），CPU密集型换ProcessPoolExecutor
    with ThreadPoolExecutor(max_workers=2) as pool_inner:
        # 提交任务并等待完成
        future1 = pool_inner.submit(funA, f"{varB}_x1", f"{varB}_x2", f"{varB}_x3")
        future2 = pool_inner.submit(funA, f"{varB}_y1", f"{varB}_y2", f"{varB}_y3")
        # 获取结果（不需要结果可省略，但必须等待任务完成）
        res1 = future1.result()
        res2 = future2.result()
        print(f"funB {varB} got results: {res1}, {res2}")
    
    # 业务代码示例
    print(f"funB {varB} finished")
    return f"Result from funB: {varB}"

if __name__ == '__main__':
    # 外层用ProcessPoolExecutor实现funB并行
    with ProcessPoolExecutor(max_workers=2) as pool_outer:
        futures = [pool_outer.submit(funB, z) for z in ["z1", "z2"]]
        # 等待所有任务完成并获取结果
        for future in futures:
            res = future.result()
            print(res)
    print("All tasks finished")

关键说明 #

• if __name__ == '__main__':：Windows无fork机制，子进程会重新导入脚本，此判断避免子进程重复执行主逻辑
• 上下文管理器（with语句）：自动处理池的创建与关闭，替代手动shutdown，减少错误
• 内层池选择：IO密集型任务用线程池更高效，CPU密集型用进程池避免GIL限制
• result()方法：必须调用才能确保任务执行完成，否则池可能提前关闭导致任务被丢弃

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