如何让 rich.progress 在多线程中安全更新同一进度条

2026-01-31 00:00:00 作者：冰川箭仙

多线程共用 Progress 实例会崩溃，因 Console 非线程安全且 Progress 共享状态未加锁；应使用 thread-safe Console + 独立 task_id 隔离更新，并统一所有输出到该 Console。

rich.progress 多线程直接共用 Progress 实例会崩溃

直接在多个线程里调用同一个 Progress 对象的 update() 或 advance()，大概率触发 RuntimeError: Working outside of application context（尤其搭配 Console 的非主线程刷新）或更隐蔽的显示错乱、卡死。根本原因是 rich.console.Console 默认不是线程安全的，且 Progress 内部共享状态（如任务列表、渲染计时器）未加锁。

推荐方案：用 thread-safe Console + task_id 隔离更新

核心是让所有线程共用一个线程安全的 Console 实例，并通过 add_task() 为每个线程分配独立 task_id，再用该 ID 更新——避免状态竞争。关键点：

Console 必须显式传入 force_terminal=True, color_system="truecolor" 等参数，并设置 soft_wrap=True（减少渲染冲突）
主线程创建 Progress 时，传入这个共享 Console；各线程只调用 update(task_id, advance=1)，绝不调用 add_task() 或修改任务元数据
所有线程必须等 Progress 启动后（即进入 with progress: 块）才能开始更新，否则 task_id 无效

示例：

from rich.progress import Progress from rich.console import Console from threading import Thread import time

console = Console(force_terminal=True, soft_wrap=True) progress = Progress(console=console)

def worker(taskid, n): for in range(n): time.sleep(0.1) progress.update(task_id, advance=1) # 安全：只更新自己 task

with progress: task1 = progress.add_task("[red]Thread-1", total=10) task2 = progress.add_task("[blue]Thread-2", total=15)

t1 = Thread(target=worker, args=(task1, 10))
t2 = Thread(target=worker, args=(task2, 15))
t1.start(); t2.start()
t1.join(); t2.join()

替代方案：用 multiprocessing.Manager 共享进度值（适合进程间）

如果实际用的是 multiprocessing（而非 threading），Progress 本身不支持跨进程。此时需绕过 UI 层，用 Manager.dict() 或 Value 同步原始进度数值，再由主线程轮询更新 UI：

各子进程写入 shared_dict['task1'] = 7 这类键值对
主线程在 Progress 的 refresh_per_second 控制下定期读取并调用 update(task_id, completed=shared_dict['task1'])
注意避免频繁轮询导致 CPU 升高，建议 refresh_per_second=4 起步