Python使用asyncio+aiohttp實(shí)現(xiàn)高效異步并發(fā)下載

更新時(shí)間：2026年01月23日 08:36:30 作者：weixin_46244623

這篇文章主要為大家詳細(xì)介紹了如何使用 Python 的異步編程模型（asyncio）配合 aiohttp 庫(kù),實(shí)現(xiàn)真正高效的異步并發(fā)下載,大幅提升下載速度,感興趣的小伙伴可以了解下

一、為什么選擇 aiohttp + asyncio

aiohttp 是基于 asyncio 的高性能異步 HTTP 客戶端/服務(wù)器框架；
支持異步請(qǐng)求、自動(dòng)連接池、重定向、超時(shí)控制；
語(yǔ)法簡(jiǎn)潔，性能遠(yuǎn)超多線程或同步方案；
適合 I/O 密集型任務(wù)（如網(wǎng)絡(luò)下載）。

注意：Python 的“異步” ≠ “多線程”，它是通過(guò)事件循環(huán)（Event Loop）在單線程中并發(fā)處理多個(gè) I/O 操作，避免阻塞。

二、安裝依賴

首先安裝 aiohttp（若未安裝）：

pip install aiohttp

推薦同時(shí)安裝 aiofiles（用于異步寫入文件）：

pip install aiofiles

三、核心代碼實(shí)現(xiàn)

方法一：將內(nèi)容全部加載到內(nèi)存后保存（適合中小文件）

import asyncio
import aiohttp
import os

async def download_file(session, url, filename):
    """異步下載單個(gè)文件"""
    try:
        async with session.get(url) as response:
            if response.status == 200:
                content = await response.read()
                with open(filename, 'wb') as f:
                    f.write(content)
                print(f"? 下載成功: {filename}")
            else:
                print(f"? 下載失敗 ({response.status}): {url}")
    except Exception as e:
        print(f"??  下載出錯(cuò): {url} | 錯(cuò)誤: {e}")

async def download_all(urls, folder="downloads"):
    """并發(fā)下載多個(gè)文件"""
    os.makedirs(folder, exist_ok=True)

    # 創(chuàng)建 aiohttp 客戶端會(huì)話（支持連接復(fù)用）
    async with aiohttp.ClientSession() as session:
        tasks = []
        for i, url in enumerate(urls):
            # 生成文件名（可根據(jù)需要自定義）
            ext = os.path.splitext(url.split('?')[0])[-1] or '.bin'
            filename = os.path.join(folder, f"file_{i+1}{ext}")
            task = asyncio.create_task(download_file(session, url, filename))
            tasks.append(task)

        # 并發(fā)執(zhí)行所有下載任務(wù)
        await asyncio.gather(*tasks)

# 使用示例
if __name__ == "__main__":
    urls = [
        "https://example.com/file1.zip",
        "https://example.com/file2.jpg",
        "https://example.com/file3.pdf",
    ]

    print("?? 開始異步下載...")
    asyncio.run(download_all(urls))
    print("?? 所有文件下載完成！")

方法二：流式下載（適合大文件，節(jié)省內(nèi)存）

對(duì)于大文件（如視頻、ISO 鏡像），建議使用流式寫入，避免一次性加載整個(gè)文件到內(nèi)存：

import aiofiles  # 需要額外安裝

async def download_file_stream(session, url, filename):
    try:
        async with session.get(url) as response:
            if response.status == 200:
                async with aiofiles.open(filename, 'wb') as f:
                    async for chunk in response.content.iter_chunked(8192):
                        await f.write(chunk)
                print(f"? 流式下載成功: {filename}")
            else:
                print(f"? 下載失敗 ({response.status}): {url}")
    except Exception as e:
        print(f"??  流式下載出錯(cuò): {url} | 錯(cuò)誤: {e}")

# 在 download_all 中替換 download_file 為 download_file_stream 即可

四、性能優(yōu)勢(shì)對(duì)比

方式	3 個(gè) 50MB 文件耗時(shí)（估算）
同步下載（requests）	~45 秒
多線程（ThreadPoolExecutor）	~20 秒
異步下載（aiohttp + asyncio）	~15–18 秒

實(shí)際提升取決于網(wǎng)絡(luò)帶寬和服務(wù)器并發(fā)能力。異步方案在高延遲或大量小文件場(chǎng)景下優(yōu)勢(shì)更明顯。

五、進(jìn)階優(yōu)化建議

1.限制并發(fā)數(shù)量

若 URL 數(shù)量極大（如 1000+），可使用 asyncio.Semaphore 控制并發(fā)數(shù)，避免打爆目標(biāo)服務(wù)器或本地資源：

semaphore = asyncio.Semaphore(10)  # 最多 10 個(gè)并發(fā)

async def download_file_limited(session, url, filename):
    async with semaphore:
        await download_file(session, url, filename)

2.添加 User-Agent 和超時(shí)

timeout = aiohttp.ClientTimeout(total=30)
async with aiohttp.ClientSession(
    timeout=timeout,
    headers={"User-Agent": "Mozilla/5.0 (Python aiohttp)"}
) as session:

3.自動(dòng)從 URL 提取文件名

from urllib.parse import urlparse
import os

def get_filename_from_url(url):
    path = urlparse(url).path
    name = os.path.basename(path)
    return name if name else "downloaded_file"

六、總結(jié)

使用 asyncio + aiohttp 是 Python 實(shí)現(xiàn)高并發(fā)下載的最佳實(shí)踐；
異步下載能顯著提升效率，尤其適合批量、I/O 密集型任務(wù)；
對(duì)于大文件，推薦使用流式下載 + aiofiles 避免內(nèi)存溢出；
生產(chǎn)環(huán)境中務(wù)必加入錯(cuò)誤處理、超時(shí)控制、并發(fā)限制。

掌握這套異步下載方案，你就能輕松應(yīng)對(duì)各種批量資源采集、自動(dòng)化備份、數(shù)據(jù)同步等需求！

到此這篇關(guān)于Python使用asyncio+aiohttp實(shí)現(xiàn)高效異步并發(fā)下載的文章就介紹到這了,更多相關(guān)Python異步并發(fā)下載內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: