在Python的多线程编程场景中，如果需要在多个线程中共同更新同一个rich.progress进度条，由于rich的进度条组件本身没有内置线程安全机制，直接并发调用更新方法会导致显示异常或者程序报错。要解决这个问题，核心思路是对进度条的更新操作做序列化处理，避免多个线程同时修改进度条状态。
为什么多线程直接更新rich进度条不安全
rich的Progress类在更新进度时，会同时修改内部的状态变量和终端的输出内容，这些操作都不是原子操作。当多个线程同时调用advance或者update方法时，可能会出现以下问题：
- 进度数值计算错误，多个线程的增量没有正确累加
- 终端输出内容重叠、错乱，显示格式被破坏
- 内部状态变量被并发修改，触发异常导致程序崩溃
方案一：使用线程锁保证更新串行化
最直接的解决方式是给进度条的更新操作加一把线程锁，确保同一时间只有一个线程可以执行更新逻辑。这种方式实现简单，适合线程数量不多、更新频率不高的场景。
实现步骤
首先导入需要的模块，创建进度条和线程锁，然后在每个线程的更新逻辑中先获取锁，完成更新后再释放锁。
import threading
from rich.progress import Progress
# 创建进度条和线程锁
progress = Progress()
task_id = progress.add_task("多线程任务", total=100)
lock = threading.Lock()
def worker(thread_num, step):
# 模拟线程执行任务
for i in range(step):
# 获取锁，保证更新操作串行执行
with lock:
progress.advance(task_id, 1)
# 模拟任务耗时
import time
time.sleep(0.01)
print(f"线程{thread_num}任务完成")
if __name__ == "__main__":
progress.start()
# 创建3个线程，每个线程更新30次，总共更新90次
threads = []
for i in range(3):
t = threading.Thread(target=worker, args=(i, 30))
threads.append(t)
t.start()
# 等待所有线程完成
for t in threads:
t.join()
progress.stop()
方案优缺点
优点是实现逻辑简单，不需要额外引入复杂的组件，容易理解。缺点是如果线程数量很多、更新频率很高，锁的竞争会比较激烈，可能会影响程序的整体性能。
方案二：使用任务队列中转更新请求
如果更新频率较高，可以使用一个单独的后台线程来消费更新请求队列，所有工作线程只需要把更新请求放入队列，由后台线程统一执行进度条更新，从根源上避免并发修改的问题。
实现步骤
创建一个线程安全的队列，启动一个后台消费者线程不断从队列中取出更新请求并执行，工作线程只需要往队列里放入更新参数即可。
import threading
import queue
import time
from rich.progress import Progress
# 创建进度条和更新请求队列
progress = Progress()
task_id = progress.add_task("队列更新任务", total=100)
update_queue = queue.Queue()
def progress_consumer():
# 消费者线程，统一处理所有更新请求
progress.start()
while True:
item = update_queue.get()
# 收到结束信号则退出
if item is None:
update_queue.task_done()
break
# 执行更新操作
progress.advance(task_id, item)
update_queue.task_done()
progress.stop()
def worker(thread_num, step):
for i in range(step):
# 把更新请求放入队列，不需要加锁
update_queue.put(1)
time.sleep(0.01)
print(f"线程{thread_num}任务完成")
if __name__ == "__main__":
# 启动消费者线程
consumer_thread = threading.Thread(target=progress_consumer, daemon=True)
consumer_thread.start()
# 创建工作线程
threads = []
for i in range(3):
t = threading.Thread(target=worker, args=(i, 30))
threads.append(t)
t.start()
# 等待所有工作线程完成
for t in threads:
t.join()
# 放入结束信号，停止消费者线程
update_queue.put(None)
consumer_thread.join()
方案优缺点
优点是工作线程不需要等待锁，更新请求可以异步处理，性能更好，适合高并发更新场景。缺点是需要额外维护一个消费者线程和队列，实现逻辑相对复杂一些。
两种方案的选择建议
如果线程数量少、更新频率低，优先选择线程锁方案，实现成本低。如果线程数量多、更新频繁，或者需要解耦工作线程和UI更新逻辑，优先选择任务队列方案。无论选择哪种方案，核心都是避免多个线程同时直接调用rich.progress的更新方法，保证进度条状态修改的串行化。