Python中为asyncio异步任务实现动态更新的单条tqdm进度条

两种方案的异常点对应原因如下：

• asyncio.as_completed 写法进度卡在0%/100%：本质是事件循环被阻塞，要么是self.run_scan内部存在未做异步适配的阻塞IO（比如同步文件读写、同步HTTP请求），所有任务实际串行执行完才交还控制权给主逻辑，要么是误用了同步版tqdm，刷新逻辑无法在异步间隙触发。
• tqdm.gather 写法出现多条进度条：额外手动创建了tqdm上下文，tqdm.asyncio.tqdm.gather本身会自动创建进度条实例，两层实例叠加就会重复输出；如果导入的是同步版tqdm的gather，同样会出现刷新不及时的问题。

方案1：tqdm异步原生gather写法（最简洁，推荐） #

不需要额外套上下文管理器，直接导入异步版本的tqdm调用gather即可，自动生成单条实时更新的进度条：

from tqdm.asyncio import tqdm

scan_results = await tqdm.gather(
    *[self.run_scan(i) for i in input_paths],
    desc="Scanning files"
)

如果run_scan内部存在阻塞逻辑，要先把阻塞操作放到线程池执行，避免卡死事件循环：

import asyncio
import functools

# 原有阻塞式扫描逻辑
def blocking_scan(path):
    return scan_result

async def run_scan(self, path):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, functools.partial(blocking_scan, path))

方案2：as_completed手动控制写法（自定义程度更高） #

如果需要对每个任务的返回值做即时处理，用该写法，注意必须使用异步版本的tqdm初始化进度条：

import asyncio
from tqdm.asyncio import tqdm

scan_results = []
tasks = [asyncio.create_task(self.run_scan(i)) for i in input_paths]

pbar = tqdm(total=len(tasks), desc='Scanning files')
for coro in asyncio.as_completed(tasks):
    value = await coro
    pbar.update(1)
    scan_results.append(value)
pbar.close()

• 必须从tqdm.asyncio导入tqdm，不要直接导入同步版本的tqdm，否则异步场景下刷新逻辑会被阻塞。
• 所有异步任务内部禁止直接调用阻塞式函数，必须通过run_in_executor包装，否则事件循环卡住，进度条无法实时更新。
• 不要给tqdm.gather额外嵌套tqdm上下文，避免重复生成进度条。

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