使用requests库下载1GB+大文件遇超时/连接错误的优化方案咨询

核心问题分析 #

处理1-5GB大文件时，requests默认单连接下载易因超时、服务器断开导致失败，且过小的chunk_size会增加IO开销拖慢速度。以下是针对性优化方案：

方案1：断点续传（解决连接中断核心方案） #

利用HTTP的Range头实现断点续传，下载中断后可从上次位置继续，无需重新下载整个文件。

示例代码：

import requests
import os

def download_large_file(url, file_path, chunk_size=1024*1024):  # 1MB块大小
    # 获取文件总大小
    file_size = 0
    try:
        head_resp = requests.head(url, headers={"Accept-Encoding": "identity"})
        file_size = int(head_resp.headers.get("Content-Length", 0))
    except Exception as e:
        print(f"获取文件大小失败: {e}")
    
    # 检查本地已下载进度
    downloaded_size = os.path.getsize(file_path) if os.path.exists(file_path) else 0
    
    if downloaded_size >= file_size and file_size != 0:
        print("文件已下载完成")
        return
    
    # 设置Range头，从断点位置续传
    headers = {"Range": f"bytes={downloaded_size}-"}
    with requests.get(url, headers=headers, stream=True, timeout=30) as r:
        r.raise_for_status()
        with open(file_path, "ab") as f:  # 追加模式写入
            for chunk in r.iter_content(chunk_size=chunk_size):
                if chunk:
                    f.write(chunk)
                    # 非必要情况下注释flush，减少IO开销
                    # f.flush()

方案2：添加自动重试机制 #

针对服务器连接中断、超时等异常，给requests添加重试策略，自动恢复连接。

示例代码：

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_retry_session(retries=3, backoff_factor=0.3):
    session = requests.Session()
    retry = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=backoff_factor,
        status_forcelist=[500, 502, 503, 504]  # 针对服务器错误自动重试
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session

# 使用带重试的会话执行下载
session = create_retry_session()
with session.get(url, stream=True, timeout=60) as r:
    with open(file_path, "wb") as f:
        for chunk in r.iter_content(chunk_size=1024*1024):
            if chunk:
                f.write(chunk)

方案3：优化chunk_size与IO操作 #

• chunk_size建议设置为1MB-8MB（1024*1024 ~ 8*1024*1024），过小会增加循环次数和IO调用，过大则占用过多内存。
• 除非需要实时监控下载进度，否则不要频繁调用flush()，系统会自动处理缓冲区写入，强制flush会大幅增加IO开销。

方案4：多线程分块并行下载 #

将文件拆分为多个块，用多线程同时下载不同区块，最后合并，可大幅提升下载速度（需服务器支持Range请求）。

示例代码：

import requests
import os
from threading import Thread

def download_chunk(url, start, end, file_path, chunk_id):
    headers = {"Range": f"bytes={start}-{end}"}
    with requests.get(url, headers=headers, stream=True, timeout=30) as r:
        r.raise_for_status()
        with open(f"{file_path}.part{chunk_id}", "wb") as f:
            for chunk in r.iter_content(chunk_size=1024*1024):
                if chunk:
                    f.write(chunk)

def parallel_download(url, file_path, num_threads=4):
    # 获取文件总大小
    head_resp = requests.head(url, headers={"Accept-Encoding": "identity"})
    file_size = int(head_resp.headers.get("Content-Length", 0))
    chunk_size = file_size // num_threads
    
    # 启动多线程下载
    threads = []
    for i in range(num_threads):
        start = i * chunk_size
        end = start + chunk_size - 1 if i != num_threads-1 else file_size - 1
        thread = Thread(target=download_chunk, args=(url, start, end, file_path, i))
        threads.append(thread)
        thread.start()
    
    # 等待所有线程完成
    for thread in threads:
        thread.join()
    
    # 合并分块文件
    with open(file_path, "wb") as f:
        for i in range(num_threads):
            part_file = f"{file_path}.part{i}"
            with open(part_file, "rb") as pf:
                f.write(pf.read())
            os.remove(part_file)

额外注意事项 #

• 必须设置stream=True，避免将整个文件加载到内存。
• 调整timeout参数至30-60秒，避免因服务器响应慢导致超时。
• 部分服务器会限制单连接速度或并发数，可根据实际情况调整线程数。

内容的提问来源于stack exchange，提问作者G Naik