multiprocessing.Process运行tqdm出现异常输出的解决方案

多进程场景下每个进程的tqdm实例独立操作终端光标、执行输出刷新，进程间没有同步写入机制，多个进程同时抢占终端输出时，就会出现已完成进度条残留、行位置错乱的问题。原代码中给每个子进程的tqdm单独传nrows参数无效，该参数仅在单进程场景下生效，跨进程不共享终端行占用状态。

核心修改点：

• 引入跨进程输出锁，保证同一时间只有一个进程执行tqdm刷新，避免写入冲突
• 主进程统一为每个子进程分配固定的进度条行位置，子进程不单独设置nrows参数
• 子进程执行完成后主动关闭tqdm实例
  注意：该方案要求终端窗口高度大于进程数+2，32进程场景下对终端尺寸要求较高，优先选方案2

可运行示例代码：

import multiprocessing
from tqdm import tqdm
import time
import random

def cluster(indexes, process_n, lock):
    # 传入跨进程锁，tqdm内部刷新时自动加锁
    pbar = tqdm(
        indexes,
        position=process_n + 1,  # 第0行预留，避免进度条顶到终端顶部
        desc=f'Process {process_n}',
        leave=True,  # 进程完成后保留进度条记录
        dynamic_ncols=True,  # 自动适配终端宽度
        lock=lock
    )
    for index in pbar:
        # 替换为你的实际业务逻辑
        time.sleep(random.uniform(0.05, 0.15))
    pbar.close()

if __name__ == '__main__':
    N_CORES = 8  # 32进程场景修改为32，需保证终端高度足够
    tasks_per_process = 127
    processes = []
    # 初始化跨进程输出锁
    output_lock = multiprocessing.Lock()

    tqdm.write(f"Starting {N_CORES} processes...")

    for process_n in range(N_CORES):
        indexes = range(tasks_per_process)
        p = multiprocessing.Process(
            target=cluster,
            args=(indexes, process_n, output_lock)
        )
        p.start()
        processes.append(p)
    
    for p in processes:
        p.join()
    
    # 所有任务完成后将光标移到所有进度条下方，避免shell提示符错位
    print(f"\n" * (N_CORES + 1))

32个独立进度条会占满整个终端屏幕，可读性极差。全局进度条仅占用1行终端，由主进程单独维护，子进程完全不操作终端输出，从根源上避免多进程写入冲突，稳定性极高。
核心实现逻辑：用跨进程原子计数器统计总完成任务数，主进程轮询计数器值刷新进度。

可运行示例代码：

import multiprocessing
from tqdm import tqdm
import time
import random

def cluster(indexes, counter, lock):
    for index in indexes:
        # 替换为你的实际业务逻辑
        time.sleep(random.uniform(0.05, 0.15))
        # 任务完成后原子更新计数器
        with lock:
            counter.value += 1

if __name__ == '__main__':
    N_CORES = 32
    tasks_per_process = 127
    total_tasks = N_CORES * tasks_per_process

    # 初始化跨进程计数器与锁
    counter = multiprocessing.Value('i', 0)
    counter_lock = multiprocessing.Lock()
    processes = []

    # 主进程初始化全局进度条
    pbar = tqdm(total=total_tasks, desc='Total progress', dynamic_ncols=True)

    for process_n in range(N_CORES):
        indexes = range(tasks_per_process)
        p = multiprocessing.Process(
            target=cluster,
            args=(indexes, counter, counter_lock)
        )
        p.start()
        processes.append(p)
    
    # 轮询更新进度
    while counter.value < total_tasks:
        pbar.n = counter.value
        pbar.refresh()
        time.sleep(0.1)
    
    for p in processes:
        p.join()
    
    pbar.close()
    print("\nAll tasks completed!")

• 所有多进程启动代码必须放在if __name__ == '__main__':块内，Windows/macOS默认spawn进程启动模式下，不这么写会触发进程递归启动报错
• 不要在子进程中直接调用print()输出内容，需要打日志请用tqdm.write("日志内容")，否则会打乱进度条显示位置
• 如果任务量拆分不均匀，全局进度条方案依然可以准确统计总进度，不受单个进程执行速度影响

内容的提问来源于stack exchange，提问作者Shehryar Ahmed Subhani