基于asyncio的文件下载脚本优化：并发、容错及边下边播需求

import asyncio
import os.path
import shutil
import aiofiles

import aiohttp
import lxml.html as htmlparser
import cssselect
import regex, json
from tempfile import TemporaryDirectory

domain = "https://doma.com/"
url = 'https://doma.com/ust/xxxx'
CONTENT_ID = regex.compile(r"/ust/([^?#&/]+)")

def parts_generator(size, start=0, part_size=5 * 1024 ** 2):
    while size - start > part_size:
        yield start, start + part_size
        start += part_size
    yield start, size
    
async def main():
 async def download(url, headers, save_path):
        async with session.get(url, headers=headers) as request:
         file = await aiofiles.open(save_path, 'wb')
         await file.write(await request.content.read())
 
 async with aiohttp.ClientSession() as session:
  async with session.get(url) as first:
   cs = await first.text()
   csrf_token = htmlparser.fromstring(cs).cssselect("meta[name='csrf-token']")[0].get("content")
   content_id = CONTENT_ID.search(url).group(1)
   Headers={"x-requested-with": "XMLHttpRequest", "x-csrf-token": csrf_token}
   async with session.post(domain + "api/get&user=xxx&pass=yyy", headers=Headers, json={"id": content_id}) as resp:
       res = json.loads(await resp.text())
       re = res['result']['Original']['file']
   async with session.get(re) as request:
       size = request.content_length
       tasks = []
       file_parts = []
       filename = 'File.mp4'
       tmp_dir = TemporaryDirectory(prefix=filename, dir=os.path.abspath('.'))
       for number, sizes in enumerate(parts_generator(size)):
        part_file_name = os.path.join(tmp_dir.name, f'{filename}.part{number}')
        file_parts.append(part_file_name)
        tasks.append(await download(re, {'Range': f'bytes={sizes[0]}-{sizes[1]}'}, part_file_name))
       await asyncio.gather(*tasks)
       with open(filename, 'wb') as wfd:
        for f in file_parts:
         with open(f, 'rb') as fd:
          shutil.copyfileobj(fd, wfd)
           
asyncio.run(main())

1. asyncio vs 线程池/多进程：哪个更适合下载场景？ #

下载属于IO密集型任务，asyncio是最优选择：

• 线程池：线程切换存在内核态开销，大量并发线程会占用更多系统资源，且Python的GIL会限制线程并行效率，IO密集场景下性能不如asyncio。
• 多进程：多进程专为CPU密集型任务设计，下载任务CPU占用极低，多进程反而会增加进程间通信、内存占用的额外开销，完全没必要。
• asyncio：协程切换是用户态操作，开销极小，单线程就能处理大量并发IO请求，资源利用率更高，更适配这类场景。

2. 修复并发下载功能 #

原脚本核心问题是tasks.append(await download(...))中使用了await，导致任务串行执行。正确做法是直接添加协程对象到任务列表，最后用asyncio.gather并发执行：

修改后的核心片段：

# 移除await，直接添加协程到任务列表
tasks.append(download(re, {'Range': f'bytes={sizes[0]}-{sizes[1]-1}'}, part_file_name))
# 并发执行所有任务
await asyncio.gather(*tasks)

注意：HTTP规范中Range: bytes=start-end是闭区间，原代码的start+part_size会导致分段重叠，需将结束字节减1（如第一个分段0-5242879对应5MB，第二个分段从5242880开始）。

3. 实现空payload的重试机制 #

在download函数中添加循环重试逻辑，限制重试次数并加入休眠：

async def download(url, headers, save_path, max_retries=3, retry_delay=2):
    for attempt in range(max_retries):
        try:
            async with session.get(url, headers=headers, timeout=10) as request:
                if request.status not in (206, 200):
                    raise Exception(f"Invalid status code: {request.status}")
                content = await request.content.read()
                if not content:
                    raise Exception("Empty payload received")
                async with aiofiles.open(save_path, 'wb') as file:
                    await file.write(content)
                return
        except Exception as e:
            print(f"Attempt {attempt+1} failed: {str(e)}, retrying in {retry_delay}s...")
            await asyncio.sleep(retry_delay)
    raise Exception(f"Failed after {max_retries} attempts")

• 添加超时设置避免无限等待
• 校验响应状态码（206对应Range请求的部分内容成功）
• 空payload触发重试，达到最大次数后抛出最终异常

4. 优化下载速度与可靠性 #

要追平IDM的速度，需从以下几点优化：

1. 调整分段大小：默认5MB可改为10-20MB，减少请求次数
2. 控制并发数：根据服务器限制设置8-16个并发任务（IDM默认8线程）
3. 流式写入文件：分块读取写入，减少内存占用，提升速度：

   async with aiofiles.open(save_path, 'wb') as file:
       async for chunk in request.content.iter_chunked(1024*1024):
           await file.write(chunk)
   

4. 复用TCP连接：aiohttp的ClientSession默认复用连接，避免重复建立连接的开销
5. 多镜像下载：若API返回多个文件源，可并发测试速度，选择最快的源进行分段下载
6. 进度监控：实时跟踪下载进度，方便动态调整并发数与分段大小

5. 实现边下载边播放功能 #

核心是预分配文件空间并写入对应偏移位置，而非使用临时分段文件：

实现步骤： #

1. 预分配与目标文件大小一致的空文件：

   filename = 'File.mp4'
   # 预分配空间
   with open(filename, 'wb') as f:
       f.seek(size - 1)
       f.write(b'\0')
   

2. 修改download函数，写入到文件对应偏移位置：

   async def download(url, headers, save_path, file_offset, max_retries=3, retry_delay=2):
       for attempt in range(max_retries):
           try:
               async with session.get(url, headers=headers, timeout=10) as request:
                   if request.status not in (206, 200):
                       raise Exception(f"Invalid status code: {request.status}")
                   async with aiofiles.open(save_path, 'rb+') as file:
                       await file.seek(file_offset)
                       async for chunk in request.content.iter_chunked(1024*1024):
                           await file.write(chunk)
                   return
           except Exception as e:
               print(f"Attempt {attempt+1} failed: {str(e)}, retrying in {retry_delay}s...")
               await asyncio.sleep(retry_delay)
       raise Exception(f"Failed after {max_retries} attempts")
   

3. 主逻辑中直接传入偏移量，无需临时目录：

   tasks = []
   filename = 'File.mp4'
   # 预分配文件
   with open(filename, 'wb') as f:
       f.seek(size - 1)
       f.write(b'\0')
   for number, (start, end) in enumerate(parts_generator(size)):
       headers = {'Range': f'bytes={start}-{end-1}'}
       tasks.append(download(re, headers, filename, start))
   await asyncio.gather(*tasks)
   

注意： #

MP4元数据通常在文件末尾，部分播放器无法直接播放未完成文件，建议使用VLC等支持流式播放的播放器，或优先下载文件开头的元数据片段（若服务器支持）。

内容的提问来源于stack exchange，提问作者Tathastu Pandya