自己写一个支持断点续传的分块下载器（Python 实战）
市面上的下载器（aria2、迅雷、IDM）已经很强，但有些场景你得自己撸一个：比如要在自家服务里内嵌下载能力、要按字节精确控制进度、要对接特殊的鉴权/签名逻辑、或者要在受限环境里跑。这篇文章用 Python 从零写一个支持 HTTP Range 分块、多线程、断点续传、校验完整性的下载器，把原理和坑一次讲透。
一、HTTP Range 是什么
普通 GET 一次拿全部内容。加上 Range 请求头后，可以只取其中一段：
GET /big.mp4 HTTP/1.1
Host: example.com
Range: bytes=0-1023
服务端回 206 Partial Content，只返回第 0–1023 字节。这就是分块下载和断点续传的底层基础。
二、最朴素的单段下载
import requests
url = 'https://example.com/big.mp4'
r = requests.get(url, stream=True)
with open('out.mp4', 'wb') as f:
for chunk in r.iter_content(8192):
f.write(chunk)
缺点：不支持断点，中途断了得从头再来；单连接速度受单 TCP 限制。
三、先探测服务器能力
head = requests.head(url, timeout=10)
accepts = head.headers.get('Accept-Ranges') == 'bytes'
size = int(head.headers.get('Content-Length', 0))
Accept-Ranges: bytes表示支持分块；Content-Length给出总大小；- 若不支持 Range，只能退化成单段下载。
四、分块多线程下载
把文件按大小切成 N 块，每块一个线程下到自己那一段，最后落进同一个预分配文件：
import requests, os
from concurrent.futures import ThreadPoolExecutor
def download_chunk(url, start, end, path):
headers = {'Range': f'bytes={start}-{end}'}
r = requests.get(url, headers=headers, stream=True, timeout=30)
with open(path, 'r+b') as f:
f.seek(start)
for chunk in r.iter_content(8192):
f.write(chunk)
def multi_download(url, path, threads=8):
size = int(requests.head(url, timeout=10).headers['Content-Length'])
open(path, 'wb').close()
with open(path, 'r+b') as f:
f.truncate(size) # 预分配空间，避免并发写越界
step = size // threads
tasks = []
for i in range(threads):
s = i * step
e = size - 1 if i == threads - 1 else s + step - 1
tasks.append((s, e))
with ThreadPoolExecutor(max_workers=threads) as ex:
for s, e in tasks:
ex.submit(download_chunk, url, s, e, path)
multi_download('https://example.com/big.mp4', 'out.mp4', 8)
关键点：每块写不同偏移（f.seek），范围不重叠，并发写同一个文件是安全的。
五、断点续传：记录已下字节
核心是维护一个进度文件（比如 .prog），记录每个块下到哪了。重启时跳过已完成部分：
import json, os
def resume_download(url, path, threads=8):
prog = path + '.prog'
size = int(requests.head(url, timeout=10).headers['Content-Length'])
if not os.path.exists(path):
open(path, 'wb').close()
with open(path, 'r+b') as f:
f.truncate(size)
done = {i: 0 for i in range(threads)}
else:
done = json.load(open(prog)) # 已下载字节数，按块记录
# 对每个块，Range 从已下字节开始接着下
...
json.dump(done, open(prog, 'w'))
完整实现按块粒度记录偏移，重启时 Range: bytes=start+downloaded- 接着下。下载完删掉 .prog。
六、健壮性：重试、限速、校验
生产级下载器还要考虑：
# 失败重试
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1))
def get_chunk(url, headers):
return requests.get(url, headers=headers, stream=True, timeout=30)
# 限速（避免打爆带宽）
import time
def throttle(gen, bytes_per_sec):
bucket = bytes_per_sec
for chunk in gen:
yield chunk
bucket -= len(chunk)
if bucket <= 0:
time.sleep(1); bucket = bytes_per_sec
# 完整性校验
import hashlib
def sha256(path):
h = hashlib.sha256()
with open(path, 'rb') as f:
for b in iter(lambda: f.read(65536), b''):
h.update(b)
return h.hexdigest()
七、和现成工具对比
| 维度 | 自写 | aria2 / curl -C |
|---|---|---|
| 可控性 | 完全自定义 | 高但受接口限制 |
| 开发成本 | 高 | 零 |
| 速度 | 取决于实现 | 极优（连接优化、磁盘调度） |
| 特殊需求 | 易内嵌 | 需外部调用 |
| 断点续传 | 自己实现 | 原生支持 |
结论：一般需求直接调 aria2c -x 16 -s 16 URL 或 curl -C - URL；要内嵌进自家服务、或对接特殊逻辑才自写。
八、常见翻车现场
| 现象 | 根因 | 解法 |
|---|---|---|
| 多线程写乱 | 偏移范围重叠 | 严格计算每块 [start,end) 不重叠 |
| 服务端不支持 Range | 返回 200 而非 206 | 退化单段下载 |
| 速度上不去 | 单 IP 被限流 | 上代理/多出口 |
| 文件损坏 | 某块缺失/重叠 | 下完校验大小或 hash |
| 断点失效 | 没记录偏移 | 维护进度文件 |
九、常见问题
Q：多线程写同一文件安全吗？ 只要每块写不同偏移且不重叠，就安全；预分配 truncate 避免末尾越界。
Q：Range 的 end 能省略吗？ 能，bytes=1000- 表示从 1000 到末尾，适合断点续传的"接着下"。
Q：为什么还要自己写？ 内嵌进服务、要自定义鉴权头、要做进度回调 UI、要适配私有协议时，现成工具接入成本高。
Q：能用异步（aiohttp）吗？ 能，IO 密集型用 asyncio 更省资源，思路一样：按块发 Range 请求，写各自偏移。
小结
分块下载器三要素：Range 请求头拿片段、预分配文件 + 按偏移并发写、进度文件实现断点续传。自写适合内嵌场景，普通下载直接用 aria2/curl 更省心。