Agent下载入门指南：从零搭建高可靠下载服务

1次阅读

共计 2754 个字符，预计需要花费 7 分钟才能阅读完成。

传统的下载工具（如浏览器内置下载、wget 等）在 Agent 自动化场景下常遇到三大问题：

连接不可靠：网络波动导致下载中断后需重新开始，浪费带宽和时间
缺乏细粒度控制：无法灵活暂停 / 恢复下载任务，难以集成到自动化流程
性能瓶颈：单线程下载无法充分利用现代多核 CPU 和高速网络

举个真实案例：某爬虫项目需批量下载 10GB 视频文件，使用传统工具时因网络问题失败 5 次，最终耗时是理论值的 3 倍。

特性	HTTP/1.1	FTP	自定义协议
断点续传支持	✓ (Range 头部)	✓ (REST 命令)	需自行实现
并发连接数	通常 6 - 8 个 / 域名	无限	可自定义
防火墙穿透性	优秀	较差	依赖实现
实现复杂度	低	中等	高

对于大多数 Agent 场景，HTTP 协议是最佳选择：

现代 CDN 普遍支持 Range 请求
无需额外服务端部署
天然适应云原生环境

import requests
from concurrent.futures import ThreadPoolExecutor

def download_chunk(url, start_byte, end_byte, retries=3):
    """下载指定字节范围的数据块"""
    headers = {'Range': f'bytes={start_byte}-{end_byte}'}

    for attempt in range(retries):
        try:
            resp = requests.get(url, headers=headers, timeout=10)
            resp.raise_for_status()
            return resp.content
        except Exception as e:
            if attempt == retries - 1:
                raise
            print(f"重试 {url} [{start_byte}-{end_byte}] (尝试 {attempt + 1}/{retries})")

class Downloader:
    def __init__(self, url, num_threads=4):
        self.url = url
        self.num_threads = num_threads

    def run(self):
        file_size = int(requests.head(self.url).headers['Content-Length'])
        chunk_size = file_size // self.num_threads

        with ThreadPoolExecutor(max_workers=self.num_threads) as executor:
            futures = []
            for i in range(self.num_threads):
                start = i * chunk_size
                end = start + chunk_size - 1 if i < self.num_threads - 1 else file_size - 1
                futures.append(executor.submit(download_chunk, self.url, start, end))

            # 按顺序拼接分片
            return b''.join(f.result() for f in futures)

import sqlite3
from pathlib import Path

class ResumeDownloader:
    def __init__(self, url, db_path='downloads.db'):
        self.url = url
        self.conn = sqlite3.connect(db_path)
        self._init_db()

    def _init_db(self):
        self.conn.execute('''CREATE TABLE IF NOT EXISTS downloads
                            (url TEXT PRIMARY KEY, path TEXT, downloaded INTEGER)''')

    def get_saved_progress(self):
        cursor = self.conn.cursor()
        cursor.execute("SELECT downloaded FROM downloads WHERE url=?", (self.url,))
        return cursor.fetchone()[0] if cursor.fetchone() else 0

    def save_progress(self, downloaded):
        self.conn.execute("INSERT OR REPLACE INTO downloads VALUES (?, ?, ?)",
                         (self.url, str(Path('downloads')/self.url.split('/')[-1]), downloaded))
        self.conn.commit()

当遇到不支持 Range 请求的服务端时，可以这样降级处理：

先发送 HEAD 请求检查 Accept-Ranges 头
若不存在则回退到单线程下载
记录该 URL 到黑名单避免下次重复检测

推荐线程数计算方式：

max_threads = min(
    CPU 核心数 * 2,
    带宽(Mbps) / 单个连接平均速度(Mbps)
)

实际测试表明：对于 100Mbps 带宽和 10Mbps/ 连接的 CDN，10 个线程比 4 线程快 2.3 倍，但 20 线程反而因 TCP 拥塞导致速度下降 15%。

在下载完成后添加校验步骤：

import hashlib

def verify_file(path, expected_md5):
    with open(path, 'rb') as f:
        file_hash = hashlib.md5()
        while chunk := f.read(8192):
            file_hash.update(chunk)
    return file_hash.hexdigest() == expected_md5

根据网络状况自动调整分片大小：

def calculate_chunk_size(file_size, last_speed_mbps):
    base_size = 1 * 1024 * 1024  # 1MB
    if last_speed_mbps > 50:     # 高速网络
        return min(base_size * 4, file_size // 20)
    else:                        # 低速网络
        return min(base_size, file_size // 10)