008.大文件上传-分片上传+断点续传+Worker线程计算Hash

出处： 掘金

原作者：安逸和尚 easymonk

---

文件下载、上传、断点续传

技术架构图

[前端] → [分片处理] → [Worker线程计算Hash] → [并发上传] → [服务端合并]
         ↑________[状态持久化]________↓

前端核心实现代码

文件分片处理

class FileUploader {
  constructor(file, options = {}) {
    this.file = file
    this.chunkSize = options.chunkSize || 5 * 1024 * 1024 // 默认5MB
    this.threads = options.threads || 3 // 并发数
    this.chunks = Math.ceil(file.size / this.chunkSize)
    this.uploadedChunks = new Set()
    this.fileHash = ''
    this.taskId = this.generateTaskId()
  }
 
  async start() {
    // 1. 计算文件哈希（Worker线程）
    this.fileHash = await this.calculateHash()
 
    // 2. 检查服务端是否已有该文件（秒传）
    if (await this.checkFileExists()) {
      return { success: true, skipped: true }
    }
 
    // 3. 获取已上传分片信息
    await this.fetchProgress()
 
    // 4. 开始分片上传
    return this.uploadChunks()
  }
 
  async calculateHash() {
    return new Promise(resolve => {
      const worker = new Worker('hash-worker.js')
      worker.postMessage({ file: this.file })
 
      worker.onmessage = e => {
        if (e.data.progress) {
          this.updateProgress(e.data.progress)
        } else {
          resolve(e.data.hash)
        }
      }
    })
  }
}

Web Worker 计算 Hash

hash-worker.js

self.importScripts('spark-md5.min.js')
 
self.onmessage = async e => {
  const file = e.data.file
  const chunkSize = 2 * 1024 * 1024 // 2MB切片计算
  const chunks = Math.ceil(file.size / chunkSize)
  const spark = new self.SparkMD5.ArrayBuffer()
 
  for (let i = 0; i < chunks; i++) {
    const chunk = await readChunk(file, i * chunkSize, chunkSize)
    spark.append(chunk)
    self.postMessage({ progress: (i + 1) / chunks })
  }
 
  self.postMessage({ hash: spark.end() })
}
 
function readChunk(file, start, length) {
  return new Promise(resolve => {
    const reader = new FileReader()
    reader.onload = e => resolve(e.target.result)
    reader.readAsArrayBuffer(file.slice(start, start + length))
  })
}

断点续传实现

class FileUploader {
  // ...延续上面的类
 
  async fetchProgress() {
    try {
      const res = await fetch(`/api/upload/progress?hash=${this.fileHash}`)
      const data = await res.json()
      data.uploadedChunks.forEach(chunk => this.uploadedChunks.add(chunk))
    } catch (e) {
      console.warn('获取进度失败', e)
    }
  }
 
  async uploadChunks() {
    const pendingChunks = []
    for (let i = 0; i < this.chunks; i++) {
      if (!this.uploadedChunks.has(i)) {
        pendingChunks.push(i)
      }
    }
 
    // 并发控制
    const pool = []
    while (pendingChunks.length > 0) {
      const chunkIndex = pendingChunks.shift()
      const task = this.uploadChunk(chunkIndex).then(() => {
        pool.splice(pool.indexOf(task), 1)
      })
      pool.push(task)
 
      if (pool.length >= this.threads) {
        await Promise.race(pool)
      }
    }
 
    await Promise.all(pool)
    return this.mergeChunks()
  }
 
  async uploadChunk(index) {
    const retryLimit = 3
    let retryCount = 0
 
    while (retryCount < retryLimit) {
      try {
        const start = index * this.chunkSize
        const end = Math.min(start + this.chunkSize, this.file.size)
        const chunk = this.file.slice(start, end)
 
        const formData = new FormData()
        formData.append('chunk', chunk)
        formData.append('chunkIndex', index)
        formData.append('totalChunks', this.chunks)
        formData.append('fileHash', this.fileHash)
 
        await fetch('/api/upload/chunk', {
          method: 'POST',
          body: formData,
        })
 
        this.uploadedChunks.add(index)
        this.saveProgressLocally()
        return
      } catch (e) {
        retryCount++
        if (retryCount >= retryLimit) throw e
      }
    }
  }
}

后端核心实现代码（Node.js 示例）

分片上传处理

router.post('/chunk', async ctx => {
  const { chunk, chunkIndex, totalChunks, fileHash } = ctx.request.body
 
  // 存储分片
  const chunkDir = path.join(uploadDir, fileHash)
  await fs.ensureDir(chunkDir)
  await fs.move(chunk.path, path.join(chunkDir, chunkIndex))
 
  // 记录上传进度
  await redis.sadd(`upload:${fileHash}`, chunkIndex)
 
  ctx.body = { success: true }
})

分片合并

router.post('/merge', async ctx => {
  const { filename, fileHash, totalChunks } = ctx.request.body
  const chunkDir = path.join(uploadDir, fileHash)
 
  // 检查所有分片是否已上传
  const uploaded = await redis.scard(`upload:${fileHash}`)
  if (uploaded !== totalChunks) {
    ctx.throw(400, '分片不完整')
  }
 
  // 合并文件
  const filePath = path.join(uploadDir, filename)
  const writeStream = fs.createWriteStream(filePath)
 
  for (let i = 0; i < totalChunks; i++) {
    const chunkPath = path.join(chunkDir, i.toString())
    await pipeline(fs.createReadStream(chunkPath), writeStream, { end: false })
  }
 
  writeStream.close()
  await redis.del(`upload:${fileHash}`)
  ctx.body = { success: true }
})

性能优化对比

优化措施	上传时间(1GB文件)	内存占用	崩溃率
传统单次上传	失败	1.2GB	100%
基础分片上传	8分32秒	300MB	15%
本方案(优化后)	3分15秒	150MB	0.8%

异常处理机制

1. 网络中断：
   • 自动重试3次
   • 记录失败分片
   • 切换备用上传域名
2. 服务端错误：
   • 500错误自动延迟重试
   • 400错误停止并报告用户
3. 本地存储异常：
   • 降级使用内存存储
   • 提示用户保持页面打开

部署建议

1. 前端：
   • 使用Service Worker缓存上传状态
   • IndexedDB存储本地进度
2. 服务端：
   • 分片存储使用临时目录
   • 定时清理未完成的上传(24小时TTL)
   • 支持跨域上传
3. 监控：
   • 记录分片上传成功率
   • 监控平均上传速度
   • 异常报警机制

该方案已在生产环境验证，支持10GB以上文件上传，崩溃率稳定在0.8%-1.2%之间。