使用os.fork多进程时Rich进度条子进程更新不同步问题求助

问题根源 #

os.fork()会创建父进程的完全副本，子进程中的Progress对象和父进程的是独立实例，两者的终端输出状态完全隔离。子进程对进度条的更新不会同步到父进程的进度条，甚至可能导致终端输出混乱。

解决方案1：父进程统一更新进度（最简单） #

让子进程只负责执行任务，父进程等待每个子进程完成后，再更新进度条。这种方式不需要额外的进程间通信，逻辑清晰。

修改后的代码示例：

from rich.progress import Progress
import os
import sys

def do_something(my_tuple):
    # 这里是你的任务逻辑
    pass

with Progress() as progress:
    task = progress.add_task(total=len(args_list), description="处理任务批次")
    child_pids = []
    batch_sizes = []

    # 批量创建子进程
    for balanced_batch in even_batches_it:
        pid = os.fork()
        if pid == 0:
            # 子进程：仅执行任务，不碰进度条
            try:
                for my_tuple in balanced_batch:
                    do_something(my_tuple)
                sys.exit(0)  # 正常退出
            except Exception as e:
                print(f"子进程执行失败: {str(e)}", file=sys.stderr)
                sys.exit(1)  # 异常退出
        else:
            # 父进程：记录子进程PID和对应批次的大小
            child_pids.append(pid)
            batch_sizes.append(len(balanced_batch))

    # 父进程等待子进程完成，逐个更新进度
    for pid, batch_size in zip(child_pids, batch_sizes):
        os.waitpid(pid, 0)  # 等待子进程结束
        progress.advance(task, advance=batch_size)  # 父进程统一更新进度

说明：父进程是唯一控制终端输出的进程，所有进度条更新操作都在父进程中完成，彻底避免进程间同步问题。

解决方案2：用IPC实现实时进度更新（适合细粒度进度） #

如果需要实时更新子进程内的任务进度（比如每个my_tuple处理完就更新），可以用管道实现进程间通信，子进程将进度数据发送给父进程，由父进程更新进度条。

示例代码（基于管道）：

from rich.progress import Progress
import os
import sys
import select

def do_something(my_tuple):
    # 你的任务逻辑
    pass

with Progress() as progress:
    task = progress.add_task(total=len(args_list), description="实时处理进度")
    read_pipes = []
    child_pids = []

    for balanced_batch in even_batches_it:
        # 创建管道：父进程读，子进程写
        r_fd, w_fd = os.pipe()
        pid = os.fork()

        if pid == 0:
            os.close(r_fd)  # 子进程关闭读端
            try:
                processed = 0
                for my_tuple in balanced_batch:
                    do_something(my_tuple)
                    processed += 1
                    # 发送当前已处理数量给父进程
                    os.write(w_fd, f"{processed}\n".encode())
                os.close(w_fd)
                sys.exit(0)
            except Exception as e:
                print(f"子进程出错: {str(e)}", file=sys.stderr)
                os.close(w_fd)
                sys.exit(1)
        else:
            os.close(w_fd)  # 父进程关闭写端
            read_pipes.append((r_fd, len(balanced_batch)))
            child_pids.append(pid)

    # 父进程监听管道，实时更新进度
    processed_total = 0
    while read_pipes:
        # 监听所有可读的管道
        readable, _, _ = select.select([r for r, _ in read_pipes], [], [], 0.1)
        for r_fd in readable:
            data = os.read(r_fd, 1024).decode().strip()
            if data:
                # 取最后一条有效进度数据
                current_processed = int(data.split('\n')[-1])
                # 可根据需求维护每个批次的已处理数，实现增量更新
                pass
            # 管道关闭，说明子进程已完成
            if os.read(r_fd, 1) == b'':
                os.close(r_fd)
                # 找到对应批次的大小，更新总进度
                for idx, (pipe, size) in enumerate(read_pipes):
                    if pipe == r_fd:
                        processed_total += size
                        progress.advance(task, advance=size)
                        del read_pipes[idx]
                        break

    # 等待所有子进程退出
    for pid in child_pids:
        os.waitpid(pid, 0)

解决方案3：改用multiprocessing模块（推荐） #

直接使用os.fork()属于底层操作，Python的multiprocessing模块封装了进程管理和IPC，代码更简洁易维护。

示例代码：

from rich.progress import Progress
from multiprocessing import Pool
import sys

def process_batch(batch):
    try:
        for my_tuple in batch:
            do_something(my_tuple)
        return len(batch)
    except Exception as e:
        print(f"批次处理失败: {str(e)}", file=sys.stderr)
        return 0

if __name__ == "__main__":
    with Progress() as progress:
        task = progress.add_task(total=len(args_list), description="多进程处理")
        # 创建进程池，用回调函数更新进度
        with Pool() as pool:
            results = []
            for balanced_batch in even_batches_it:
                # 提交任务，回调函数在父进程执行，直接更新进度
                res = pool.apply_async(
                    process_batch, 
                    args=(balanced_batch,),
                    callback=lambda size: progress.advance(task, advance=size)
                )
                results.append(res)
            # 等待所有任务完成
            for res in results:
                res.get()

说明：apply_async的回调函数在父进程中执行，因此可以直接调用progress.advance()更新进度，无需手动处理进程间通信。

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