Python递归位置探索中ProcessPoolExecutor预创建进程池优化问询

我正在开发一个Python项目，需要通过递归方式探索不同位置，希望用ProcessPoolExecutor实现并行化。具体需求：

• 执行初期预创建最多20个子进程的进程池
• 递归任意深度仅额外使用2个进程
• 探索过程中根据递归深度按需分配这些预创建的进程

当前代码可正常运行，但每次递归都会动态创建进程池，导致开销过大，代码如下：

from concurrent.futures import ProcessPoolExecutor
import multiprocessing
import time

def explore(position, depth=0, max_depth=3):
    print(f"Process {multiprocessing.current_process().pid}: Exploring position {position} at depth {depth}")

    if depth < max_depth:
        new_positions = [
            (position[0] + 1, position[1]), 
            (position[0], position[1] + 1)
        ]

        with ProcessPoolExecutor(max_workers=len(new_positions)) as executor:
            futures = [executor.submit(explore, pos, depth + 1, max_depth) for pos in new_positions]

            for future in futures:
                future.result()

        print(f"Process {multiprocessing.current_process().pid}: Additional processing after future.result()")
        time.sleep(0.1)

    return position

if __name__ == "__main__":
    initial_position = (0, 0)

    start_time = time.time()
    with ProcessPoolExecutor(max_workers=1) as executor:
        future = executor.submit(explore, initial_position)
        result = future.result()

    print("Exploration complete.")
    print("Time taken:", time.time() - start_time)

解决方案 #

核心问题是每次递归都新建ProcessPoolExecutor实例，反复启动/销毁进程会带来巨大开销。正确做法是在顶层创建全局进程池，让所有递归任务复用预创建的进程。

修改后的代码 #

from concurrent.futures import ProcessPoolExecutor
import multiprocessing
import time

def explore(position, depth=0, max_depth=3, executor=None):
    print(f"Process {multiprocessing.current_process().pid}: Exploring position {position} at depth {depth}")

    if depth < max_depth:
        new_positions = [
            (position[0] + 1, position[1]), 
            (position[0], position[1] + 1)
        ]

        # 复用全局进程池提交任务，不再创建新池
        futures = [executor.submit(explore, pos, depth + 1, max_depth, executor) for pos in new_positions]

        for future in futures:
            future.result()

        print(f"Process {multiprocessing.current_process().pid}: Additional processing after future.result()")
        time.sleep(0.1)

    return position

if __name__ == "__main__":
    initial_position = (0, 0)

    start_time = time.time()
    # 预创建最多20个进程的全局进程池
    with ProcessPoolExecutor(max_workers=20) as executor:
        future = executor.submit(explore, initial_position, executor=executor)
        result = future.result()

    print("Exploration complete.")
    print("Time taken:", time.time() - start_time)

关键修改点 #

1. 全局进程池复用：在主进程中创建max_workers=20的进程池，所有递归任务都通过这个池提交，彻底消除动态创建进程的开销
2. 传递进程池实例：将executor作为参数传入递归函数，确保所有层级都能访问到同一个进程池
3. 控制并发数：每个递归节点提交2个新任务，刚好符合你“递归任意深度仅额外使用2个进程”的需求，进程池会自动调度任务到预创建的进程中

内容的提问来源于stack exchange，提问作者Gabriel Díaz