在基于 os.fork() 的多进程场景中，Rich 进度条无法直接跨父子进程共享状态；本文提供两种专业可行的替代方案：一是通过主进程统一调度 + 子进程信号反馈（推荐），二是重构为单进程分步执行并嵌入进度更新（适用于轻量任务）。
在基于 `os.fork()` 的多进程场景中，rich 进度条无法直接跨父子进程共享状态；本文提供两种专业可行的替代方案：一是通过主进程统一调度 + 子进程信号反馈（推荐），二是重构为单进程分步执行并嵌入进度更新（适用于轻量任务）。
使用 os.fork() 实现并行处理时，每个子进程都拥有独立的内存空间和 Python 解释器状态，因此 rich.Progress 实例（包括其内部计数器、渲染线程和终端句柄）无法在父子进程间自动同步。你在子进程中调用 progress.advance()，实际操作的是该子进程私有的 Progress 副本，对父进程的进度条毫无影响——这正是你遇到问题的根本原因。
✅ 推荐方案：主进程驱动 + 进程间通信（IPC）
最健壮、可扩展的方式是将进度控制权完全交还给父进程，子进程仅负责计算并通知完成量。可借助 multiprocessing.Queue 或 multiprocessing.Value 实现安全通信：
from rich.progress import Progress
import multiprocessing as mp
import os
def worker(batch, result_queue):
"""子进程执行函数，完成后向队列发送完成数量"""
try:
for my_tuple in batch:
do_something(my_tuple)
# 发送本次 batch 处理完成的项数（非基因数，需按实际逻辑调整）
result_queue.put(len(batch))
except Exception as e:
result_queue.put(0) # 或发送错误标记
raise
# 主流程
if __name__ == "__main__":
total_items = sum(len(b) for b in even_batches_it)
with Progress() as progress:
task = progress.add_task("Processing batches...", total=total_items)
# 使用 multiprocessing.Manager().Queue 替代普通 Queue（支持 fork）
result_queue = mp.Manager().Queue()
processes = []
for balanced_batch in even_batches_it:
p = mp.Process(target=worker, args=(balanced_batch, result_queue))
p.start()
processes.append(p)
# 父进程持续监听完成信号并更新进度
completed = 0
while completed < total_items:
try:
# 非阻塞获取结果（避免卡死）
n = result_queue.get_nowait()
completed += n
progress.update(task, advance=n)
except: # queue.Empty or other errors
pass
time.sleep(0.01) # 防止忙轮询
# 等待所有子进程结束
for p in processes:
p.join()⚠️ 注意事项：
- 不要在子进程中创建或操作 Progress 实例，所有渲染必须由父进程单点控制；
- mp.Manager().Queue() 是 fork-safe 的，而普通 queue.Queue 在 fork 后行为未定义；
- 若需捕获异常或返回详细结果，可改用 result_queue.put((success, n, error_info)) 结构化通信；
- 对于超大数据集，建议添加超时机制与重试逻辑。
⚠️ 不推荐方案：exec() 动态注入（原答案方法的问题）
原答案中通过 inspect.getsource() + exec() 动态插入 progress.update() 的方式存在严重缺陷：
立即学习“Python免费学习笔记（深入）”；
快速生成专业的 Python 脚本和应用代码。一键创建完整项目结构，支持CLI、API、爬虫、Bot、Django等多种项目类型，包含完整的项目结构、配置文件、依赖管理、测试、README和文档。
- ❌ 破坏可调试性：堆栈跟踪指向生成的字符串而非原始代码；
- ❌ 不兼容 fork：exec() 仍在单进程内执行，未解决多进程同步问题；
- ❌ 语法脆弱：缩进解析易出错，无法处理装饰器、嵌套函数、注释等复杂结构；
- ❌ 安全隐患：exec() 执行任意代码，违反最小权限原则。
因此，该方法仅适用于纯单进程、教学演示或极简脚本，绝不应用于生产环境或 fork 场景。
✅ 替代思路：放弃 fork，改用 concurrent.futures（更现代、更安全）
若无强依赖 os.fork() 的底层需求，强烈建议迁移到 concurrent.futures.ProcessPoolExecutor —— 它内置进程隔离与结果收集，配合 Rich 可无缝协作：
from rich.progress import Progress, track
from concurrent.futures import ProcessPoolExecutor, as_completed
def process_batch(batch):
results = []
for my_tuple in batch:
results.append(do_something(my_tuple))
return len(batch) # 返回处理数量
# 主流程（简洁清晰）
if __name__ == "__main__":
all_batches = list(even_batches_it)
total = sum(len(b) for b in all_batches)
with Progress() as progress:
task = progress.add_task("Processing...", total=total)
with ProcessPoolExecutor() as executor:
# 提交所有任务
futures = {executor.submit(process_batch, b): b for b in all_batches}
# 按完成顺序更新进度
for future in as_completed(futures):
n = future.result()
progress.update(task, advance=n)此方式语义清晰、异常透明、资源自动管理，是 Python 并行编程的现代标准实践。
总结：fork + Rich 进度条的本质矛盾在于内存隔离。解决方案的核心原则是——进度状态必须中心化，计算任务必须去中心化。优先选择 multiprocessing IPC 或 concurrent.futures，彻底规避跨进程状态同步难题。
