多线程环境下TQDM进度条显示异常问题求助

核心问题原因 #

多线程环境下，多个tqdm进度条同时操作终端光标位置，缺乏同步机制导致终端输出错乱，出现移位、残留痕迹等问题。

解决方案 #

使用全局线程锁同步所有tqdm的更新操作，确保同一时间只有一个进度条修改终端输出。同时调整进度条的关闭逻辑，避免光标位置异常。

修改后的完整代码 #

from concurrent.futures import ThreadPoolExecutor, as_completed
import random
from tqdm.auto import tqdm
from time import sleep 
from threading import Lock, Thread

# 全局锁，用于同步所有tqdm进度条的更新操作
tqdm_lock = Lock()

def latest_time(dir:str):
    return random.randint(0,100), str(random.randint(0,100)/10), False
    
def monitor_run(
        dir:str,
        name: str,
        position: int,
        max_iter: int,
        refresh_progress: float=2,
) -> None:
    sleep(1 + (position)/10)  # 调整延迟，避免初始位置重叠
    with tqdm(
        total = max_iter,
        bar_format="{l_bar}{bar:30}{r_bar}",
        desc=f"        {name}: 0.0 Progress",
        position= position,
        ncols = 100,
        leave = True,
        ascii= True,
        colour ='#00ff00',

    ) as pbar:
        desc_prev: float = 0
        while True:
            sleep(refresh_progress)
            time, desc, error = latest_time(dir)
            
            if desc is None:
                desc = desc_prev
            else:
                desc_prev = desc
                # 修改desc时加锁
                with tqdm_lock:
                    pbar.desc = f"         {name}: {desc} Progress"
            
            if error:
                # 使用tqdm.write时也建议加锁，避免输出混乱
                with tqdm_lock:
                    pbar.write(f"Encountered Error at {desc}")
                break
            
            if time is None:
                continue

            # 更新进度条数值并刷新时加锁
            with tqdm_lock:
                pbar.n = int(time)
                pbar.refresh()  # 去掉nolock=True，改用全局锁控制

            if time >= max_iter:
                # 确保进度条完成时的状态正确，由with块自动处理关闭
                with tqdm_lock:
                    pbar.n = max_iter
                    pbar.refresh()
                break

def serial_monitor(
    dir:str,
    position: int,
    max_iter: int,
    refresh_progress: float = 2,
)-> None:
    monitor_run(dir, str(position), position, max_iter, refresh_progress)


def serial_monitor_star(args)-> None:
    serial_monitor(*args)

def parallel_monitor(
        dirs: list[str],
        max_iter: int,
        refresh_progress: float =2,
) -> None:
    args_list = [
        [
            dir, position, max_iter, refresh_progress  # 修正position，从0开始连续分配
        ] for position, dir in enumerate(dirs)
    ]
    # 移除不必要的外层tqdm，避免干扰多进度条显示
    with ThreadPoolExecutor(max_workers= len(dirs)) as ex:
        futures = [
            ex.submit(
                serial_monitor_star,
                args
            ) for args in args_list
        ]
        for future in as_completed(futures):
            future.result()


if __name__ == "__main__":
    parallel_monitor(
        ['','','','','','',''],
        100,
        2
    )

关键修改点说明 #

• 全局锁tqdm_lock：所有涉及tqdm终端输出的操作（修改desc、更新n、refresh、write）都在锁的保护下执行，避免多线程同时修改光标位置。
• 移除nolock=True：原来的refresh(nolock=True)会跳过tqdm内部的锁，改用全局锁统一控制，更可靠。
• 修正position分配：从position+1改为直接使用enumerate的索引，确保进度条位置从0开始连续排列，避免空白行或重叠。
• 调整进度条完成逻辑：完成时手动将n设为max_iter并刷新，确保进度条显示100%后再退出循环，由with块自动处理关闭，避免光标异常。
• 移除外层冗余tqdm：原代码中的外层tqdm(total=2*len(dirs))会干扰多进度条的位置计算，直接移除后显示更清晰。

替代方案（无需全局锁） #

如果不想用全局锁，也可以将锁作为参数传递给每个monitor_run函数，本质和全局锁逻辑一致，只是锁的作用域更明确：

• 在parallel_monitor中创建锁，然后添加到args_list中
• 在monitor_run中接收锁参数，更新时使用该锁

这种方式更符合模块化设计，但对于简单场景，全局锁足够高效。

内容的提问来源于stack exchange，提问作者Tryfonas