文件上传接口设计：从单文件到分片上传，一篇把 multipart / Base64 / 分片 / 断点续传 / 秒传讲透
去年帮一个视频团队做后台，用户上传 1.2GB 的素材，传到 80% 断网，重连后又得从 0 开始。那天我重新审视了一下「上传」这件事——看起来是 HTTP 最基础的功能，真要做得稳，里面的道道比想象中多。
这篇整理一下从最简单的单文件上传，一步步走到分片、断点续传、秒传的完整演进。代码用 Python（FastAPI + requests），能直接跑。
先抛三个重点：
- multipart、Base64、分片这几种上传方式不是「哪个更好」的关系，是「哪个场景该用哪个」——选型比实现重要
- 分片上传的核心不在切片，在状态管理。客户端断点续传、服务端合并校验、秒传判定，都围绕一个
file_hash展开 - 上线之后真正会让你加班的坑，藏在并发、分片大小、临时文件清理、安全这些地方
一、三种基础上传方式：先搞清楚为什么会有这么多方案
1.1 multipart/form-data：标准做法，但不是万能
HTTP 的 multipart 编码方式本身就是为文件设计的，浏览器 <input type="file"> 用的也是它。FastAPI 里接收特别简单：
# server.py —— multipart 上传
import aiofiles
from pathlib import Path
from fastapi import FastAPI, UploadFile, File
app = FastAPI()
UPLOAD_DIR = Path("./uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
@app.post("/upload/multipart")
async def upload_multipart(file: UploadFile = File(...)):
save_path = UPLOAD_DIR / file.filename
# 流式写入，避免大文件一次性读入内存
async with aiofiles.open(save_path, "wb") as f:
while chunk := await file.read(1024 * 1024): # 每次读 1MB
await f.write(chunk)
return {
"filename": file.filename, "size": save_path.stat().st_size}
客户端用 requests 也只要一行：
# client.py —— multipart 客户端
import requests
def upload_multipart(path):
with open(path, "rb") as f:
r = requests.post(
"http://localhost:8000/upload/multipart",
files={
"file": (path.split("/")[-1], f)},
)
return r.json()
看着挺美好，但有几个问题心里要有数：
- 没有断点续传。传一半网络抖一下，整文件得重来
- 大文件占内存。有些框架默认会把整个文件读到内存再处理。FastAPI 的
UploadFile走的是临时文件，问题不大，但用 Flask 时要小心request.files的行为 - 网关 / 代理有 body 限制。Nginx 默认
client_max_body_size 1m，不调一下 5MB 的文件都传不上去
所以 multipart 适合中小文件（几 MB 到几十 MB）、一次性提交、不需要续传的场景——头像、Excel 导入、表单附件这类。
1.2 Base64 上传：被诟病但又离不开
Base64 把二进制塞进 JSON，体积会膨胀约 33%。听起来挺蠢的，但有些场景它就是合适：
- 移动端走统一 JSON 接口，不想为了一个文件单独搞 multipart 客户端逻辑
- 文件很小（几 KB 的图标、配置），又必须走 HTTPS + JSON
- 跟其他字段一起提交（用户头像 + 昵称 + 简介放一个 JSON）
# server.py —— Base64 上传
import base64
from fastapi import Request, HTTPException
@app.post("/upload/base64")
async def upload_base64(request: Request):
body = await request.json()
filename = body["filename"]
raw = body["data"].encode()
# 关键：解码前先校验大小，防止有人塞 500MB 字符串把内存打满
# Base64 编码后体积约为原文件的 4/3，反推原始大小
estimated_size = len(raw) * 3 // 4
if estimated_size > 10 * 1024 * 1024: # 限制 10MB
raise HTTPException(413, "Base64 上传限制 10MB 以内")
data = base64.b64decode(raw)
save_path = UPLOAD_DIR / filename
async with aiofiles.open(save_path, "wb") as f:
await f.write(data)
return {
"filename": filename, "size": len(data)}
客户端：
# client.py —— Base64 客户端
import base64
def upload_base64(path):
with open(path, "rb") as f:
data = base64.b64encode(f.read()).decode()
r = requests.post(
"http://localhost:8000/upload/base64",
json={
"filename": path.split("/")[-1], "data": data},
)
return r.json()
1.3 三种方式对比
| 维度 | multipart | Base64 in JSON | 分片上传 |
|---|---|---|---|
| 协议复杂度 | 中 | 低 | 高 |
| 体积开销 | 几乎无 | +33% | 几乎无 |
| 大文件支持 | 一般 | 差 | 好 |
| 断点续传 | 不支持 | 不支持 | 支持 |
| 秒传 | 不支持 | 不支持 | 支持 |
| 客户端实现成本 | 简单 | 极简 | 复杂 |
一句话总结：小文件用 multipart，必须走 JSON 接口的小文件用 Base64，大文件用分片。
不要为了「看起来高级」一上来就分片——分片方案的复杂度会让你的客户端、服务端、运维都跟着受累。
二、分片上传：核心是状态管理
很多人理解分片上传就是「把大文件切成小块逐个 POST」，技术上没错，但工程上这只是冰山一角。真正的难点在于：
- 客户端怎么知道哪些块传过了（断点续传）
- 服务端怎么知道这个文件之前传过没（秒传）
- 合并的时候怎么保证数据没坏（完整性校验）
这三件事都指向同一个东西——文件指纹。
2.1 文件指纹：整个体系的基石
文件指纹就是文件内容的哈希，常用 MD5 或 SHA-1。秒传场景 MD5 就够了，追求安全用 SHA-256。
# client.py —— 计算文件 MD5
import hashlib
def file_md5(path):
hasher = hashlib.md5()
with open(path, "rb") as f:
while chunk := f.read(8192 * 1024): # 8MB 一读
hasher.update(chunk)
return hasher.hexdigest()
坑提示：文件越大，算 MD5 越慢。1GB 文件在普通机器上要几秒，用户体验上得加个「准备中…」的提示，不然用户以为卡死了。
更进阶的做法是分片指纹：每个分片单独算 MD5，整文件指纹是分片指纹的组合哈希。好处是能并发算，断点续传时也只需重算未传分片。但对绝大多数场景，整文件 MD5 已经够用。
2.2 分片上传完整流程
整个流程三步：检查 → 上传分片 → 合并。
第一步 check 是关键。客户端拿着 file_hash 来问服务端：「这个文件你见过吗？」服务端有三种回答：
- 见过且已合并完成 → 直接秒传，返回 URL
- 见过但没传完 → 返回已传分片号，客户端跳过这些
- 没见过 → 返回新的 upload_id，从头传
# server.py —— 检查接口（秒传判定 + 续传准备）
import hashlib
# 已上传文件指纹登记表（生产环境用 Redis / DB）
FILE_REGISTRY = {
} # {file_hash: "xxx.bin"}
# 分片上传任务状态
UPLOAD_TASKS = {
} # {upload_id: {"file_hash":..., "total_chunks":..., "uploaded": set()}}
CHUNK_DIR = Path("./chunks")
CHUNK_DIR.mkdir(exist_ok=True)
@app.post("/upload/check")
async def check_upload(file_hash: str = Form(...), total_chunks: int = Form(...)):
# ① 秒传判定：指纹命中，直接返回
if file_hash in FILE_REGISTRY:
return {
"instant": True, "url": f"/files/{FILE_REGISTRY[file_hash]}"}
# ② 生成或复用 upload_id
# 用 file_hash 派生，保证同一文件断点续传时拿到相同 upload_id
upload_id = hashlib.md5(f"{file_hash}{total_chunks}".encode()).hexdigest()[:16]
# ③ 扫描已传分片（断点续传）
chunk_dir = CHUNK_DIR / upload_id
uploaded = []
if chunk_dir.exists():
uploaded = [int(f.stem) for f in chunk_dir.glob("*.part")]
UPLOAD_TASKS[upload_id] = {
"file_hash": file_hash,
"total_chunks": total_chunks,
"uploaded": set(uploaded),
}
return {
"instant": False,
"upload_id": upload_id,
"uploaded_chunks": sorted(uploaded),
}
第二步是逐片上传。客户端跳过 uploaded_chunks 里的分片，只传剩余的：
# server.py —— 单分片上传
@app.post("/upload/chunk")
async def upload_chunk(
upload_id: str = Form(...),
chunk_index: int = Form(...),
chunk: UploadFile = File(...),
):
if upload_id not in UPLOAD_TASKS:
raise HTTPException(400, "upload_id 无效，请先调用 /upload/check")
chunk_dir = CHUNK_DIR / upload_id
chunk_dir.mkdir(exist_ok=True)
chunk_path = chunk_dir / f"{chunk_index}.part"
# 幂等：同一分片重复上传直接覆盖，不会出错
async with aiofiles.open(chunk_path, "wb") as f:
while data := await chunk.read(1024 * 1024):
await f.write(data)
UPLOAD_TASKS[upload_id]["uploaded"].add(chunk_index)
return {
"chunk_index": chunk_index, "received": True}
第三步合并。合并后必须重新算一遍 MD5 跟客户端给的对比——网络传输、磁盘写入都可能出错，不校验就是给自己埋雷。
# server.py —— 合并分片
import shutil
@app.post("/upload/merge")
async def merge_chunks(upload_id: str = Form(...)):
if upload_id not in UPLOAD_TASKS:
raise HTTPException(400, "upload_id 无效")
task = UPLOAD_TASKS[upload_id]
if len(task["uploaded"]) != task["total_chunks"]:
raise HTTPException(
400,
f"分片不完整: {len(task['uploaded'])}/{task['total_chunks']}"
)
chunk_dir = CHUNK_DIR / upload_id
final_name = f"{task['file_hash']}.bin"
final_path = UPLOAD_DIR / final_name
# 按顺序合并分片
async with aiofiles.open(final_path, "wb") as out:
for i in range(task["total_chunks"]):
async with aiofiles.open(chunk_dir / f"{i}.part", "rb") as part:
while data := await part.read(1024 * 1024):
await out.write(data)
# 完整性校验：合并后重新算 MD5，必须和客户端报的一致
hasher = hashlib.md5()
async with aiofiles.open(final_path, "rb") as f:
while data := await f.read(1024 * 1024):
hasher.update(data)
actual_hash = hasher.hexdigest()
if actual_hash != task["file_hash"]:
final_path.unlink(missing_ok=True) # 校验失败，删掉坏的
raise HTTPException(400, f"文件校验失败: 期望 {task['file_hash']}, 实际 {actual_hash}")
# 清理分片，登记指纹
shutil.rmtree(chunk_dir, ignore_errors=True)
FILE_REGISTRY[task["file_hash"]] = final_name
del UPLOAD_TASKS[upload_id]
return {
"url": f"/files/{final_name}", "size": final_path.stat().st_size}
2.3 断点续传的本质
回头看，「断点续传」其实就是一句话：每次上传前先 check 一次，跳过已传分片。
代码上就一个判断：
# client.py 片段
for i in range(total_chunks):
if i in done: # 这片传过了，跳过
f.seek(CHUNK_SIZE, 1)
continue
chunk = f.read(CHUNK_SIZE)
# 上传这一片
这玩意儿的价值在于：
- 网络中断后不用重来
- 用户主动暂停后可以接着传
- 同一文件在不同设备上可以续传（只要 file_hash 一致，upload_id 就一致）
2.4 秒传原理：用空间换时间
秒传的逻辑特别朴素：如果服务端已经有 file_hash 对应的完整文件，那这个文件根本不用再传一遍。
服务端维护一个 file_hash -> file_url 的映射表，每次 check 的时候查一下：
if file_hash in FILE_REGISTRY:
return {
"instant": True, "url": ...}
百度网盘的秒传就是这个原理。它有效，是因为互联网上大量文件是重复的——同一部电影、同一首歌、同一个安装包被成千上万人上传。对平台来说，省下的是 PB 级的存储和带宽。
秒传会失效的几种情况：
- 文件被哪怕改一个字节，hash 就变了，秒传失败
- 服务端清理了文件（比如 30 天未访问清理）
- file_hash 算法变了——所以 hash 算法一旦上线不要随便换，换了所有旧指纹都失效
三、上线才会遇到的坑
3.1 分片大小怎么选
太小：请求数爆炸，HTTP 头开销占比高，整体慢
太大：单片失败重传成本高，内存占用大
经验值：5MB ~ 10MB。1GB 文件用 5MB 分片约 200 片，单片失败重传也就 5MB，可接受。
10GB+ 的大文件可以用 20MB 甚至 50MB，但要注意网关的 body 限制（Nginx 默认 1MB，得调）。
3.2 并发上传
逐片上传太慢，客户端应该并发上传——但别无脑开几十个并发，不然服务端会被打爆。
# client.py —— 并发上传
import concurrent.futures
def upload_concurrent(path, upload_id, done, total_chunks, max_workers=4):
def upload_one(index, data):
files = {
"chunk": (f"chunk_{index}", data)}
form = {
"upload_id": upload_id, "chunk_index": index}
resp = requests.post(
"http://localhost:8000/upload/chunk", files=files, data=form
)
return resp.json()
# 注意：在主线程读文件，子线程只发送，避免多线程 read 同一 file 的竞态
tasks = []
with open(path, "rb") as f:
for i in range(total_chunks):
if i in done:
f.seek(CHUNK_SIZE, 1)
continue
tasks.append((i, f.read(CHUNK_SIZE)))
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as ex:
futures = [ex.submit(upload_one, i, data) for i, data in tasks]
for fut in concurrent.futures.as_completed(futures):
print(fut.result())
注意上面这个写法：在主线程把所有分片读出来再并发发，简单但内存占用大。更省内存的做法是用 seek 在子线程内读，但要加锁保证 file pointer 串行移动。
3.3 失败重试 + 退避
单片失败要重试，但重试次数要有上限（比如 3 次）。重试前最好加个退避（exponential backoff），不然服务端一抖，所有客户端同时重试，雪崩。
import time
def upload_with_retry(index, data, max_retry=3):
for attempt in range(max_retry):
try:
resp = requests.post(...)
if resp.ok:
return resp.json()
except requests.RequestException:
pass
# 指数退避：1s, 2s, 4s
time.sleep(2 ** attempt)
raise RuntimeError(f"分片 {index} 上传失败，重试 {max_retry} 次仍不成功")
3.4 临时分片清理：很容易被忽略
用户上传到一半放弃了，分片文件会一直留在服务器上。跑久了磁盘就满了。
解决方案：
- 定时任务扫描
chunks/下超过 1 小时没更新的目录，删掉 - 或者每个分片上传时记录时间戳，合并时校验是否过期
# server.py —— 启动时跑一次清理（生产环境改成定时任务）
import time
def cleanup_stale_chunks(max_age_seconds=3600):
now = time.time()
for chunk_dir in CHUNK_DIR.iterdir():
if not chunk_dir.is_dir():
continue
if now - chunk_dir.stat().st_mtime > max_age_seconds:
shutil.rmtree(chunk_dir, ignore_errors=True)
print(f"清理过期分片: {chunk_dir}")
3.5 安全：上传接口的几个底线
- 鉴权：upload_id 必须跟用户绑定。不然 A 用户能用 B 用户的 upload_id 续传 / 合并，越权
- 文件类型校验：别信
Content-Type，也别信扩展名，校验文件头（magic number）
# 校验 PNG 文件头
PNG_MAGIC = b"\x89PNG\r\n\x1a\n"
async def validate_png(file_path):
async with aiofiles.open(file_path, "rb") as f:
head = await f.read(len(PNG_MAGIC))
if head != PNG_MAGIC:
raise HTTPException(400, "不是合法的 PNG 文件")
- 限速：单用户上传带宽限制，防止恶意大文件占满带宽
- 路径穿越：合并时别直接用客户端给的文件名拼路径
# 错误示范
save_path = UPLOAD_DIR / filename # filename 可能是 "../../etc/passwd"
# 正确做法：用 file_hash 命名
save_path = UPLOAD_DIR / f"{file_hash}.bin"
完整可运行代码
把上面的片段拼起来，加上启动入口。
server.py
# server.py
import os
import json
import time
import shutil
import hashlib
from pathlib import Path
import aiofiles
from fastapi import FastAPI, UploadFile, File, Form, HTTPException, Request
app = FastAPI(title="文件上传服务")
UPLOAD_DIR = Path("./uploads")
CHUNK_DIR = Path("./chunks")
UPLOAD_DIR.mkdir(exist_ok=True)
CHUNK_DIR.mkdir(exist_ok=True)
# 已上传文件指纹登记表（生产用 Redis / DB）
FILE_REGISTRY = {
} # {file_hash: "xxx.bin"}
# 分片上传任务状态
UPLOAD_TASKS = {
} # {upload_id: {"file_hash":..., "total_chunks":..., "uploaded": set()}}
CHUNK_SIZE = 5 * 1024 * 1024 # 5MB
# ---------- 1. multipart ----------
@app.post("/upload/multipart")
async def upload_multipart(file: UploadFile = File(...)):
save_path = UPLOAD_DIR / file.filename
async with aiofiles.open(save_path, "wb") as f:
while chunk := await file.read(1024 * 1024):
await f.write(chunk)
return {
"filename": file.filename, "size": save_path.stat().st_size}
# ---------- 2. base64 ----------
@app.post("/upload/base64")
async def upload_base64(request: Request):
body = await request.json()
filename = body["filename"]
raw = body["data"].encode()
estimated_size = len(raw) * 3 // 4
if estimated_size > 10 * 1024 * 1024:
raise HTTPException(413, "Base64 上传限制 10MB 以内")
import base64
data = base64.b64decode(raw)
save_path = UPLOAD_DIR / filename
async with aiofiles.open(save_path, "wb") as f:
await f.write(data)
return {
"filename": filename, "size": len(data)}
# ---------- 3. 分片：检查 ----------
@app.post("/upload/check")
async def check_upload(file_hash: str = Form(...), total_chunks: int = Form(...)):
# 秒传
if file_hash in FILE_REGISTRY:
return {
"instant": True, "url": f"/files/{FILE_REGISTRY[file_hash]}"}
upload_id = hashlib.md5(f"{file_hash}{total_chunks}".encode()).hexdigest()[:16]
chunk_dir = CHUNK_DIR / upload_id
uploaded = []
if chunk_dir.exists():
uploaded = [int(f.stem) for f in chunk_dir.glob("*.part")]
UPLOAD_TASKS[upload_id] = {
"file_hash": file_hash,
"total_chunks": total_chunks,
"uploaded": set(uploaded),
}
return {
"instant": False,
"upload_id": upload_id,
"uploaded_chunks": sorted(uploaded),
}
# ---------- 3. 分片：上传单片 ----------
@app.post("/upload/chunk")
async def upload_chunk(
upload_id: str = Form(...),
chunk_index: int = Form(...),
chunk: UploadFile = File(...),
):
if upload_id not in UPLOAD_TASKS:
raise HTTPException(400, "upload_id 无效，请先调用 /upload/check")
chunk_dir = CHUNK_DIR / upload_id
chunk_dir.mkdir(exist_ok=True)
chunk_path = chunk_dir / f"{chunk_index}.part"
async with aiofiles.open(chunk_path, "wb") as f:
while data := await chunk.read(1024 * 1024):
await f.write(data)
UPLOAD_TASKS[upload_id]["uploaded"].add(chunk_index)
return {
"chunk_index": chunk_index, "received": True}
# ---------- 3. 分片：合并 ----------
@app.post("/upload/merge")
async def merge_chunks(upload_id: str = Form(...)):
if upload_id not in UPLOAD_TASKS:
raise HTTPException(400, "upload_id 无效")
task = UPLOAD_TASKS[upload_id]
if len(task["uploaded"]) != task["total_chunks"]:
raise HTTPException(
400, f"分片不完整: {len(task['uploaded'])}/{task['total_chunks']}"
)
chunk_dir = CHUNK_DIR / upload_id
final_name = f"{task['file_hash']}.bin"
final_path = UPLOAD_DIR / final_name
async with aiofiles.open(final_path, "wb") as out:
for i in range(task["total_chunks"]):
async with aiofiles.open(chunk_dir / f"{i}.part", "rb") as part:
while data := await part.read(1024 * 1024):
await out.write(data)
# 完整性校验
hasher = hashlib.md5()
async with aiofiles.open(final_path, "rb") as f:
while data := await f.read(1024 * 1024):
hasher.update(data)
actual_hash = hasher.hexdigest()
if actual_hash != task["file_hash"]:
final_path.unlink(missing_ok=True)
raise HTTPException(
400, f"文件校验失败: 期望 {task['file_hash']}, 实际 {actual_hash}"
)
shutil.rmtree(chunk_dir, ignore_errors=True)
FILE_REGISTRY[task["file_hash"]] = final_name
del UPLOAD_TASKS[upload_id]
return {
"url": f"/files/{final_name}", "size": final_path.stat().st_size}
# ---------- 清理过期分片（建议用定时任务调） ----------
@app.on_event("startup")
async def cleanup_on_start():
now = time.time()
for chunk_dir in CHUNK_DIR.iterdir():
if chunk_dir.is_dir() and now - chunk_dir.stat().st_mtime > 3600:
shutil.rmtree(chunk_dir, ignore_errors=True)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
client.py
# client.py
import os
import hashlib
import requests
from pathlib import Path
BASE_URL = "http://localhost:8000"
CHUNK_SIZE = 5 * 1024 * 1024 # 5MB
def file_md5(path):
"""整文件 MD5（大文件会慢，可改为分片指纹）"""
hasher = hashlib.md5()
with open(path, "rb") as f:
while chunk := f.read(8192 * 1024):
hasher.update(chunk)
return hasher.hexdigest()
# ---------- 1. multipart ----------
def upload_multipart(path):
with open(path, "rb") as f:
r = requests.post(
f"{BASE_URL}/upload/multipart",
files={
"file": (Path(path).name, f)},
)
return r.json()
# ---------- 2. base64 ----------
def upload_base64(path):
import base64
with open(path, "rb") as f:
data = base64.b64encode(f.read()).decode()
r = requests.post(
f"{BASE_URL}/upload/base64",
json={
"filename": Path(path).name, "data": data},
)
return r.json()
# ---------- 3. 分片 + 断点续传 + 秒传 ----------
def upload_chunked(path):
file_hash = file_md5(path)
total_chunks = (os.path.getsize(path) + CHUNK_SIZE - 1) // CHUNK_SIZE
# ① 先检查：可能秒传 / 续传
r = requests.post(
f"{BASE_URL}/upload/check",
data={
"file_hash": file_hash, "total_chunks": total_chunks},
)
info = r.json()
if info.get("instant"):
print(f"[秒传命中] 文件已存在: {info['url']}")
return info
upload_id = info["upload_id"]
done = set(info["uploaded_chunks"])
print(f"[断点续传] 已传 {len(done)}/{total_chunks} 片，从断点继续")
# ② 逐片上传（跳过已传）
with open(path, "rb") as f:
for i in range(total_chunks):
if i in done:
f.seek(CHUNK_SIZE, 1)
continue
chunk = f.read(CHUNK_SIZE)
files = {
"chunk": (f"chunk_{i}", chunk)}
form = {
"upload_id": upload_id, "chunk_index": i}
requests.post(f"{BASE_URL}/upload/chunk", files=files, data=form)
print(f"\r分片 {i + 1}/{total_chunks}", end="", flush=True)
print()
# ③ 合并
r = requests.post(f"{BASE_URL}/upload/merge", data={
"upload_id": upload_id})
return r.json()
if __name__ == "__main__":
import sys
path = sys.argv[1] if len(sys.argv) > 1 else "test.bin"
print("分片上传结果:", upload_chunked(path))
跑起来
pip install fastapi uvicorn aiofiles requests
uvicorn server:app --reload --port 8000
# 另一个终端
python client.py large_file.mp4
跑两遍同一个文件，第二次会看到 [秒传命中]——这就是秒传生效了。
选型决策
最后给个决策路径，照着选就行：
| 文件大小 / 场景 | 推荐方案 |
|---|---|
| < 5MB，普通业务 | multipart |
| < 5MB，必须走 JSON 接口 | Base64 |
| 5MB ~ 100MB | multipart（注意调网关限制） |
| > 100MB | 分片 + 断点续传 |
| 网盘类高频重复文件场景 | 分片 + 秒传 |
没有银弹。分片方案的复杂度高、运维成本高，能不做就不做。一旦做了，把状态管理、并发、清理、安全这四件事想清楚，基本就稳了。
剩下的，就是上线之后被各种边界 case 教育的过程了——比如断网、比如分片顺序乱、比如同一个 upload_id 被两个客户端同时合并。这些坑一个一个踩过来，你也就成了团队里上传这块的"专家"。
