python异步并发任务进度条

import asyncio
import aiohttp
from tqdm import tqdm
from typing import List, Coroutine, Anyclass AsyncProgressBar:"""异步任务进度条管理器"""def __init__(self, total_tasks: int, desc: str = "异步任务进度"):self.total = total_tasksself.completed = 0self.pbar = tqdm(total=total_tasks, desc=desc, unit="task")def update(self):"""更新进度条（线程安全）"""self.completed += 1self.pbar.update(1)def close(self):"""关闭进度条"""self.pbar.close()async def async_task(session: aiohttp.ClientSession, url: str, pbar: AsyncProgressBar) -> Any:"""单个异步任务示例（以HTTP请求为例）"""try:async with session.get(url) as response:# 模拟任务处理耗时（可替换为实际业务逻辑）# await asyncio.sleep(0.1)result = {"url": url,"status": response.status,"content_length": response.content_length}return resultexcept Exception as e:return {"url": url, "error": str(e)}finally:# 任务完成后更新进度条pbar.update()async def run_async_tasks(tasks: List[Coroutine], max_concurrency: int = 10) -> List[Any]:"""运行异步任务并显示进度条:param tasks: 异步任务列表:param max_concurrency: 最大并发数（控制并发量）:return: 所有任务的结果列表"""# 初始化进度条pbar = AsyncProgressBar(total_tasks=len(tasks))# 限制并发量（防止请求过多被封禁）semaphore = asyncio.Semaphore(max_concurrency)# 包装任务以支持并发限制async def bounded_task(task: Coroutine) -> Any:async with semaphore:return await task# 创建带并发限制的任务列表bounded_tasks = [bounded_task(task) for task in tasks]# 并发执行任务并收集结果results = await asyncio.gather(*bounded_tasks)# 关闭进度条pbar.close()return resultsasync def main():"""主函数：示例使用"""# 示例任务列表（100个HTTP请求任务）urls = [f"https://httpbin.org/get?num={i}" for i in range(100)]# 创建aiohttp会话（复用连接，提升性能）async with aiohttp.ClientSession() as session:# 构建任务列表pbar = AsyncProgressBar(total_tasks=len(urls))  # 初始化进度条tasks = [async_task(session, url, pbar) for url in urls]# 运行任务（最大并发20）results = await run_async_tasks(tasks, max_concurrency=20)# 输出结果统计success = sum(1 for r in results if "error" not in r)print(f"\n任务完成：成功 {success}/{len(results)}")if __name__ == "__main__":# 适配Python 3.7+的异步运行方式asyncio.run(main())