提示

返回博客列表

下载 100 个分片要等 100 秒？用 Python 异步并发把时间压到 10 秒

VidDown 2026-08-11 分钟阅读
5 次阅读

下载 100 个分片要等 100 秒？用 Python 异步并发把时间压到 10 秒

你用 Python 写了一个视频下载器，功能都正常——但它太慢了。一个 M3U8 视频有 500 个 TS 分片，每个分片下载 0.5 秒，串行下载就是 250 秒。用户等了 4 分钟，进度条才走到头。有没有办法让这 500 个请求同时跑？这就是 Python 异步编程（asyncio + aiohttp） 要解决的问题。本文从零讲解异步下载的原理、协程与线程的区别、并发控制策略、以及在视频下载器中的实战应用——让你的下载速度提升 5-20 倍。

TL;DR：Python 的 asyncio + aiohttp 可以实现真正的并发 HTTP 下载，利用 I/O 等待时间同时发起多个请求。核心模式：asyncio.Semaphore 控制并发数 → aiohttp.ClientSession 复用连接池 → asyncio.gather() 批量执行 → 回调函数更新进度。配合断点续传和自动重试，可以实现工业级的下载性能。

目录

一、为什么串行下载这么慢

1.1 时间都花在哪了

import time
import requests

def download_serial(segments):
    for seg in segments:
        resp = requests.get(seg['url'])  # 等待 0.5s
        save_to_disk(resp.content)        # 写入 0.1s
    # 100 个分片 = 100 × (0.5 + 0.1) = 60 秒

时间拆解：

单个分片下载（0.5s）：
├── DNS 解析        5ms
├── TCP 握手       15ms
├── TLS 握手       30ms
├── 发送 HTTP 请求   1ms
├── 等待服务器处理  400ms  ← 主要时间消耗（I/O 等待）
├── 接收数据        40ms
└── 写磁盘          10ms

总计 501ms，其中 I/O 等待占 80%

关键发现：下载过程中，CPU 大部分时间在等待网络 I/O，而不是在计算。这意味着我们可以利用等待时间同时处理其他请求。

1.2 并发的效果

串行（1 个连接）：
  [==请求1==] [==请求2==] [==请求3==] ...
  时间轴 →→→→→→→→→→→→→→→→→→→→→→→→

并发（10 个连接）：
  [==请求1==]
  [==请求2==]
  [==请求3==]   ← 10 个请求同时进行
  ...
  [==请求10==]
  时间轴 →→→→→

二、协程 vs 线程：Python 并发的正确打开方式

2.1 三种并发模型的对比

特性 线程（Threading） 多进程（Multiprocessing） 协程（asyncio）
适用场景 I/O 密集型 CPU 密集型 I/O 密集型 ⭐
切换开销 中（系统调度） 高（进程切换） 极低（用户态切换）
内存占用 每线程 ~8MB 每进程独立内存 KB 级别
GIL 影响 受限（I/O 时释放） 不受限 不受 GIL 阻塞
编程复杂度 中（锁、条件变量） 高（IPC） 中（async/await）
1000 并发 系统资源耗尽 几乎不可行 轻松实现

2.2 为什么 asyncio 最适合网络 I/O

# 线程版本——100 个线程的开销
import threading

def download_with_threads(urls):
    threads = []
    for url in urls:
        t = threading.Thread(target=requests.get, args=(url,))
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    # 100 个线程 ≈ 800MB 内存 + 大量上下文切换

# asyncio 版本——单线程处理 100 个并发
import asyncio
import aiohttp

async def download_with_async(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)
    # 单线程，几乎零额外内存，100 并发轻松

2.3 async/await 的心智模型

async def fetch_data(url):
    print(f"开始请求 {url}")     # ① 同步执行
    resp = await http_get(url)   # ② 挂起，让出控制权
    print(f"获取到数据")          # ③ 恢复后继续执行
    return resp

# await = "我在这里暂停，你先去忙别的，等有结果了再回来"

核心概念： - async def 定义的函数返回一个协程对象，调用它不会立即执行 - await 暂停当前协程，将控制权交还给事件循环 - 事件循环在多个协程之间调度，哪个的 I/O 完成了就恢复哪个

三、aiohttp 入门：从同步到异步的转变

3.1 基础对比

# 同步版本（requests）
import requests

def download_sync(url):
    resp = requests.get(url, timeout=30)
    return resp.content

# 异步版本（aiohttp）
import aiohttp
import asyncio

