文件分片上传合并断点续传

wen java案例 2026-07-27 140

本文目录导读：

文件分片上传合并断点续传

整体架构
前端实现
后端实现（Node.js + Express）
关键特性说明
注意事项

我来为你详细讲解文件分片上传、合并和断点续传的实现方案。

整体架构

客户端（浏览器）                   服务器
     |                              |
     |---- 1. 初始化上传请求 ------>|
     |<--- 返回 uploadId ---------|
     |                              |
     |---- 2. 分片上传(chunk) ---->|
     |---- 3. 上传进度反馈 -------->|
     |---- 4. 断点续传查询 -------->|
     |<--- 返回已上传分片列表 ----|
     |                              |
     |---- 5. 合并分片请求 -------->|
     |<--- 返回完整文件URL ------|

前端实现

1 核心上传类

class FileUploader {
  constructor(options = {}) {
    this.chunkSize = options.chunkSize || 2 * 1024 * 1024; // 默认2MB
    this.parallelLimit = options.parallelLimit || 3; // 并发数
    this.retryCount = options.retryCount || 3; // 重试次数
    this.baseUrl = options.baseUrl || '/api/upload';
    this.pendingQueue = [];
    this.activeCount = 0;
    this.fileHash = null;
    this.uploadId = null;
  }
  // 计算文件哈希
  async calculateHash(file) {
    const chunkSize = 1024 * 1024; // 1MB
    const chunks = Math.ceil(file.size / chunkSize);
    const spark = new SparkMD5.ArrayBuffer();
    for (let i = 0; i < chunks; i++) {
      const start = i * chunkSize;
      const end = Math.min(start + chunkSize, file.size);
      const blob = file.slice(start, end);
      const reader = new FileReader();
      await new Promise((resolve) => {
        reader.onload = (e) => {
          spark.append(e.target.result);
          this.updateProgress({
            type: 'hash',
            loaded: i + 1,
            total: chunks
          });
          resolve();
        };
        reader.readAsArrayBuffer(blob);
      });
    }
    return spark.end();
  }
  // 初始化上传
  async initUpload(file) {
    this.fileHash = await this.calculateHash(file);
    const response = await axios.post(`${this.baseUrl}/init`, {
      fileName: file.name,
      fileSize: file.size,
      fileHash: this.fileHash,
      chunkSize: this.chunkSize
    });
    this.uploadId = response.data.uploadId;
    return response.data;
  }
  // 切片文件
  sliceFile(file) {
    const chunks = [];
    let start = 0;
    let index = 0;
    while (start < file.size) {
      const end = Math.min(start + this.chunkSize, file.size);
      const blob = file.slice(start, end);
      chunks.push({
        blob,
        index,
        start,
        end,
        size: blob.size
      });
      start = end;
      index++;
    }
    return chunks;
  }
  // 上传单个分片
  async uploadChunk(chunk, retryCount = 0) {
    const formData = new FormData();
    formData.append('chunk', chunk.blob);
    formData.append('index', chunk.index);
    formData.append('uploadId', this.uploadId);
    formData.append('fileHash', this.fileHash);
    try {
      const response = await axios.post(`${this.baseUrl}/chunk`, formData, {
        headers: {
          'Content-Type': 'multipart/form-data'
        },
        onUploadProgress: (progressEvent) => {
          if (this.onChunkProgress) {
            this.onChunkProgress(chunk.index, progressEvent);
          }
        }
      });
      return response.data;
    } catch (error) {
      if (retryCount < this.retryCount) {
        // 指数退避重试
        await new Promise(resolve => setTimeout(resolve, 1000 * Math.pow(2, retryCount)));
        return this.uploadChunk(chunk, retryCount + 1);
      }
      throw error;
    }
  }
  // 获取已上传分片（断点续传）
  async getUploadedChunks() {
    try {
      const response = await axios.get(`${this.baseUrl}/chunks`, {
        params: {
          uploadId: this.uploadId,
          fileHash: this.fileHash
        }
      });
      return response.data.uploadedChunks || [];
    } catch (error) {
      return [];
    }
  }
  // 控制并发上传
  async uploadWithConcurrency(chunks, uploadedChunks = []) {
    const uploadPromises = [];
    const uploadedIndices = new Set(uploadedChunks);
    //过滤已上传的分片
    const remainingChunks = chunks.filter(chunk => !uploadedIndices.has(chunk.index));
    for (let i = 0; i < this.parallelLimit && i < remainingChunks.length; i++) {
      uploadPromises.push(this.processNextChunk(remainingChunks));
    }
    let completedChunks = uploadedIndices.size;
    const totalChunks = chunks.length;
    while (uploadPromises.length > 0) {
      const completedPromise = await Promise.race(uploadPromises);
      completedChunks++;
      // 更新进度
      if (this.onProgress) {
        this.onProgress({
          loaded: completedChunks,
          total: totalChunks,
          percent: (completedChunks / totalChunks) * 100
        });
      }
      // 移除已完成的任务
      const index = uploadPromises.indexOf(completedPromise);
      uploadPromises.splice(index, 1);
      // 添加新的任务
      if (remainingChunks.length > 0) {
        uploadPromises.push(this.processNextChunk(remainingChunks));
      }
    }
    return completedChunks;
  }
  // 处理下一个分片
  async processNextChunk(remainingChunks) {
    const chunk = remainingChunks.shift();
    await this.uploadChunk(chunk);
    return chunk.index;
  }
  // 合并分片
  async mergeChunks() {
    const response = await axios.post(`${this.baseUrl}/merge`, {
      uploadId: this.uploadId,
      fileHash: this.fileHash,
      fileName: this.file.name,
      totalChunks: Math.ceil(this.file.size / this.chunkSize)
    });
    return response.data;
  }
  // 主上传方法
  async upload(file) {
    this.file = file;
    try {
      // 1. 初始化上传
      const initData = await this.initUpload(file);
      console.log('Upload initialized:', initData);
      // 2. 切片
      const chunks = this.sliceFile(file);
      // 3. 检查已上传的分片（断点续传）
      const uploadedChunks = await this.getUploadedChunks();
      if (uploadedChunks.length > 0) {
        console.log(`Resuming upload, ${uploadedChunks.length} chunks already uploaded`);
      }
      // 4. 上传分片
      await this.uploadWithConcurrency(chunks, uploadedChunks);
      // 5. 合并分片
      const result = await this.mergeChunks();
      console.log('Upload completed:', result);
      return result;
    } catch (error) {
      console.error('Upload failed:', error);
      throw error;
    }
  }
}

