上传 2GB 文件别让 Django 接：分片上传、对象存储直传与秒传

我们的工具站有几个功能需要用户上传文件（视频处理、PDF 合并、格式转换）。小文件没问题，直到有人传了一个 2GB 的录屏。

当时的实现是标准的 Django 文件上传：<input type="file"> → request.FILES['file'] → 保存。后果：

Django 把上传内容先写到临时文件（还好，没进内存），但整个上传期间 worker 被占住；

上传花了 20 多分钟（用户网络慢），worker 就占了 20 多分钟；

中途网络断了 → 从头再来（用户崩溃）；

上传完成才发现文件类型不对 → 白传 2GB。

后来重做了整套上传：前端切片、并发传、直传对象存储、支持断点续传和秒传。服务端只在最后做一次合并和校验。这篇写完整的方案选择和实现细节。

TL;DR：大文件上传不要让应用服务器"中转"。推荐 前端直传对象存储（服务端只签发预签名 URL，字节不经过 Django）。流程：init（创建上传任务）→ 前端切片并发 PUT 到预签名 URL → complete（服务端校验 + 合并 + 入库）。断点续传靠"已上传分片列表"接口；秒传靠文件哈希查询（命中就直接完成）。三个必须做：分片要有序号和大小校验、合并后要校验总大小和哈希、服务端要清理未完成的碎片（否则对象存储里会堆一堆垃圾）。

一、为什么 Django 直接接大文件不行
二、三种架构怎么选
三、方案 A：分片上传到服务端再合并
四、方案 B：直传对象存储（推荐）
五、断点续传
六、秒传：同一个文件不用再传一遍
七、校验：分片、总量、类型
八、前端那些必须处理好的细节
九、服务端接口设计
十、实测数据与坑清单

一、为什么 Django 直接接大文件不行

Django 的文件上传机制其实已经比"读进内存"好很多了：

< 2.5MB：内存（InMemoryUploadedFile）；
> 2.5MB：写临时文件（TemporaryUploadedFile），磁盘。

但问题不在这里，而在整个请求模型：

问题	说明
worker 被长期占用	上传多久，worker 就占用多久（同步 worker 下尤其致命）
超时	nginx `proxy_read_timeout`、gunicorn `timeout` 都要设得比上传时间长
中断即重来	网络断了，2GB 白传
无法校验在先	传完了才知道类型不对、大小超限
多份副本	临时文件 → 目标位置 → 对象存储，搬了好几次
并发上传拖垮服务	几个大文件同时传就把 worker 占满

本质上和"大文件下载"是同一个问题：让应用服务器搬运字节。解决的思路也一样——让专业的组件做这件事（对象存储）。

二、三种架构怎么选

方案	字节路径	优点	缺点	适合
A. 服务端接收	浏览器 → nginx → Django → 磁盘	最简单	占 worker、无断点、无秒传	小文件（< 50MB）
B. 分片 → 服务端合并	浏览器 → Django（分片）→ 合并	能断点续传	字节仍经过 Django	中等规模、没有对象存储
C. 直传对象存储	浏览器 → 对象存储（预签名），Django 只签 URL	不占应用服务器、可断点、可秒传	实现复杂一些	推荐

我们的选择：

小文件（< 20MB，比如头像、封面）→ 方案 A（简单，够用）；
大文件（视频、大 PDF）→ 方案 C。

判断标准是"上传耗时是否可能超过几秒"。几秒以内的用 A，超过的用 C。

三、方案 A：分片上传到服务端再合并

如果你的环境没有对象存储，这个方案也能解决大部分问题。

服务端接口：

import os
import hashlib
from django.conf import settings
from django.core.files.storage import default_storage


def chunk_dir(upload_id):
    return os.path.join(settings.MEDIA_ROOT, 'chunks', upload_id)


def save_chunk(upload_id, index, chunk):
    """保存一个分片。文件名用序号，保证顺序。"""
    d = chunk_dir(upload_id)
    os.makedirs(d, exist_ok=True)
    path = os.path.join(d, f'{index:06d}.part')
    with open(path, 'wb+') as f:
        for c in chunk.chunks():
            f.write(c)
    return os.path.getsize(path)


