IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

大文件分片上传怎么做？

wen python案例 2

本文目录导读：

大文件分片上传怎么做？

  1. 基本原理
  2. 核心流程（分步详解）
  3. 关键技术点与难点
  4. 代码逻辑示例（伪代码+核心思路）
  5. 总结与最佳实践

大文件分片上传是解决大文件（如视频、高清图片、数据集等）在网络上传输不稳定、易超时、失败后需重传整个文件等问题的标准方案。

其核心思想是：化整为零，各个击破，最后聚零为整。

下面我将从基本原理、核心流程、关键技术点以及示例代码逻辑四个方面为你详细拆解。

基本原理

将一个大文件在客户端（浏览器/App）按固定大小（5MB、10MB）切割成多个小块。

  • 每个小块单独发起一个 HTTP 请求上传到服务器。
  • 服务器接收并临时保存这些小片段。
  • 所有片段上传完成后,客户端通知服务器进行合并。
  • 服务器将所有片段按序号拼接成完整的原始文件。

核心流程（分步详解）

整个过程可以简化为三个核心阶段：

sequenceDiagram
    participant Client as 客户端
    participant Server as 服务器
    Client->>Server: 1. 初始化上传（文件名、文件大小、分片数）
    Server-->>Client: 返回 UploadId（上传唯一标识）
    Client->>Client: 2. 计算分片，遍历上传
    loop 分片上传
        Client->>Server: 3. POST /upload/part (UploadId, partNumber, fileChunk)
        Server-->>Client: 返回 ETag (该分片的MD5或标识)
    end
    Client->>Server: 4. 请求合并 (UploadId, 所有分片的ETag列表)
    Server->>Server: 5. 校验、合并、生成最终文件
    Server-->>Client: 6. 返回最终文件URL/结果

初始化（Initiate）

  • 动作：客户端发起一个预请求，告知服务器：我要上传一个名为 example.mp4、大小为 500MB 的文件。
  • 服务器响应：创建一个与该文件关联的上传事务，并返回一个 UploadId（唯一标识），服务器可以返回建议的分片大小（如 5MB）。
  • 作用：建立上下文，后续所有分片都关联到这个 UploadId。

上传分片（Upload Part）

  • 动作：客户端将文件按固定大小（最后一片可能较小）切割成 N 份。
  • 循环上传：每个分片执行一次 POST 请求，请求包含：
    • uploadId：刚才拿到的标识。
    • partNumber：分片序号（1, 2, 3...N），这是保证顺序的关键。
    • file：该分片的二进制数据。
  • 并发控制：通常会使用并发池（如限制同时3-5个请求），既能加快速度，又不会打爆服务器或用户带宽。
  • 安全与校验：每个分片上传成功后，服务器会计算其 MD5，返回一个 ETag，客户端需要保存{partNumber: ETag}的映射表。

完成合并（Complete/Finish）

  • 动作：所有分片上传成功后，客户端发起一个 POST 请求，携带：
    • uploadId
    • 所有分片的 partNumber 和 ETag 列表（按序号升序）。
  • 服务器动作：
    1. 根据提供的 ETag 列表，校验每个分片的完整性。
    2. 将所有临时分片文件,按 partNumber 顺序拼接成一个完整文件。
    3. 删除临时分片文件。
    4. 返回最终文件的访问地址。

关键技术点与难点

断点续传与秒传

这是分片上传最重要的附加价值。

  • 断点续传：上传中途断了（比如用户关闭网页、网络掉线），再次上传时，客户端可以先问服务器：这个文件是否已上传过？已上传了哪些分片？（即查询 UploadId 的状态），然后只上传缺失的分片。
  • 秒传：上传前，客户端计算文件的唯一标识（如 MD5 或 SHA-1，或者更高效的 xxHash），发送给服务器查询，如果服务器发现该 MD5 已存在，直接返回上传成功，这可以秒传同一个文件。

并发控制

  • 不能一次性把几百个分片同时发送（会造成浏览器崩溃、连接池耗尽、服务器压力过大）。
  • 常见做法：使用一个队列，限制最大并发数（如 3~6 个），每当一个分片上传完成，就从队列中取出下一个分片开始上传。

失败重试

  • 自动重试：单个分片上传失败（网络抖动、服务器5xx错误），客户端需要自动重试该分片（通常重试3次）。
  • 指数退避：如果不成功，重试间隔逐渐增加（如 1秒、2秒、4秒）。

服务器端临时存储

  • 服务器不能把几百个分片一直放在内存里。
  • 策略：每个分片写入磁盘上的临时文件（通常放在 /tmp/upload/{uploadId}/part_1 目录下），合并完成后删除目录。

合并效率

  • 避免内存溢出：合并时，不要一次性读取所有分片到内存再写入，应该使用流式读写，逐个分片读取并追加写入到目标文件。
  • 操作系统层面：可以直接使用 cat 命令（Linux）或 FileStream（Node.js / Java）进行高效拼接。

代码逻辑示例（伪代码+核心思路）

这里以 前端（浏览器 JavaScript） 和 后端（Node.js + Express） 为例，展示核心逻辑。

前端逻辑 (JavaScript)

