多进程中避免进度条覆盖、实现多进度条显示的问题求助

嘿，这个多进程进度条互相覆盖的问题我之前踩过坑！本质原因是所有进程都在抢占同一个终端输出流，没有做同步或者隔离，导致进度条的光标位置混乱。给你几个实用的解决方案，按简便程度排序：

方案1：给每个进程分配独立的进度条行（最省心） #

用tqdm的position参数，给每个进程的进度条指定一个专属的终端行位置，这样它们就会垂直排列，互不干扰。同时加上进程ID作为描述，能清晰区分每个进程的进度。

示例代码：

from multiprocessing import Process
from tqdm import tqdm
import time
import random
import csv

def worker(process_id, task_count, result_queue):
    # 每个进程的进度条，position参数指定行号，desc标注进程ID
    pbar = tqdm(total=task_count, desc=f"[进程{process_id}]", position=process_id, leave=True)
    process_data = []
    for _ in range(task_count):
        # 模拟生成随机整数的任务
        rand_num = random.randint(1, 100)
        process_data.append([rand_num])
        time.sleep(0.05)  # 模拟任务耗时
        pbar.update(1)
    pbar.close()
    # 将进程数据发送到队列，供主进程汇总
    result_queue.put(process_data)

if __name__ == "__main__":
    total_trials = 30
    num_processes = 4
    tasks_per_process = total_trials // num_processes
    result_queue = []  # 或者用multiprocessing.Queue，这里简化用列表

    processes = []
    for i in range(num_processes):
        p = Process(target=worker, args=(i, tasks_per_process, result_queue))
        processes.append(p)
        p.start()

    # 等待所有进程完成，收集数据
    all_data = []
    for p in processes:
        p.join()
        all_data.extend(result_queue.pop())

    # 统一写入CSV
    with open("output.csv", "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerows(all_data)

这里的关键是position=process_id，每个进程用自己的ID作为行号，进度条就会乖乖呆在各自的行里，不会互相覆盖。leave=True会让进程完成后保留进度条，方便你查看最终状态。

方案2：主进程统一管理进度输出（更整洁） #

如果不想看到多个进度条，只想看整体进度或者统一的分进程状态，可以用进程间队列（multiprocessing.Queue）让工作进程把进度信息发给主进程，由主进程单独负责输出，彻底避免输出冲突。

示例代码：

from multiprocessing import Process, Queue
from tqdm import tqdm
import time
import random
import csv

def worker(process_id, task_count, data_queue, progress_queue):
    process_data = []
    for _ in range(task_count):
        rand_num = random.randint(1, 100)
        process_data.append([rand_num])
        time.sleep(0.05)
        # 发送进度更新给主进程
        progress_queue.put((process_id, 1))
    # 发送完成的数据
    data_queue.put(process_data)

def progress_watcher(total_trials, num_processes, progress_queue):
    # 记录每个进程的完成数
    process_progress = {i:0 for i in range(num_processes)}
    total_done = 0
    # 主进度条
    main_pbar = tqdm(total=total_trials, desc="整体完成进度")
    
    while total_done < total_trials:
        process_id, increment = progress_queue.get()
        process_progress[process_id] += increment
        total_done += increment
        main_pbar.update(increment)
        # 可选：打印每个进程的单独进度（用\r覆盖同一行）
        status = " | ".join([f"进程{i}: {process_progress[i]}/{total_trials//num_processes}" for i in range(num_processes)])
        print(f"\r{status}", end="", flush=True)
    
    main_pbar.close()
    print()  # 换行，避免后续输出覆盖状态

if __name__ == "__main__":
    total_trials = 30
    num_processes = 4
    tasks_per_process = total_trials // num_processes

    data_queue = Queue()
    progress_queue = Queue()

    # 启动进度监控进程
    watcher = Process(target=progress_watcher, args=(total_trials, num_processes, progress_queue))
    watcher.start()

    # 启动工作进程
    processes = []
    for i in range(num_processes):
        p = Process(target=worker, args=(i, tasks_per_process, data_queue, progress_queue))
        processes.append(p)
        p.start()

    # 收集所有数据
    all_data = []
    for _ in range(num_processes):
        all_data.extend(data_queue.get())

    # 等待所有进程结束
    for p in processes:
        p.join()
    watcher.join()

    # 写入CSV
    with open("output.csv", "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerows(all_data)

这个方案里所有输出都由主进程的监控器处理，不会有任何冲突，还能灵活定制进度展示方式，比如同时显示整体进度和每个进程的细节。

小提示 #

• 如果不用tqdm，自己写进度条的话，要确保每个进程的输出是原子操作，比如用print(f"[进程{pid}] 进度: {done}/{total}", flush=True)，并且尽量让每个进程的输出单独占一行，避免用\r覆盖同一行（除非你能精准控制行位置）。
• Windows系统下一定要把多进程启动的代码放在if __name__ == "__main__":块里，避免spawn模式下的重复初始化问题。

内容的提问来源于stack exchange，提问作者Pythonic Guy 21421