2 使用示例

// HTML部分
/*
<div id="upload-container">
  <input type="file" id="fileInput" />
  <div id="progress-bar">
    <div id="progress-fill"></div>
  </div>
  <span id="progress-text">0%</span>
  <button id="uploadBtn">上传</button>
  <button id="pauseBtn">暂停</button>
</div>
*/
// JavaScript部分
const uploader = new FileUploader({
  chunkSize: 1024 * 1024, // 1MB
  parallelLimit: 3,
  retryCount: 3
});
// 设置进度回调
uploader.onProgress = (progress) => {
  document.getElementById('progress-fill').style.width = `${progress.percent}%`;
  document.getElementById('progress-text').textContent = `${Math.round(progress.percent)}%`;
};
// 暂停功能
let isPaused = false;
document.getElementById('pauseBtn').addEventListener('click', () => {
  isPaused = !isPaused;
  if (isPaused) {
    // 取消当前的请求
    axios.CancelToken.source().cancel('Upload paused');
  }
});
// 上传按钮
document.getElementById('uploadBtn').addEventListener('click', async () => {
  const fileInput = document.getElementById('fileInput');
  const file = fileInput.files[0];
  if (!file) {
    alert('请选择文件');
    return;
  }
  try {
    const result = await uploader.upload(file);
    alert('上传成功！');
  } catch (error) {
    if (axios.isCancel(error)) {
      console.log('Upload paused');
    } else {
      alert('上传失败：' + error.message);
    }
  }
});

后端实现（Node.js + Express）

