基于Python multiprocessing实现多进程进度条的问题

针对并行仿真中进度条覆盖、无法给进度处理函数传参的问题，可通过闭包封装进度条实例+固定进度条行位置解决，具体实现如下：

核心思路 #

1. 用闭包让进度处理函数直接访问当前进程的进度条，避免全局共享变量的问题
2. 给每个任务的进度条分配唯一position，确保不同任务的进度条在命令行独立显示，不互相覆盖

修正后的代码 #

import sw
import multiprocessing
from tqdm import tqdm
import time

PARALLEL_RUNS = 2

def run_simulation(file, position):
    # 初始化当前任务的进度条，position固定行号，desc显示任务名称
    pbar = tqdm(total=100, position=position, desc=f"[{file}]")
    last_progress = 0

    # 闭包形式的进度处理函数，直接访问当前进程的pbar和last_progress
    def progressHandlerFn(model_name, progress):
        nonlocal last_progress
        # 根据FEA软件返回的progress值，计算当前进度百分比（需根据实际情况调整）
        current_progress = int(progress * 100)
        delta = current_progress - last_progress
        if delta > 0:
            pbar.update(delta)
            last_progress = current_progress
        pbar.refresh()  # 强制刷新，避免并行环境下输出延迟

    m = sw.Load(file)
    m.progressHanlderFunction = progressHandlerFn
    try:
        m.Run()
    finally:
        pbar.close()  # 确保进度条关闭，清理命令行

def run_simulations(files, parallel_runs):
    # 为每个文件分配唯一的进度条行位置
    tasks = [(file, idx) for idx, file in enumerate(files)]
    with multiprocessing.Pool(parallel_runs) as pool:
        pool.starmap(run_simulation, tasks)

if __name__ == "__main__":
    model_files = ["model1.ext", "model2.ext"]  # 替换为你的仿真文件列表
    run_simulations(model_files, PARALLEL_RUNS)

关键细节说明 #

• 闭包传参：progressHandlerFn作为run_simulation的内部函数，无需额外传参即可访问当前进程的pbar和last_progress，完美适配FEA包装器固定参数的要求。
• 固定位置避免覆盖：position=idx让每个任务的进度条占据命令行的固定行，多个并行任务的进度条独立显示，不会互相覆盖。
• 资源清理：finally块确保无论仿真是否正常结束，都会关闭进度条，避免命令行残留无效输出。
• 进度计算适配：需根据FEA软件返回的progress实际含义调整current_progress的计算逻辑（例如原代码中的时间比例计算）。

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