// 1. 文件选择与初始化
const file = document.getElementById('fileInput').files[0];
const CHUNK_SIZE = 5 * 1024 * 1024; // 5MB
// 计算分片总数
const totalChunks = Math.ceil(file.size / CHUNK_SIZE);
// 2. 发起初始化请求，获取 UploadId
const initResponse = await fetch('/upload/init', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ fileName: file.name, fileSize: file.size, totalChunks: totalChunks })
});
const { uploadId } = await initResponse.json();
// 3. 并发上传分片
const uploadedETags = [];
const MAX_CONCURRENCY = 3; // 最大并发数
async function uploadSingleChunk(chunkIndex) {
    const start = chunkIndex * CHUNK_SIZE;
    const end = Math.min(start + CHUNK_SIZE, file.size);
    const chunkBlob = file.slice(start, end); // 切割文件
    const formData = new FormData();
    formData.append('uploadId', uploadId);
    formData.append('partNumber', chunkIndex + 1);
    formData.append('file', chunkBlob);
    const res = await fetch('/upload/part', { method: 'POST', body: formData });
    const { etag } = await res.json();
    uploadedETags.push({ partNumber: chunkIndex + 1, etag: etag });
    // 更新进度 (可以在这里往界面发进度事件)
    console.log(`Chunk ${chunkIndex + 1} uploaded`);
}
// 限流并发执行 (简单的队列实现)
async function uploadAllChunks() {
    const chunkQueue = Array.from({ length: totalChunks }, (_, i) => i);
    const workers = [];
    async function worker() {
        while (chunkQueue.length > 0) {
            const chunkIndex = chunkQueue.shift();
            try {
                await uploadSingleChunk(chunkIndex);
            } catch (e) {
                // 这里可以加入重试逻辑
                console.error(`Chunk ${chunkIndex} failed:`, e);
                // 如需要重试，可重新推入队列并 break 当前 Worker
            }
        }
    }
    // 启动3个 worker
    for (let i = 0; i < MAX_CONCURRENCY; i++) {
        workers.push(worker());
    }
    await Promise.all(workers);
}
await uploadAllChunks();
// 4. 请求合并
const completeResponse = await fetch('/upload/complete', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ uploadId, parts: uploadedETags.sort((a, b) => a.partNumber - b.partNumber) })
});
const { fileUrl } = await completeResponse.json();
console.log('Upload completed:', fileUrl);

后端逻辑 (Node.js + Express + Multer)

const express = require('express');
const multer = require('multer');
const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const app = express();
const upload = multer({ dest: '/tmp/uploads/' }); // 临时存储分片
// 内存存储 uploadId 信息 (生产环境用 Redis/数据库)
const uploadSessions = {};
// 1. 初始化
app.post('/upload/init', (req, res) => {
    const { fileName, fileSize, totalChunks } = req.body;
    const uploadId = crypto.randomUUID();
    const dir = `/tmp/uploads/${uploadId}`;
    fs.mkdirSync(dir, { recursive: true });
    uploadSessions[uploadId] = { fileName, fileSize, totalChunks, dir };
    res.json({ uploadId });
});
// 2. 上传分片
app.post('/upload/part', upload.single('file'), (req, res) => {
    const { uploadId, partNumber } = req.body;
    const session = uploadSessions[uploadId];
    if (!session) return res.status(400).send('Invalid uploadId');
    const tempFile = req.file; // multer 保存的文件
    const ext = path.extname(tempFile.originalname);
    const targetPath = path.join(session.dir, `part_${partNumber}${ext}`);
    // 将临时文件移动到指定位置 (或直接改名)
    fs.renameSync(tempFile.path, targetPath);
    // 计算 MD5 (ETag)
    const fileBuffer = fs.readFileSync(targetPath);
    const etag = crypto.createHash('md5').update(fileBuffer).digest('hex');
    res.json({ etag });
});
// 3. 完成合并
app.post('/upload/complete', (req, res) => {
    const { uploadId, parts } = req.body;
    const session = uploadSessions[uploadId];
    if (!session) return res.status(400).send('Invalid uploadId');
    // 按序号排序
    parts.sort((a, b) => a.partNumber - b.partNumber);
    // 合并文件 (使用流式写入)
    const finalPath = path.join(__dirname, 'uploads', session.fileName);
    const writeStream = fs.createWriteStream(finalPath);
    for (const part of parts) {
        const partPath = path.join(session.dir, `part_${part.partNumber}`);
        const partData = fs.readFileSync(partPath);
        writeStream.write(partData);
        // 验证 ETag (可选)
    }
    writeStream.end(() => {
        // 清理临时目录
        fs.rmSync(session.dir, { recursive: true, force: true });
        delete uploadSessions[uploadId];
        res.json({ success: true, fileUrl: `/downloads/${session.fileName}` });
    });
});
app.listen(3000);

总结与最佳实践

  1. 前端重点：

    • 使用 File.slice() 进行分片。
    • 实现并发控制（3-6个并发为宜）。
    • 实现失败重试与断点续传逻辑（保存已上传的分片列表）。
    • 计算文件Hash（MD5/SHA1）用于秒传校验。

  2. 后端重点：

    • 生成全局唯一的 UploadId。
    • 安全地存储和拼接分片（注意磁盘空间和文件路径安全性）。
    • 实现分片过期清理机制（上传中断后，临时文件不会永远占用磁盘）。
    • 合并操作使用流，防止内存溢出。

  3. 业界方案：

    • 阿里云OSS / 腾讯云COS / AWS S3 都提供了成熟的分片上传 API（InitiateMultipartUpload, UploadPart, CompleteMultipartUpload），非常稳定，直接调用SDK即可。
    • 如果是小团队自建,可以参考上述逻辑实现一个轻量版。

上一篇断点续传怎么实现？

下一篇WebSocket推送怎么做？

相关文章

抱歉，评论功能暂时关闭!