def merge_chunks(upload_id, total, filename):
    """按序号合并所有分片。"""
    d = chunk_dir(upload_id)
    out_path = os.path.join(settings.MEDIA_ROOT, 'uploads', filename)

    h = hashlib.md5()
    total_size = 0
    with open(out_path, 'wb+') as out:
        for i in range(total):
            part = os.path.join(d, f'{i:06d}.part')
            if not os.path.exists(part):
                raise FileNotFoundError(f'缺少分片 {i}')
            with open(part, 'rb') as f:
                while True:
                    buf = f.read(1024 * 1024)
                    if not buf:
                        break
                    out.write(buf)
                    h.update(buf)
                    total_size += len(buf)
            os.remove(part)          # 合并一个删一个，避免双倍磁盘占用
    os.rmdir(d)
    return out_path, total_size, h.hexdigest()

注意"合并一个删一个"：2GB 的文件，如果所有分片 + 合并后的文件同时在磁盘上，就是 4GB。边合并边删能省一半空间。

这个方案的局限：字节还是经过了 Django，上传期间仍然占用 worker（虽然每个分片很短）。适合"没有对象存储"的场景。

四、方案 B：直传对象存储（推荐）

流程

1. 前端选择文件 → 计算 hash（可选，用于秒传）→ 调 /upload/init/
2. 服务端：创建 UploadTask（upload_id），返回 upload_id + 分片大小
3. 前端：按分片大小切文件
4. 对每个分片：调 /upload/presign/ 拿一个预签名 PUT URL
5. 前端：直接 PUT 到对象存储（不经过 Django）
6. 全部传完：调 /upload/complete/
7. 服务端：校验分片完整性 → 合并（或在对象存储侧合并）→ 入库

预签名 URL（MinIO / S3 协议）

用 boto3（兼容 S3 协议，MinIO 可以用）：

import boto3
from botocore.config import Config

s3 = boto3.client(
    's3',
    endpoint_url=settings.S3_ENDPOINT,          # 例如 http://127.0.0.1:9000
    aws_access_key_id=settings.S3_ACCESS_KEY,
    aws_secret_access_key=settings.S3_SECRET_KEY,
    config=Config(signature_version='s3v4'),
    region_name='us-east-1',
)


def presign_put(key: str, expires_in: int = 3600, content_type: str = '') -> str:
    """生成一个允许 PUT 上传的预签名 URL。"""
    params = {
        'Bucket': settings.S3_BUCKET,
        'Key': key,
    }
    if content_type:
        params['ContentType'] = content_type
    return s3.generate_presigned_url(
        'put_object',
        Params=params,
        ExpiresIn=expires_in,
        HttpMethod='PUT',
    )

MinIO 官方 SDK（minio-py）也提供：

from minio import Minio
from datetime import timedelta

client = Minio(
    '127.0.0.1:9000',
    access_key=..., secret_key=..., secure=False,
)
url = client.presigned_put_object('bucket', 'object-name', expires=timedelta(hours=1))

安全要点：

过期时间要短（1 小时以内，够传就行）；
key 由服务端生成（不要让前端决定存到哪里，否则能覆盖别人的文件）；
限制大小：S3 的预签名不支持内容长度限制（除非用 POST policy），所以要在 complete 时校验实际大小；
只允许 PUT 这一个操作（generate_presigned_url('put_object', ...) 生成的 URL 只能 PUT）。

大文件用 Multipart Upload

对象存储原生支持分片上传（S3 Multipart Upload），比"自己切片 + 自己合并"更好：

def init_multipart(key):
    resp = s3.create_multipart_upload(Bucket=settings.S3_BUCKET, Key=key)
    return resp['UploadId']


def presign_part(key, upload_id, part_number, expires_in=3600):
    return s3.generate_presigned_url(
        'upload_part',
        Params={
            'Bucket': settings.S3_BUCKET,
            'Key': key,
            'UploadId': upload_id,
            'PartNumber': part_number,
        },
        ExpiresIn=expires_in,
        HttpMethod='PUT',
    )


def complete_multipart(key, upload_id, parts):
    """parts: [{'ETag': '...', 'PartNumber': 1}, ...] 必须按 PartNumber 升序"""
    s3.complete_multipart_upload(
        Bucket=settings.S3_BUCKET,
        Key=key,
        UploadId=upload_id,
        MultipartUpload={'Parts': sorted(parts, key=lambda p: p['PartNumber'])},
    )

Multipart 的三个好处：

合并是在对象存储内部做的，不消耗你的带宽和 CPU；
每个分片可以独立重传（天然支持断点续传）；
可以并发传分片（我们实测 4 并发能跑满家用带宽）。

代价：

分片最小 5MB（最后一片除外）；
未完成的分片会占着空间，必须清理（见下）。

五、断点续传

核心是一个"已上传分片列表"接口：