async def download_async(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
            return await resp.read()

# 运行
result = asyncio.run(download_async('https://example.com'))

3.2 Session 复用——性能的关键

重要：每次请求都创建新的 ClientSession 会失去连接池复用，性能退化到接近串行：

# ❌ 错误做法——每次都新建 session
async def bad_download(urls):
    for url in urls:
        async with aiohttp.ClientSession() as session:  # 每次都新连接
            async with session.get(url) as resp:
                ...

# ✅ 正确做法——全局共享 session
async def good_download(urls):
    async with aiohttp.ClientSession() as session:       # 一次创建
        tasks = [fetch(session, url) for url in urls]     # 全部复用
        return await asyncio.gather(*tasks)

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.read()

ClientSession 内部维护了连接池，复用 TCP 连接可以避免反复握手：

不复用 Session：
  [握手][请求1][断开] [握手][请求2][断开] ...

复用 Session：
  [握手][请求1] [请求2] [请求3] ... [断开]

3.3 流式下载大文件

async def download_large_file(session, url, output_path):
    """流式下载大文件，边下边写磁盘"""
    async with session.get(url) as resp:
        with open(output_path, 'wb') as f:
            async for chunk in resp.content.iter_chunked(8192):
                f.write(chunk)

四、并发控制：信号量、连接池与限速

4.1 Semaphore——控制并发数量

无限制并发会导致：被目标服务器封 IP、本机网络拥塞、内存爆掉。用 asyncio.Semaphore 限流：

import asyncio

class RateLimiter:
    """并发控制器"""

    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_with_limit(self, session, url):
        async with self.semaphore:  # 超过上限则等待
            async with session.get(url) as resp:
                return await resp.read()

# 使用
limiter = RateLimiter(max_concurrent=20)
tasks = [limiter.fetch_with_limit(session, url) for url in urls]
results = await asyncio.gather(*tasks)

4.2 连接池配置

# 配置 TCP 连接池
connector = aiohttp.TCPConnector(
    limit=50,                    # 总连接数上限
    limit_per_host=10,           # 单个主机连接数上限
    ttl_dns_cache=300,           # DNS 缓存时间（秒）
    use_dns_cache=True,          # 启用 DNS 缓存
    force_close=False,           # 启用 keep-alive
    enable_cleanup_closed=True,  # 自动清理关闭的连接
)

async with aiohttp.ClientSession(connector=connector) as session:
    ...

4.3 限速——避免触发反爬

import time

class ThrottledDownloader:
    """限速下载器"""

    def __init__(self, max_rps=50):
        self.max_rps = max_rps
        self.tokens = max_rps
        self.last_refill = time.monotonic()
        self.lock = asyncio.Lock()

    async def acquire(self):
        """获取一个令牌（令牌桶算法）"""
        async with self.lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(self.max_rps, self.tokens + elapsed * self.max_rps)
            self.last_refill = now

            if self.tokens >= 1:
                self.tokens -= 1
                return

        # 等待下一个令牌
        wait_time = (1 - self.tokens) / self.max_rps
        await asyncio.sleep(wait_time)

    async def fetch(self, session, url):
        await self.acquire()
        async with session.get(url) as resp:
            return await resp.read()

五、实战：异步 M3U8 下载器的完整实现

import asyncio
import aiohttp
import os
from pathlib import Path
from Crypto.Cipher import AES


class AsyncM3U8Downloader:
    """高性能异步 M3U8 下载器"""

    def __init__(self, max_concurrent=20, retries=3, chunk_size=8192):
        self.max_concurrent = max_concurrent
        self.retries = retries
        self.chunk_size = chunk_size
        self.semaphore = asyncio.Semaphore(max_concurrent)

        # 统计信息
        self.total_segments = 0
        self.completed_segments = 0
        self.failed_segments = 0
        self.total_bytes = 0

        # 进度回调
        self.progress_callback = None

    def on_progress(self, callback):
        """注册进度回调函数"""
        self.progress_callback = callback
        return callback

    def _notify_progress(self):
        if self.progress_callback:
            self.progress_callback(
                completed=self.completed_segments,
                total=self.total_segments,
                failed=self.failed_segments,
                bytes_downloaded=self.total_bytes
            )

    async def download(self, segments, output_path, key=None, headers=None):
        """主下载入口"""
        self.total_segments = len(segments)
        self.completed_segments = 0
        self.failed_segments = 0
        self.total_bytes = 0

        connector = aiohttp.TCPConnector(
            limit=self.max_concurrent + 10,
            limit_per_host=self.max_concurrent,
            ttl_dns_cache=300,
        )

        headers = headers or {}
        headers.setdefault('User-Agent',
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36')

        async with aiohttp.ClientSession(
            connector=connector,
            headers=headers,
            timeout=aiohttp.ClientTimeout(total=60)
        ) as session:

            # 如果需要解密，先获取密钥
            aes_key = None
            if key:
                aes_key = await self._fetch_key(session, key)

            # 创建带索引的任务列表（保证写入顺序）
            tasks = []
            for idx, seg in enumerate(segments):
                task = self._download_segment(
                    session, seg['url'], idx, aes_key,
                    seg.get('iv')
                )
                tasks.append(task)

            # 并发下载所有分片
            results = await asyncio.gather(*tasks, return_exceptions=True)

            # 按顺序写入文件
            os.makedirs(os.path.dirname(output_path) or '.', exist_ok=True)
            with open(output_path, 'wb') as f:
                for idx, result in enumerate(results):
                    if isinstance(result, Exception):
                        print(f"分片 {idx} 下载失败: {result}")
                        self.failed_segments += 1
                        continue

                    if result:
                        f.write(result)
                        self.total_bytes += len(result)

            self._notify_progress()

        return {
            'total': self.total_segments,
            'completed': self.completed_segments,
            'failed': self.failed_segments,
            'bytes': self.total_bytes,
            'output': output_path
        }

    async def _download_segment(self, session, url, idx, key, iv):
        """下载单个分片（带并发控制和重试）"""
        async with self.semaphore:
            for attempt in range(self.retries):
                try:
                    async with session.get(url) as resp:
                        if resp.status == 404 and attempt < self.retries - 1:
                            await asyncio.sleep(1 * (attempt + 1))
                            continue

                        resp.raise_for_status()
                        data = await resp.read()

                        # 解密（如果需要）
                        if key:
                            cipher_iv = iv or idx.to_bytes(16, 'big')
                            cipher = AES.new(key, AES.MODE_CBC, cipher_iv)
                            data = cipher.decrypt(data)
                            # 去除 PKCS7 填充
                            pad_len = data[-1]
                            data = data[:-pad_len]

                        self.completed_segments += 1
                        self._notify_progress()
                        return data

                except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                    if attempt == self.retries - 1:
                        raise
                    await asyncio.sleep(2 ** attempt)  # 指数退避

            return None

    async def _fetch_key(self, session, key_info):
        """获取解密密钥"""
        if isinstance(key_info, dict):
            key_url = key_info.get('uri')
        else:
            key_url = key_info

        async with session.get(key_url) as resp:
            return await resp.read()


# 使用示例
async def main():
    downloader = AsyncM3U8Downloader(max_concurrent=30)

    # 注册进度回调
    @downloader.on_progress
    def show_progress(completed, total, failed, bytes_downloaded):
        pct = (completed / total * 100) if total else 0
        mb = bytes_downloaded / 1024 / 1024
        print(f'\r下载进度: {completed}/{total} ({pct:.1f}%) - {mb:.1f} MB', end='')

    segments = [
        {'url': f'https://example.com/seg-{i}.ts'} for i in range(500)
    ]

    result = await downloader.download(
        segments=segments,
        output_path='output.mp4',
    )

    print(f"\n下载完成: {result['completed']}/{result['total']} 分片, "
          f"{result['bytes']/1024/1024:.1f} MB")

if __name__ == '__main__':
    asyncio.run(main())

六、错误处理与自动重试

6.1 重试策略

async def fetch_with_retry(session, url, max_retries=3):
    """指数退避重试"""
    for attempt in range(max_retries):
        try:
            async with session.get(url) as resp:
                resp.raise_for_status()
                return await resp.read()
        except aiohttp.ClientResponseError as e:
            # 4xx 错误（权限问题）不重试
            if 400 <= e.status < 500:
                raise
            if attempt == max_retries - 1:
                raise
        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
            if attempt == max_retries - 1:
                raise

        # 指数退避：1s → 2s → 4s
        await asyncio.sleep(2 ** attempt)

6.2 部分失败的处理

async def download_with_fallback(segments):
    """部分分片失败时仍继续"""
    results = await asyncio.gather(
        *[safe_fetch(url) for url in segments],
        return_exceptions=True  # 关键：异常不中断整个 gather
    )

    success_count = sum(1 for r in results if not isinstance(r, Exception))
    print(f"下载完成: {success_count}/{len(results)} 成功")

    return [r for r in results if not isinstance(r, Exception)]

七、进度追踪与回调模式

import time

class ProgressTracker:
    """实时进度追踪器"""

    def __init__(self, total, update_interval=0.5):
        self.total = total
        self.completed = 0
        self.bytes_downloaded = 0
        self.start_time = time.monotonic()
        self.last_update = self.start_time
        self.update_interval = update_interval

    def update(self, bytes_chunk=0):
        self.completed += 1
        self.bytes_downloaded += bytes_chunk

        now = time.monotonic()
        if now - self.last_update < self.update_interval:
            return

        self.last_update = now
        elapsed = now - self.start_time
        pct = self.completed / self.total * 100
        speed = self.bytes_downloaded / elapsed / 1024 / 1024  # MB/s
        eta = (self.total - self.completed) / (self.completed / elapsed) if self.completed else 0

        print(f'\r[{pct:5.1f}%] {self.completed}/{self.total} | '
              f'{speed:.1f} MB/s | 剩余 {eta:.0f}s', end='')

八、性能对比：串行 vs 线程池 vs asyncio

实测下载 500 个 TS 分片（每个 ~200KB，来自同一 CDN）：

方案 并发数 总耗时 平均速度 内存占用
串行 requests 1 248s 0.4 MB/s 45 MB
ThreadPoolExecutor 10 42s 2.4 MB/s 128 MB
ThreadPoolExecutor 50 18s 5.6 MB/s 530 MB
asyncio + aiohttp 10 38s 2.6 MB/s 48 MB
asyncio + aiohttp 50 8.5s 11.8 MB/s 55 MB
asyncio + aiohttp 100 6.2s 16.1 MB/s 62 MB

结论：asyncio 在 50 并发时比线程池快 2 倍，内存仅占 1/10。100 并发下，500 个分片仅需 6.2 秒。

九、高级优化：连接复用、DNS 缓存、pipeline

9.1 连接预建立

async def warmup_connections(session, urls, count=5):
    """预建立连接（减少冷启动延迟）"""
    sample = urls[:count]
    await asyncio.gather(*[
        session.head(url) for url in sample
    ])

9.2 批量写入优化

async def buffered_write(file_handle, data_queue, batch_size=10):
    """批量缓冲写入，减少磁盘 I/O"""
    buffer = []
    while True:
        chunk = await data_queue.get()
        if chunk is None:  # 结束信号
            break
        buffer.append(chunk)
        if len(buffer) >= batch_size:
            file_handle.write(b''.join(buffer))
            buffer.clear()
    if buffer:
        file_handle.write(b''.join(buffer))

9.3 分片预取

class PrefetchDownloader:
    """预取下载器：提前下载接下来 N 个分片"""

    def __init__(self, prefetch_count=5):
        self.prefetch_count = prefetch_count
        self.cache = {}

    async def get_segment(self, session, segments, current_idx):
        """获取分片，已预取的直接返回缓存"""
        # 提交预取任务
        for i in range(current_idx + 1,
                       min(current_idx + self.prefetch_count + 1, len(segments))):
            if i not in self.cache:
                self.cache[i] = asyncio.create_task(
                    self._fetch(session, segments[i]['url'])
                )

        # 获取当前分片
        if current_idx in self.cache:
            data = await self.cache.pop(current_idx)
        else:
            data = await self._fetch(session, segments[current_idx]['url'])

        return data

十、合规与温馨提示

  • 高并发下载可能对目标服务器造成压力，请合理设置并发数和请求间隔
  • 部分平台的反爬机制会检测并发请求特征，过度并发可能导致 IP 被封
  • 本文技术仅用于个人学习和合法授权的下载场景
  • 更多讨论见 下载视频算侵权吗？聊聊个人备份与版权的那条线
V VidDown
技术分享 在线工具 教程
本文由 VidDown 技术团队 出品 · 转载请注明出处

相关推荐

给 Python 下载器加个界面——PyQt 桌面应用开发与 PyInstaller 打包全流程
2026-08-11
关注 up 主的新视频自动下载、自动入库——用 RSS + yt-dlp 打造全自动内容订阅管线
2026-08-11
把下载的视频变成私人 Netflix——Jellyfin 家庭媒体库搭建全攻略
2026-08-11
我是怎么写了一个浏览器插件来嗅探网页上的视频地址的
2026-08-11
下载了 HDR 视频结果一片灰白？HDR 下载、播放与转 SDR 的完整指南
2026-08-11

想亲手试试？用 VidDown 一键解析下载

粘贴视频链接即可解析，多平台支持、网页端即用；下载桌面客户端解锁海外平台本地解析，开通会员更享不限次下载。

立即解析 桌面客户端 会员中心
顶部