📌 摘要

本文详细介绍了如何使用 Python + asyncio + aiohttp + tqdm 实现一个 带进度条的异步多文件下载器。相比传统的同步下载方式，异步下载不仅能够显著提高下载速度，还能在终端显示每个文件的实时下载进度条，适用于大文件、多文件并发下载等场景。文章提供了详细中文注释的完整代码示例，读者可以直接拷贝运行。同时也介绍了核心原理、依赖库、适用场景等内容，非常适合学习 Python 异步 IO 编程的开发者参考。

🧰 使用的库说明

• asyncio：Python 内置的异步协程库，用于并发执行任务。

• aiohttp：支持用于异步 HTTP 请求，比 requests 更适合同步下载。

• tqdm：终端显示进度条的工具。

安装依赖：

pip install aiohttp tqdm

🚀 异步下载的优势

1. 无需多线程，也能同时下载多个文件

2. 避免同步阻塞，大幅提升下载速度

3. 每个任务都能显示独立进度条

📦 核心思路

1. 使用 aiohttp.ClientSession() 进行异步请求

2. 使用 await response.content.read(1024) 异步读取内容

3. 使用 tqdm 显示进度条

4. 使用 asyncio.gather() 并发执行多个下载任务

🎯 适用场景

• 下载多个大文件（Zip、MP4、模型文件等）

• 写爬虫批量下载文件

• 后台批处理任务

• 分布式下载系统

import aiohttp
import asyncio
from tqdm import tqdm

# 异步下载函数
async def download_file(url, destination):
    """
    异步下载文件并显示下载进度条

    :param url:          文件下载地址
    :param destination:  本地保存文件路径
    """
    # 创建 aiohttp 会话
    async with aiohttp.ClientSession() as session:
        # 发送 GET 请求
        async with session.get(url) as response:
            # 获取文件总大小（如果服务器支持）
            total_size = int(response.headers.get("Content-Length", 0))

            # 打开本地文件，准备写入下载内容
            with open(destination, "wb") as f:
                # tqdm 进度条
                progress_bar = tqdm(
                    total=total_size,         # 总大小
                    unit="B",                 # 单位：字节
                    unit_scale=True,          # 自动转换 KB / MB
                    desc=f"下载中: {destination}"  # 显示的标题
                )

                # 持续读取数据（每次读取 1024 字节）
                while True:
                    chunk = await response.content.read(1024)
                    if not chunk:   # 读取完毕
                        break
                    f.write(chunk)
                    progress_bar.update(len(chunk))  # 更新进度条

                progress_bar.close()  # 下载完成后关闭进度条


# 主异步函数
async def main():
    """
    批量创建多个下载任务，并异步并发执行
    """

    # 要下载的文件列表（可自行扩展）
    download_tasks = [
        {
            "url": "http://example.com/file1.zip",
            "destination": "file1.zip"
        },
        {
            "url": "http://example.com/file2.zip",
            "destination": "file2.zip"
        },
        # 可以继续添加更多文件
    ]

    # 为每个任务创建一个 download_file 协程
    tasks = [
        download_file(task["url"], task["destination"])
        for task in download_tasks
    ]

    # 并发执行所有下载任务
    await asyncio.gather(*tasks)


# 程序入口
if __name__ == "__main__":
    # 运行异步主函数
    asyncio.run(main())