def uploaded_parts(request, upload_id):
    """返回已上传的分片序号列表，前端据此跳过。"""
    task = get_object_or_404(UploadTask, upload_id=upload_id, user=request.user)

    # 方式 1：查自己的数据库记录
    parts = UploadPart.objects.filter(task=task).values_list('part_number', flat=True)

    # 方式 2：直接问对象存储（更权威）
    resp = s3.list_parts(
        Bucket=settings.S3_BUCKET,
        Key=task.object_key,
        UploadId=task.upload_id,
    )
    done = [{'PartNumber': p['PartNumber'], 'ETag': p['ETag'], 'Size': p['Size']}
            for p in resp.get('Parts', [])]

    return JsonResponse({'parts': done})

前端逻辑：

async function uploadFile(file, uploadId) {
    const CHUNK = 5 * 1024 * 1024;                     // 必须 >= 5MB（S3 要求）
    const total = Math.ceil(file.size / CHUNK);

    // 1. 问服务端哪些已经传过了
    const done = await fetch(`/upload/parts/${uploadId}/`).then(r => r.json());
    const doneSet = new Set(done.parts.map(p => p.PartNumber));

    const etags = [];
    // 2. 并发上传未完成的分片
    const queue = [];
    for (let i = 1; i <= total; i++) {
        if (!doneSet.has(i)) queue.push(i);
    }

    const CONCURRENCY = 4;
    async function worker() {
        while (queue.length) {
            const i = queue.shift();
            const start = (i - 1) * CHUNK;
            const blob = file.slice(start, Math.min(start + CHUNK, file.size));
            const {url} = await fetch(`/upload/presign/`, {
                method: 'POST',
                body: JSON.stringify({upload_id: uploadId, part: i}),
            }).then(r => r.json());
            const resp = await fetch(url, {method: 'PUT', body: blob});
            etags.push({PartNumber: i, ETag: resp.headers.get('ETag').replace(/"/g, '')});
        }
    }
    await Promise.all(Array.from({length: CONCURRENCY}, worker));

    // 3. 通知服务端完成
    await fetch('/upload/complete/', {
        method: 'POST',
        body: JSON.stringify({upload_id: uploadId, parts: etags}),
    });
}

断点续传的关键：upload_id 要持久化（存数据库），用户关掉页面再打开，用同一个 upload_id 继续——而不是重新开始。

我们的做法：上传任务记录关联用户 + 文件 hash + 大小，用户回到页面时前端先调 /upload/resume/?hash=xxx&size=xxx，服务端找出未完成的同 hash 任务返回，前端接着传。

六、秒传：同一个文件不用再传一遍

原理：上传前先算文件的哈希，问服务端"这个文件是不是已经有了"。有了就直接完成。

def check_exists(request):
    """秒传检查。"""
    file_hash = request.GET.get('hash')
    size = int(request.GET.get('size', 0))

    obj = UploadedFile.objects.filter(hash=file_hash, size=size, status='READY').first()
    if obj:
        # 已存在：直接给用户建一条引用，不用再传
        user_file = UserFile.objects.create(
            user=request.user,
            file=obj,
            filename=request.GET.get('filename'),
        )
        return JsonResponse({'exists': True, 'file_id': user_file.id})
    return JsonResponse({'exists': False})

哈希怎么算：

小文件（< 100MB）：全文件 MD5/SHA1，前端用 SparkMD5 增量算；
大文件：抽样哈希（首 1MB + 中间 1MB + 末 1MB + 文件大小），速度快，碰撞概率足够低；
更快：xxhash（比 MD5 快 5~10 倍），但前端要用 WASM 版本。

// 抽样哈希（大文件用）
async function sampleHash(file) {
    const size = file.size;
    const chunks = [
        file.slice(0, 1024 * 1024),
        file.slice(Math.floor(size / 2), Math.floor(size / 2) + 1024 * 1024),
        file.slice(Math.max(0, size - 1024 * 1024), size),
    ];
    const bufs = await Promise.all(chunks.map(c => c.arrayBuffer()));
    const combined = new Uint8Array(bufs.reduce((n, b) => n + b.byteLength, 0));
    let offset = 0;
    bufs.forEach(b => { combined.set(new Uint8Array(b), offset); offset += b.byteLength; });
    const digest = await crypto.subtle.digest('SHA-256', combined);
    return Array.from(new Uint8Array(digest)).map(b => b.toString(16).padStart(2, '0')).join('') + '_' + size;
}

安全考虑：

哈希 + 大小双重匹配（只匹配哈希可能被碰撞攻击）；
秒传成功的文件要做权限检查——不能因为"文件已存在"就把别人的文件关联给当前用户（我们只在"该文件对当前用户可见"或者"是公开素材"时才秒传，否则老老实实传）；
敏感文件不做秒传（哈希本身会泄露"你有这个文件"）。

实测效果：我们的工具站里，同一批素材被反复上传的比例大概 15%，秒传让这部分上传时间从"几十秒"变成"瞬间"。

七、校验：分片、总量、类型

三道校验：

1. 上传前（前端）：文件名、大小、扩展名（只是友好提示，不是安全校验）。

2. 分片时（服务端）：

def presign_part(request):
    upload_id = request.POST['upload_id']
    part_number = int(request.POST['part'])
    size = int(request.POST['size'])

    task = UploadTask.objects.get(upload_id=upload_id, user=request.user)
    # 分片大小必须在合理范围
    if not (5 * 1024 * 1024 <= size <= 100 * 1024 * 1024):
        return JsonResponse({'error': '分片大小不合法'}, status=400)
    # 分片序号不能超范围
    if not (1 <= part_number <= task.total_parts):
        return JsonResponse({'error': '分片序号不合法'}, status=400)
    ...

3. 完成后（服务端，最重要）：

def complete_upload(upload_id, parts):
    task = UploadTask.objects.get(upload_id=upload_id)

    # a. 向对象存储核实（不能只信前端传来的 etag 列表）
    resp = s3.list_parts(Bucket=..., Key=task.object_key, UploadId=task.upload_id)
    actual = {p['PartNumber']: p['ETag'] for p in resp.get('Parts', [])}
    for p in parts:
        if actual.get(p['PartNumber']) != p['ETag']:
            raise ValueError(f'分片 {p["PartNumber"]} 校验失败')

    # b. 合并（对象存储侧）
    s3.complete_multipart_upload(...)

    # c. 合并后核实总大小
    head = s3.head_object(Bucket=..., Key=task.object_key)
    real_size = head['ContentLength']
    if real_size != task.declared_size:
        # 大小不符 → 删除并拒绝
        s3.delete_object(Bucket=..., Key=task.object_key)
        raise ValueError(f'大小不符：{real_size} != {task.declared_size}')

    # d. 类型校验（读文件头）
    raw = s3.get_object(Bucket=..., Key=task.object_key, Range='bytes=0-2047')['Body'].read()
    mime = magic.from_buffer(raw, mime=True)
    if mime not in ALLOWED_MIME:
        s3.delete_object(Bucket=..., Key=task.object_key)
        raise ValueError(f'不支持的文件类型：{mime}')

    task.status = 'READY'
    task.save()

关键点：不能只信前端传来的信息。前端说"我传了 100 片、每片 5MB"，服务端一定要向对象存储核实（或者直接 head_object 看实际大小）。否则有人可以直接调 complete 接口伪造。

八、前端那些必须处理好的细节

1. 并发数：3~5 个比较合适。太多会被限流、也会让进度条跳得厉害。我们实测 4 并发在大部分网络下能跑满带宽。

2. 失败重试：分片上传失败很常见（网络抖动），要指数退避重试：

async function putWithRetry(url, blob, maxRetry = 3) {
    for (let i = 0; i < maxRetry; i++) {
        try {
            const resp = await fetch(url, {method: 'PUT', body: blob});
            if (resp.ok) return resp;
            throw new Error(`HTTP ${resp.status}`);
        } catch (e) {
            if (i === maxRetry - 1) throw e;
            await new Promise(r => setTimeout(r, Math.min(1000 * 2 ** i, 8000) + Math.random() * 500));
        }
    }
}

3. 哈希计算要放 Web Worker：主线程算 2GB 文件的哈希会卡死页面。

4. 进度显示：按"已完成分片数 / 总分片数"，并且要考虑断点续传（已传的分片直接算 100%）。

5. 暂停/继续：用 AbortController 中断当前请求，已传的分片保留（下次续传）。

6. 页面刷新/关闭：上传状态要持久化（upload_id 存 localStorage），回来继续。另外 beforeunload 提醒用户"上传还没完成"。

7. 分片大小：S3 的 multipart 要求每片 ≥ 5MB（最后一片除外），且最多 10000 片。所以：

2GB 文件、5MB 分片 → 410 片（OK）
100GB 文件、5MB 分片 → 20480 片（超限！要用更大的分片，比如 50MB → 2048 片）

前端应该根据实际大小动态计算分片大小：

function chunkSizeFor(size) {
    const MIN = 5 * 1024 * 1024;
    let cs = MIN;
    while (size / cs > 10000) cs *= 2;     // 保证不超过 10000 片
    return cs;
}

九、服务端接口设计

三个接口 + 一个清理任务：

接口	方法	入参	返回
`/upload/init/`	POST	filename, size, hash, mime	upload_id, chunk_size, 是否秒传
`/upload/presign/`	POST	upload_id, part_number, size	url
`/upload/parts/`	GET	upload_id	已上传分片列表
`/upload/complete/`	POST	upload_id, parts	file_id
`/upload/cancel/`	POST	upload_id	中止并删除碎片

清理未完成的任务（非常重要）：

@shared_task
def cleanup_abandoned_uploads():
    """清理超过 24 小时未完成的上传任务（对象存储里的碎片会一直占空间）。"""
    cutoff = timezone.now() - timedelta(hours=24)
    tasks = UploadTask.objects.filter(status='INIT', created_at__lt=cutoff)[:500]

    for t in tasks:
        try:
            if t.upload_id:
                s3.abort_multipart_upload(
                    Bucket=settings.S3_BUCKET, Key=t.object_key, UploadId=t.upload_id,
                )
            else:
                s3.delete_object(Bucket=settings.S3_BUCKET, Key=t.object_key)
        except Exception as e:
            logger.warning('abort failed %s: %s', t.upload_id, e)
        t.status = 'ABORTED'
        t.save(update_fields=['status'])

这个任务不做的话，对象存储里会堆积大量"半截文件"——我们有次发现桶里有 200GB 都是没人认领的碎片。

对象存储侧也可以配生命周期规则自动清理（比如给 tmp/ 前缀设 1 天过期），双保险。

十、实测数据与坑清单

实测（2GB 文件，家用 50Mbps 上行）

方案	耗时	Django worker 占用	中断后
服务端直接接收	6 分 10 秒	全程占用	从头再来
分片 + 服务端合并	6 分 30 秒	分片期间占用（短）+ 合并时 CPU	续传
直传对象存储（4 并发）	5 分 40 秒	几乎为 0（只签 URL）	续传
秒传（命中）	< 1 秒	忽略不计	—

直传方案最大的收益不是速度（差不多），而是 worker 占用从 6 分钟降到接近 0——这意味着上传功能再也不会拖垮站点。

坑清单

用 Django 直接接大文件 → worker 被占、超时、无断点。改直传。
分片小于 5MB → S3 multipart 报错（EntityTooSmall）。
分片数超过 10000 → 超限，要动态调大分片。
让前端决定 object key → 能覆盖别人的文件。key 必须服务端生成。
complete 只信前端传的 etag → 可伪造。向对象存储核实。
合并后不校验总大小 → 大小不符的文件被接受。
只校验扩展名/Centent-Type → 内容可以是任何东西。读文件头判断。
没有清理未完成的分片 → 对象存储堆几百 GB 垃圾。定时任务 + 生命周期规则。
预签名 URL 有效期太长 → 泄露后被滥用。1 小时以内。
秒传不检查权限 → A 用户的文件被关联给 B 用户（隐私泄露）。
秒传只用 hash 不用 size → 碰撞风险。hash + size 双匹配。
哈希计算在主线程 → 页面卡死。Web Worker。
没有失败重试 → 网络抖动导致整个上传失败。每个分片独立重试。
并发数太大 → 触发对象存储限流，或者进度条跳变。3~5 个。
前端没做暂停/继续 → 用户关了页面就全丢。upload_id 要持久化。
上传接口没做鉴权和配额 → 任何人能传任何东西塞满你的桶。必须鉴权 + 按用户限制总量。
nginx 的 client_max_body_size 没调 → 即使是方案 A 也会 413。直传方案不受影响（这也是它的好处之一）。

最后说说这次重构的整体感受。

上传这个功能，看起来是"前端的事"——选个文件、发个请求、完事。但做过大文件之后会发现，它其实是个分布式系统问题：有状态（哪些分片传了）、有失败（网络中断）、有并发（同时传多个分片）、有资源限制（磁盘、带宽、配额）、有安全（伪造、越权、垃圾文件）。

所以我的建议是：如果你的上传文件可能超过几十 MB，一开始就用直传方案。虽然前期多花一两天，但省掉的是后面无数次"上传失败"的客诉和"服务器又被上传拖垮"的救火。

还有一点：上传和下载是镜像的问题，解法也是镜像的。

下载：不要让 Django 发字节 → 用 X-Accel-Redirect 交给 nginx；
上传：不要让 Django 收字节 → 用预签名 URL 交给对象存储。

背后的原则是同一条：应用服务器负责"决策"（鉴权、校验、记录），专业组件负责"搬运"（nginx、对象存储）。想清楚这一点，很多架构问题都会变得简单。