nestjs 大文件分片上传

1
2
3
4
5
6
7
8
9
const CHUNK_SIZE = 5 * 1024 * 1024 // 5MB

function createChunks(file: File) {
  const chunks: Blob[] = []
  for (let cur = 0; cur < file.size; cur += CHUNK_SIZE) {
    chunks.push(file.slice(cur, cur + CHUNK_SIZE))
  }
  return chunks
}

1
2
3
4
5
6
7
8
9
10
11
12
// hash.worker.ts
import SparkMD5 from 'spark-md5'

self.onmessage = async ({ data: chunks }) => {
  const spark = new SparkMD5.ArrayBuffer()
  for (let i = 0; i < chunks.length; i++) {
    const buf = await chunks[i].arrayBuffer()
    spark.append(buf)
    self.postMessage({ progress: ((i + 1) / chunks.length) * 100 })
  }
  self.postMessage({ hash: spark.end() })
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
async function upload(file: File) {
  const chunks = createChunks(file)
  const hash = await calcHash(chunks)

  // 1. 秒传 / 断点续传查询
  const { data } = await axios.post('/verify', {
    hash,
    name: file.name,
    total: chunks.length,
  })
  if (data.exists) return data.url

  // 2. 跳过已上传的分片
  const uploaded = new Set<number>(data.uploaded)
  const tasks = chunks
    .map((chunk, index) => ({ chunk, index }))
    .filter(({ index }) => !uploaded.has(index))
    .map(({ chunk, index }) => () => {
      const fd = new FormData()
      fd.append('chunk', chunk)
      fd.append('hash', hash)
      fd.append('index', String(index))
      return axios.post('/upload', fd)
    })

  await runWithLimit(tasks, 4) // 并发 4

  // 3. 通知合并
  const { data: merged } = await axios.post('/merge', {
    hash,
    name: file.name,
    total: chunks.length,
  })
  return merged.url
}

// 简单的并发限制
async function runWithLimit<T>(tasks: (() => Promise<T>)[], limit: number) {
  const ret: Promise<T>[] = []
  const executing = new Set<Promise<T>>()
  for (const task of tasks) {
    const p = task().finally(() => executing.delete(p))
    ret.push(p)
    executing.add(p)
    if (executing.size >= limit) await Promise.race(executing)
  }
  return Promise.all(ret)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
// upload.controller.ts
import {
  Body,
  Controller,
  Post,
  UploadedFile,
  UseInterceptors,
} from '@nestjs/common'
import { FileInterceptor } from '@nestjs/platform-express'
import { diskStorage } from 'multer'
import * as fs from 'fs-extra'
import * as path from 'path'

const TMP_DIR = path.resolve(process.cwd(), 'uploads/tmp')
const TARGET_DIR = path.resolve(process.cwd(), 'uploads/files')

@Controller()
export class UploadController {
  // 1. 秒传 / 断点续传查询
  @Post('verify')
  async verify(@Body() body: { hash: string; name: string }) {
    const ext = path.extname(body.name)
    const target = path.join(TARGET_DIR, body.hash + ext)
    if (await fs.pathExists(target)) {
      return { exists: true, url: `/files/${body.hash}${ext}` }
    }
    const chunkDir = path.join(TMP_DIR, body.hash)
    const uploaded = (await fs.pathExists(chunkDir))
      ? (await fs.readdir(chunkDir)).map(Number).sort((a, b) => a - b)
      : []
    return { exists: false, uploaded }
  }

  // 2. 接收分片
  @Post('upload')
  @UseInterceptors(
    FileInterceptor('chunk', {
      storage: diskStorage({
        destination: async (req, _file, cb) => {
          const dir = path.join(TMP_DIR, req.body.hash)
          await fs.ensureDir(dir)
          cb(null, dir)
        },
        filename: (req, _file, cb) => cb(null, req.body.index),
      }),
    }),
  )
  uploadChunk(@UploadedFile() _chunk: Express.Multer.File) {
    return { ok: true }
  }

  // 3. 合并
  @Post('merge')
  async merge(
    @Body() body: { hash: string; name: string; total: number },
  ) {
    const { hash, name, total } = body
    const ext = path.extname(name)
    const chunkDir = path.join(TMP_DIR, hash)
    const target = path.join(TARGET_DIR, hash + ext)
    await fs.ensureDir(TARGET_DIR)

    // 用流按序拼接，避免一次性读进内存
    const writeStream = fs.createWriteStream(target)
    for (let i = 0; i < total; i++) {
      const chunkPath = path.join(chunkDir, String(i))
      await new Promise<void>((resolve, reject) => {
        fs.createReadStream(chunkPath)
          .on('end', resolve)
          .on('error', reject)
          .pipe(writeStream, { end: false })
      })
    }
    writeStream.end()

    // 清理临时分片
    await fs.remove(chunkDir)

    return { url: `/files/${hash}${ext}` }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// app.module.ts
import { Module } from '@nestjs/common'
import { ServeStaticModule } from '@nestjs/serve-static'
import { UploadController } from './upload.controller'
import * as path from 'path'

@Module({
  imports: [
    ServeStaticModule.forRoot({
      rootPath: path.resolve(process.cwd(), 'uploads/files'),
      serveRoot: '/files',
    }),
  ],
  controllers: [UploadController],
})
export class AppModule {}

1
2
3
4
5
6
7
8
9
10
11
12
// main.ts
import { NestFactory } from '@nestjs/core'
import { json, urlencoded } from 'express'
import { AppModule } from './app.module'

async function bootstrap() {
  const app = await NestFactory.create(AppModule)
  app.use(json({ limit: '50mb' }))
  app.use(urlencoded({ extended: true, limit: '50mb' }))
  await app.listen(3000)
}
bootstrap()

1
2
3
4
5
6
7
8
9
10
11
12
13
import { Cron } from '@nestjs/schedule'

@Cron('0 3 * * *') // 每天 3 点
async cleanup() {
  const dirs = await fs.readdir(TMP_DIR)
  const now = Date.now()
  for (const d of dirs) {
    const stat = await fs.stat(path.join(TMP_DIR, d))
    if (now - stat.mtimeMs > 7 * 24 * 3600 * 1000) {
      await fs.remove(path.join(TMP_DIR, d))
    }
  }
}