多线程下需用 threading.Lock 串行化 rich.progress.update() 调用,主线程创建 Progress 和锁,子线程持 task_id 并在锁内更新;或改用 Live 配合线程安全状态管理;multiprocessing 不支持 Progress 共享。

rich.progress 多线程下直接更新会崩溃
直接在多个线程里调用 progress.update() 会导致 RuntimeError: Cannot update progress bar from multiple threads。这是因为 Progress 内部的渲染器(ConsoleRender)不是线程安全的,且进度条状态(如任务计数、已用时间)共享但未加锁。
用 threading.Lock 手动同步更新操作
最轻量、可控性最强的做法:所有对 progress.update() 的调用都包裹在一个全局 threading.Lock 里。这不是“绕过” rich 的限制,而是显式地串行化 UI 更新逻辑。
- 必须在主线程创建
Progress实例和Lock,然后传给子线程 - 每次调用
progress.update(task_id, advance=1)前必须lock.acquire(),完成后lock.release()(推荐用with lock:) - 不要在锁内做耗时操作(比如文件读写、网络请求),否则会严重拖慢整体进度更新响应
-
task_id必须由主线程分配并传入线程,不能在线程里调用progress.add_task()
from rich.progress import Progress import threading import timeprogress = Progress() lock = threading.Lock() task = progress.add_task("[cyan]Processing...", total=100)
def worker(i): time.sleep(0.1) # 模拟工作 with lock: progress.update(task, advance=1)
threads = [threading.Thread(target=worker, args=(i,)) for i in range(100)] for t in threads: t.start() for t in threads: t.join()
progress.stop()
改用 rich.live.Live 配合手动状态管理
当需要更灵活的多线程 UI(比如显示多个实时指标、非进度类状态),Live 比 Progress 更适合——它只负责刷新整个渲染结果,不管理任务生命周期,因此线程安全责任完全落在你身上。
- 用一个线程安全的数据结构(如
threading.local()或带锁的dict)收集各线程状态 - 单独起一个刷新线程,定期读取状态、构造新文本,再调用
live.update() - 避免在
live.update()中做计算;所有聚合逻辑放在刷新线程外 - 注意
Live默认启用refresh_per_second=4,太高会刷屏,太低会卡顿
别碰 multiprocessing —— rich.progress 本身不支持进程间共享
Progress 对象无法被序列化(pickle),也不能跨进程传递。即使你用 Manager 或 Queue 试图同步进度,在子进程中调用 progress.update() 会直接报 AttributeError 或静默失败。
- 如果必须用多进程,只能让子进程通过
Queue发送完成信号,由主线程统一更新Progress - 别尝试在子进程里 import rich 或创建新
Progress—— 控制台输出会混乱,甚至阻塞 - 考虑换用
tqdm+concurrent.futures.as_completed这类更适配多进程的组合
真正麻烦的不是“怎么更新”,而是“谁来决定更新时机”。rich 的设计默认把进度控制权交给单一线程,强行多线程更新容易掩盖数据竞争问题——比如两个线程同时 advance=1,却只生效一次。锁只是表层解法,背后得理清你的工作流是否真的需要并发更新 UI。









