如何让 rich.progress 在多线程中安全更新进度条_技术教程

多线程下需用 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 time
progress = 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()