如何集中跟踪Python multiprocessing.Pool中Worker的运行进度？

问题背景 #

使用multiprocessing.Pool运行耗时任务时，需要集中跟踪每个Worker的进度，定期输出类似：

A: 13%, B: 12%, C: 0%, D: 0%

的状态。尝试用Manager.Queue传递进度时出现KeyError等异常，核心问题是未共享同一个队列实例，且进程退出逻辑不完善。

错误代码分析 #

你之前的代码中，给每个Worker和打印进程都创建了新的m.Queue()，导致它们使用的不是同一个队列，无法传递数据。此外，打印进程没有退出条件，会一直阻塞等待消息，引发后续清理异常。

错误代码示例：

def f(tag, q):
    for pct in range(10):
        q.put((tag, pct))
        time.sleep(0.2)

def print_status(q: Queue):
    status = {}
    while True:
        tag, pct = q.get(block=True, timeout=3)
        status[tag] = pct
        print(status)
        time.sleep(3)

if __name__ == '__main__':
    m = Manager()
    p = Process(target=print_status, args=(m.Queue(),))  # 新队列1
    p.start()
    pool = Pool(processes=2)
    # 每个Worker都用新队列2、3...
    pool.starmap(f, [(x, m.Queue()) for x in 'ABCDEF']) 

可行解决方案 #

方案1：主进程负责状态打印 #

无需额外创建打印进程，主进程在等待Pool任务完成的间隙，定期从共享队列读取进度并更新状态。

import time
from multiprocessing import Pool, Manager
from queue import Empty

def worker(tag, q):
    # 模拟耗时任务，逐步上报进度
    for pct in range(0, 101, 10):
        q.put((tag, pct))
        time.sleep(0.5)
    # 任务完成后标记结束
    q.put((tag, 100))

def main():
    m = Manager()
    progress_queue = m.Queue()
    tags = list('ABCDEF')
    status = {tag: 0 for tag in tags}
    
    # 启动进程池
    with Pool(processes=2) as pool:
        # 提交所有任务，共享同一个队列
        pool.starmap_async(worker, [(tag, progress_queue) for tag in tags])
        
        # 主进程循环更新状态
        while any(v < 100 for v in status.values()):
            try:
                # 批量读取队列中的所有进度更新
                while True:
                    tag, pct = progress_queue.get(block=False)
                    status[tag] = pct
            except Empty:
                pass
            # 格式化输出状态
            status_str = ", ".join([f"{tag}: {pct}%" for tag, pct in status.items()])
            print(f"\r{status_str}", end="", flush=True)
            time.sleep(1)
    
    print("\n所有任务完成")

if __name__ == '__main__':
    main()

方案2：单独进程负责状态打印 #

创建专用的状态打印进程，共享同一个队列，同时添加退出标记，让打印进程能正常终止。

import time
from multiprocessing import Pool, Manager, Process
from queue import Empty

def worker(tag, q):
    for pct in range(0, 101, 10):
        q.put((tag, pct))
        time.sleep(0.5)
    q.put((tag, 100))

def status_printer(q, total_tasks):
    status = {}
    completed = 0
    while completed < total_tasks:
        try:
            while True:
                tag, pct = q.get(block=False)
                status[tag] = pct
                if pct == 100:
                    completed += 1
        except Empty:
            pass
        status_str = ", ".join([f"{tag}: {status.get(tag, 0)}%" for tag in list('ABCDEF')])
        print(f"\r{status_str}", end="", flush=True)
        time.sleep(1)
    print("\n所有任务完成")

def main():
    m = Manager()
    progress_queue = m.Queue()
    tags = list('ABCDEF')
    
    # 启动状态打印进程
    printer = Process(target=status_printer, args=(progress_queue, len(tags)))
    printer.start()
    
    # 启动进程池执行任务
    with Pool(processes=2) as pool:
        pool.starmap(worker, [(tag, progress_queue) for tag in tags])
    
    # 等待打印进程结束
    printer.join()

if __name__ == '__main__':
    main()

关键注意事项 #

• 必须共享同一个Manager.Queue实例，所有Worker和状态处理进程都使用这个队列传递数据。
• 添加任务完成标记（比如pct=100），让状态处理逻辑能判断何时终止，避免无限阻塞。
• 使用\r实现单行刷新状态，避免输出刷屏。

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