// server.js
const express = require('express');
const multer = require('multer');
const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const app = express();
app.use(express.json());
// 配置存储
const UPLOAD_DIR = path.join(__dirname, 'uploads');
const TEMP_DIR = path.join(__dirname, 'temp');
// 确保目录存在
if (!fs.existsSync(UPLOAD_DIR)) fs.mkdirSync(UPLOAD_DIR, { recursive: true });
if (!fs.existsSync(TEMP_DIR)) fs.mkdirSync(TEMP_DIR, { recursive: true });
// 存储上传任务信息
const uploadTasks = new Map();
// 配置 multer
const storage = multer.diskStorage({
  destination: (req, file, cb) => {
    const chunkDir = path.join(TEMP_DIR, req.body.fileHash);
    if (!fs.existsSync(chunkDir)) {
      fs.mkdirSync(chunkDir, { recursive: true });
    }
    cb(null, chunkDir);
  },
  filename: (req, file, cb) => {
    cb(null, `${req.body.index}`);
  }
});
const upload = multer({ storage });
// 初始化上传
app.post('/api/upload/init', (req, res) => {
  const { fileName, fileSize, fileHash, chunkSize } = req.body;
  // 生成上传ID
  const uploadId = crypto.randomUUID();
  // 计算总分片数
  const totalChunks = Math.ceil(fileSize / chunkSize);
  // 保存任务信息
  uploadTasks.set(uploadId, {
    fileName,
    fileSize,
    fileHash,
    chunkSize,
    totalChunks,
    uploadedChunks: [],
    createdAt: Date.now()
  });
  res.json({
    uploadId,
    totalChunks,
    chunkSize
  });
});
// 上传分片
app.post('/api/upload/chunk', upload.single('chunk'), (req, res) => {
  const { uploadId, index, fileHash } = req.body;
  // 更新任务信息
  const task = uploadTasks.get(uploadId);
  if (task) {
    if (!task.uploadedChunks.includes(parseInt(index))) {
      task.uploadedChunks.push(parseInt(index));
    }
  }
  res.json({
    code: 0,
    message: 'Chunk uploaded successfully'
  });
});
// 获取已上传的分片
app.get('/api/upload/chunks', (req, res) => {
  const { uploadId, fileHash } = req.query;
  const task = uploadTasks.get(uploadId);
  let uploadedChunks = [];
  if (task) {
    uploadedChunks = task.uploadedChunks;
  } else {
    // 检查文件是否已存在（秒传）
    const filePath = path.join(UPLOAD_DIR, fileHash);
    if (fs.existsSync(filePath)) {
      uploadedChunks = 'all'; // 文件已存在
    }
  }
  res.json({
    uploadedChunks
  });
});
// 合并分片
app.post('/api/upload/merge', async (req, res) => {
  const { uploadId, fileHash, fileName, totalChunks } = req.body;
  const task = uploadTasks.get(uploadId);
  if (!task) {
    return res.status(404).json({ error: 'Upload task not found' });
  }
  const chunkDir = path.join(TEMP_DIR, fileHash);
  const filePath = path.join(UPLOAD_DIR, fileHash);
  try {
    // 创建写入流
    const writeStream = fs.createWriteStream(filePath);
    // 按顺序合并分片
    for (let i = 0; i < totalChunks; i++) {
      const chunkPath = path.join(chunkDir, `${i}`);
      // 检查分片是否存在
      if (!fs.existsSync(chunkPath)) {
        writeStream.destroy();
        fs.unlinkSync(filePath);
        throw new Error(`Chunk ${i} not found`);
      }
      // 读取并写入分片
      const chunkBuffer = fs.readFileSync(chunkPath);
      writeStream.write(chunkBuffer);
    }
    // 完成写入
    writeStream.end();
    // 等待写入完成
    await new Promise((resolve, reject) => {
      writeStream.on('finish', resolve);
      writeStream.on('error', reject);
    });
    // 清理临时文件
    fs.rmSync(chunkDir, { recursive: true, force: true });
    // 清理任务信息
    uploadTasks.delete(uploadId);
    res.json({
      code: 0,
      message: 'File merged successfully',
      data: {
        url: `/files/${fileHash}`,
        fileName: fileName
      }
    });
  } catch (error) {
    res.status(500).json({
      code: -1,
      message: error.message
    });
  }
});
// 文件访问
app.get('/files/:hash', (req, res) => {
  const filePath = path.join(UPLOAD_DIR, req.params.hash);
  if (fs.existsSync(filePath)) {
    res.sendFile(filePath);
  } else {
    res.status(404).json({ error: 'File not found' });
  }
});
// 定期清理过期任务
setInterval(() => {
  const now = Date.now();
  const maxAge = 24 * 60 * 60 * 1000; // 24小时
  for (const [uploadId, task] of uploadTasks) {
    if (now - task.createdAt > maxAge) {
      // 清理临时文件
      const chunkDir = path.join(TEMP_DIR, task.fileHash);
      if (fs.existsSync(chunkDir)) {
        fs.rmSync(chunkDir, { recursive: true, force: true });
      }
      uploadTasks.delete(uploadId);
    }
  }
}, 60 * 60 * 1000); // 每小时检查一次
app.listen(3000, () => {
  console.log('Server running on port 3000');
});