如何为多进程池中的每个任务显示独立进度条？

你当前的问题是多进程共享控制台输出流，导致tqdm进度条互相覆盖。要实现每个进程显示独立的进度条完全可行，核心是让每个进程的进度条拥有独立的输出位置或唯一标识，避免输出冲突。下面是具体的实现方案：

方案一：给每个进度条指定唯一的position参数 #

tqdm的position参数可以指定进度条在控制台的行位置，给每个进程的进度条分配不同的position值，就能让它们固定在不同行显示，不会互相覆盖。

修改你的代码如下：

import os
import multiprocessing as mp
import pandas as pd
from tqdm import tqdm

class OtherClass:
    def some_method(self, args):
        # 接收df和进程的位置索引
        df, pos = args
        # 给当前进程的进度条指定position，desc加上进程标识
        for index, row in tqdm(df.iterrows(), desc=f"进程{pos+1}进度", total=len(df), position=pos):
            # 这里写你的业务逻辑，比如处理row
            pass
        # 进度条完成后清空该行，避免后续输出干扰
        tqdm.write(f"进程{pos+1} 完成", end="\r")
        return "处理完成"

class App:
    def __init__(self):
        self.otherClass = OtherClass()
        self.n_cpus = os.cpu_count()
        # 模拟测试用的dataframes列表
        self.dataframes = [pd.DataFrame({'col': range(1000)}) for _ in range(self.n_cpus)]

    def call_multiprocessing(self):        
        # 把每个df和对应的位置索引打包成参数
        tasks = [(df, i) for i, df in enumerate(self.dataframes)]
        with mp.Pool(processes=self.n_cpus) as executor:
            # 用imap_unordered或者map都可以，这里用imap_unordered能实时看到进度
            for result in executor.imap_unordered(self.otherClass.some_method, tasks):
                pass

if __name__ == "__main__":
    app = App()
    app.call_multiprocessing()

关键说明：

• 给每个任务传入进程的位置索引pos，作为tqdm的position参数值，这样每个进度条会固定在控制台的对应行。
• 修改desc为包含进程编号的描述，方便区分不同进程的进度。
• 进度条完成后用tqdm.write清空该行，避免残留进度条影响后续输出。

方案二：使用tqdm官方的多进程进度条工具 #

tqdm提供了contrib.concurrent.process_map，专门支持多进程的进度条管理，无需手动处理位置，更简洁：

import os
import pandas as pd
from tqdm.contrib.concurrent import process_map
from multiprocessing import cpu_count

class OtherClass:
    def some_method(self, df):
        # 这里可以保留原有的tqdm，或者去掉，因为外层process_map会有总进度，内层可以每个进程显示自己的子进度
        for index, row in tqdm(df.iterrows(), desc="子进度", total=len(df), leave=False):
            # 业务逻辑处理
            pass
        return "处理完成"

class App:
    def __init__(self):
        self.otherClass = OtherClass()
        self.n_cpus = cpu_count()
        self.dataframes = [pd.DataFrame({'col': range(1000)}) for _ in range(self.n_cpus)]

    def call_multiprocessing(self):        
        # process_map会自动处理多进程，外层显示总进度，内层每个进程的子进度不会冲突
        results = process_map(self.otherClass.some_method, self.dataframes, max_workers=self.n_cpus)

if __name__ == "__main__":
    app = App()
    app.call_multiprocessing()

关键说明：

• process_map是tqdm封装的多进程工具，自动处理输出冲突，外层会显示总任务的进度，内层每个进程的子进度条设置leave=False，完成后会自动清除，不会干扰其他进程。
• 这种方式代码更简洁，无需手动管理进度条位置。

注意事项 #

• 必须在if __name__ == "__main__":代码块中启动多进程，这是Python多进程的要求，避免子进程重复初始化。
• 如果你的DataFrame很大，建议不要用iterrows()，因为它效率较低，可以改用df.apply()或者向量化操作，配合tqdm的progress_apply方法提升性能。

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