视频上传与处理流水线

用户上传一个视频文件到它能被在线播放，中间要经过多个处理环节：分片上传到对象存储、提取元数据（时长/分辨率）、转码为多码率 HLS 格式、截取封面图、更新数据库状态。这是一条完整的异步处理流水线，每个环节都有可能失败，需要可靠的状态管理和错误恢复。本章完整实现这条流水线。

1. 上传架构

1.1 直传 vs 代理

方案	流程	优点	缺点
Presigned URL 直传	前端 → S3/R2	不占服务端带宽	需要 CORS 配置
服务端代理	前端 → 服务端 → S3	简单	服务端成瓶颈

视频文件通常很大（数百 MB），直传是唯一可行方案——让用户的浏览器直接上传到对象存储，服务端只负责签发上传凭证。

前端请求 Presigned URL → 服务端签发 → 前端直传 S3/R2 → 上传完成通知服务端

1.2 S3 兼容对象存储

Cloudflare R2、AWS S3、阿里云 OSS 都兼容 S3 API。使用 @aws-sdk/client-s3 统一操作：

// server/utils/storage.ts
import { S3Client, PutObjectCommand, GetObjectCommand } from '@aws-sdk/client-s3'
import { getSignedUrl } from '@aws-sdk/s3-request-presigner'
 
const s3 = new S3Client({
  region: 'auto',
  endpoint: useRuntimeConfig().s3Endpoint,
  credentials: {
    accessKeyId: useRuntimeConfig().s3AccessKey,
    secretAccessKey: useRuntimeConfig().s3SecretKey,
  },
})
 
export async function createPresignedUploadUrl(
  key: string,
  contentType: string
): Promise<string> {
  const command = new PutObjectCommand({
    Bucket: useRuntimeConfig().s3Bucket,
    Key: key,
    ContentType: contentType,
  })
  return getSignedUrl(s3, command, { expiresIn: 3600 })
}

2. 分片上传

2.1 为什么需要分片

大文件上传的问题：

单次上传超时：500MB 文件在慢网络下可能需要几分钟
失败重传代价大：上传到 90% 断网，需要从头开始
浏览器内存：一次性读取大文件会占用大量内存

分片上传：把文件切成 5~10MB 的小块，逐片上传。失败只需重传失败的分片。

2.2 S3 Multipart Upload 流程

1. InitiateMultipartUpload → 获取 uploadId
2. 对每个分片：UploadPart → 获取 ETag
3. CompleteMultipartUpload → 合并所有分片

2.3 服务端接口

// server/api/upload/initiate.post.ts
import { CreateMultipartUploadCommand } from '@aws-sdk/client-s3'
 
const schema = z.object({
  filename: z.string(),
  contentType: z.string(),
  fileSize: z.number(),
})
 
export default defineEventHandler(async (event) => {
  const session = await requireAuth(event)
  const body = await readValidatedBody(event, schema.parse)
 
  // 校验文件类型和大小
  if (!body.contentType.startsWith('video/')) {
    throw createError({ statusCode: 400, message: '仅支持视频文件' })
  }
  if (body.fileSize > 2 * 1024 * 1024 * 1024) { // 2GB
    throw createError({ statusCode: 400, message: '文件不能超过 2GB' })
  }
 
  const key = `uploads/${session.user.id}/${Date.now()}-${body.filename}`
  const command = new CreateMultipartUploadCommand({
    Bucket: useRuntimeConfig().s3Bucket,
    Key: key,
    ContentType: body.contentType,
  })
 
  const result = await s3.send(command)
 
  return {
    uploadId: result.UploadId,
    key,
  }
})
 
// server/api/upload/presign-part.post.ts
import { UploadPartCommand } from '@aws-sdk/client-s3'
 
const schema = z.object({
  key: z.string(),
  uploadId: z.string(),
  partNumber: z.number().min(1).max(10000),
})
 
export default defineEventHandler(async (event) => {
  await requireAuth(event)
  const body = await readValidatedBody(event, schema.parse)
 
  const command = new UploadPartCommand({
    Bucket: useRuntimeConfig().s3Bucket,
    Key: body.key,
    UploadId: body.uploadId,
    PartNumber: body.partNumber,
  })
 
  const url = await getSignedUrl(s3, command, { expiresIn: 3600 })
  return { url }
})
 
// server/api/upload/complete.post.ts
import { CompleteMultipartUploadCommand } from '@aws-sdk/client-s3'
 
const schema = z.object({
  key: z.string(),
  uploadId: z.string(),
  parts: z.array(z.object({
    partNumber: z.number(),
    etag: z.string(),
  })),
})
 
export default defineEventHandler(async (event) => {
  const session = await requireAuth(event)
  const body = await readValidatedBody(event, schema.parse)
 
  await s3.send(new CompleteMultipartUploadCommand({
    Bucket: useRuntimeConfig().s3Bucket,
    Key: body.key,
    UploadId: body.uploadId,
    MultipartUpload: {
      Parts: body.parts.map((p) => ({
        PartNumber: p.partNumber,
        ETag: p.etag,
      })),
    },
  }))
 
  // 创建视频记录
  const db = useDB()
  const [video] = await db.insert(videos).values({
    userId: session.user.id,
    sourceType: 'upload',
    sourceUrl: `${useRuntimeConfig().s3PublicUrl}/${body.key}`,
    status: 'processing',
  }).returning()
 
  // 触发处理流水线
  await triggerProcessingPipeline(video.id)
 
  return { videoId: video.id }
})

2.4 前端分片上传 Composable

// composables/useChunkUpload.ts
const CHUNK_SIZE = 10 * 1024 * 1024 // 10MB
 
export function useChunkUpload() {
  const progress = ref(0)
  const status = ref<'idle' | 'uploading' | 'completed' | 'failed'>('idle')
  const error = ref<string | null>(null)
 
  async function upload(file: File): Promise<string | null> {
    status.value = 'uploading'
    progress.value = 0
    error.value = null
 
    try {
      // 1. 初始化分片上传
      const { uploadId, key } = await $fetch('/api/upload/initiate', {
        method: 'POST',
        body: {
          filename: file.name,
          contentType: file.type,
          fileSize: file.size,
        },
      })
 
      // 2. 计算分片
      const totalChunks = Math.ceil(file.size / CHUNK_SIZE)
      const parts: { partNumber: number; etag: string }[] = []
 
      // 3. 逐片上传
      for (let i = 0; i < totalChunks; i++) {
        const start = i * CHUNK_SIZE
        const end = Math.min(start + CHUNK_SIZE, file.size)
        const chunk = file.slice(start, end)
 
        // 获取签名 URL
        const { url } = await $fetch('/api/upload/presign-part', {
          method: 'POST',
          body: { key, uploadId, partNumber: i + 1 },
        })
 
        // 直传到 S3
        const response = await fetch(url, {
          method: 'PUT',
          body: chunk,
        })
 
        const etag = response.headers.get('ETag')!
        parts.push({ partNumber: i + 1, etag })
 
        progress.value = Math.round(((i + 1) / totalChunks) * 100)
      }
 
      // 4. 完成上传
      const { videoId } = await $fetch('/api/upload/complete', {
        method: 'POST',
        body: { key, uploadId, parts },
      })
 
      status.value = 'completed'
      return videoId
    } catch (e: any) {
      status.value = 'failed'
      error.value = e.message || '上传失败'
      return null
    }
  }
 
  return { upload, progress, status, error }
}

3. 视频处理流水线

3.1 处理流程

上传完成后，视频需要经过一系列处理：

原始文件 → 元数据提取 → 转码（多码率 HLS）→ 封面截取 → 状态更新

每个步骤是独立的、可失败的。我们用状态机来管理：

// packages/shared/types/video-pipeline.ts
type PipelineStage =
  | 'uploaded'       // 已上传
  | 'extracting'     // 提取元数据中
  | 'transcoding'    // 转码中
  | 'thumbnailing'   // 截取封面中
  | 'ready'          // 处理完成
  | 'failed'         // 处理失败

3.2 元数据提取

使用 FFprobe（FFmpeg 的探测工具）提取视频信息：

// server/utils/video-probe.ts
import { execFile } from 'node:child_process'
import { promisify } from 'node:util'
 
const execFileAsync = promisify(execFile)
 
export interface VideoMetadata {
  duration: number
  width: number
  height: number
  codec: string
  bitrate: number
  fps: number
}
 
export async function probeVideo(filePath: string): Promise<VideoMetadata> {
  const { stdout } = await execFileAsync('ffprobe', [
    '-v', 'quiet',
    '-print_format', 'json',
    '-show_format',
    '-show_streams',
    filePath,
  ])
 
  const info = JSON.parse(stdout)
  const videoStream = info.streams.find((s: any) => s.codec_type === 'video')
 
  return {
    duration: Math.floor(Number(info.format.duration)),
    width: videoStream.width,
    height: videoStream.height,
    codec: videoStream.codec_name,
    bitrate: Math.floor(Number(info.format.bit_rate)),
    fps: eval(videoStream.r_frame_rate), // "30/1" → 30
  }
}

3.3 转码为 HLS

// server/utils/video-transcode.ts
export async function transcodeToHls(
  inputPath: string,
  outputDir: string,
  metadata: VideoMetadata
): Promise<string> {
  // 根据原始分辨率决定输出码率
  const profiles = getTranscodeProfiles(metadata.height)
 
  const args = ['-i', inputPath]
 
  // 为每个码率生成一个流
  for (let i = 0; i < profiles.length; i++) {
    const p = profiles[i]
    args.push(
      `-map`, `0:v`, `-map`, `0:a`,
      `-c:v:${i}`, 'libx264',
      `-b:v:${i}`, p.videoBitrate,
      `-s:${i}`, `${p.width}x${p.height}`,
      `-c:a:${i}`, 'aac',
      `-b:a:${i}`, p.audioBitrate,
    )
  }
 
  args.push(
    '-f', 'hls',
    '-hls_time', '6',
    '-hls_list_size', '0',
    '-hls_segment_filename', `${outputDir}/%v/segment-%03d.ts`,
    '-master_pl_name', 'master.m3u8',
    '-var_stream_map', profiles.map((_, i) => `v:${i},a:${i}`).join(' '),
    `${outputDir}/%v/playlist.m3u8`,
  )
 
  await execFileAsync('ffmpeg', args)
  return `${outputDir}/master.m3u8`
}
 
function getTranscodeProfiles(sourceHeight: number) {
  const profiles = [
    { height: 360, width: 640, videoBitrate: '800k', audioBitrate: '96k' },
    { height: 480, width: 854, videoBitrate: '1400k', audioBitrate: '128k' },
    { height: 720, width: 1280, videoBitrate: '2800k', audioBitrate: '128k' },
    { height: 1080, width: 1920, videoBitrate: '5000k', audioBitrate: '192k' },
  ]
 
  // 不输出超过原始分辨率的码率
  return profiles.filter((p) => p.height <= sourceHeight)
}

3.4 封面截取

// server/utils/video-thumbnail.ts
export async function extractThumbnail(
  inputPath: string,
  outputPath: string,
  timeOffset: number = 1
): Promise<void> {
  await execFileAsync('ffmpeg', [
    '-i', inputPath,
    '-ss', String(timeOffset),
    '-vframes', '1',
    '-vf', 'scale=640:-1',
    '-q:v', '2',
    outputPath,
  ])
}

从视频的第 1 秒截取一帧作为封面。用户也可以自定义封面时间点或上传自定义封面。

4. 流水线编排

4.1 Pipeline Runner

// server/utils/video-pipeline.ts
export async function runProcessingPipeline(videoId: string) {
  const db = useDB()
 
  try {
    const [video] = await db.select().from(videos).where(eq(videos.id, videoId))
    if (!video) throw new Error(`Video ${videoId} not found`)
 
    // 1. 下载原始文件到临时目录
    const tempDir = `/tmp/video-${videoId}`
    await mkdir(tempDir, { recursive: true })
    const inputPath = `${tempDir}/source${extname(video.sourceUrl!)}`
    await downloadFile(video.sourceUrl!, inputPath)
 
    // 2. 提取元数据
    await updateStatus(videoId, 'extracting')
    const metadata = await probeVideo(inputPath)
    await db.update(videos).set({
      duration: metadata.duration,
      width: metadata.width,
      height: metadata.height,
    }).where(eq(videos.id, videoId))
 
    // 3. 转码 HLS
    await updateStatus(videoId, 'transcoding')
    const hlsDir = `${tempDir}/hls`
    await mkdir(hlsDir, { recursive: true })
    await transcodeToHls(inputPath, hlsDir, metadata)
 
    // 4. 上传 HLS 到对象存储
    const hlsKey = `videos/${videoId}/hls`
    await uploadDirectory(hlsDir, hlsKey)
    const hlsUrl = `${useRuntimeConfig().s3PublicUrl}/${hlsKey}/master.m3u8`
 
    // 5. 截取封面
    await updateStatus(videoId, 'thumbnailing')
    const thumbPath = `${tempDir}/thumbnail.jpg`
    await extractThumbnail(inputPath, thumbPath)
    const thumbKey = `videos/${videoId}/thumbnail.jpg`
    await uploadFile(thumbPath, thumbKey)
    const thumbnailUrl = `${useRuntimeConfig().s3PublicUrl}/${thumbKey}`
 
    // 6. 更新数据库
    await db.update(videos).set({
      hlsUrl,
      thumbnailUrl,
      status: 'ready',
    }).where(eq(videos.id, videoId))
 
    // 7. 清理临时文件
    await rm(tempDir, { recursive: true, force: true })
 
  } catch (error: any) {
    await db.update(videos).set({
      status: 'failed',
    }).where(eq(videos.id, videoId))
    throw error
  }
}

4.2 异步触发

处理流水线耗时较长（分钟级），不能在 API 请求中同步执行。简单方案：

// server/utils/trigger.ts
export async function triggerProcessingPipeline(videoId: string) {
  // 简单方案：在后台异步执行，不阻塞请求
  runProcessingPipeline(videoId).catch((error) => {
    console.error(`Pipeline failed for video ${videoId}:`, error)
  })
}

生产环境更好的方案是使用消息队列（BullMQ + Redis）：

// 生产方案：消息队列
import { Queue, Worker } from 'bullmq'
 
const videoQueue = new Queue('video-processing', { connection: redis })
 
export async function triggerProcessingPipeline(videoId: string) {
  await videoQueue.add('process', { videoId }, {
    attempts: 3,
    backoff: { type: 'exponential', delay: 5000 },
  })
}
 
// Worker
new Worker('video-processing', async (job) => {
  await runProcessingPipeline(job.data.videoId)
}, { connection: redis, concurrency: 2 })

5. 上传界面

<!-- components/VideoUploader.vue -->
<script setup>
const { upload, progress, status, error } = useChunkUpload()
const fileInput = useTemplateRef<HTMLInputElement>('fileInput')
 
const selectedFile = ref<File | null>(null)
 
function onFileSelect(event: Event) {
  const input = event.target as HTMLInputElement
  selectedFile.value = input.files?.[0] || null
}
 
async function handleUpload() {
  if (!selectedFile.value) return
  const videoId = await upload(selectedFile.value)
  if (videoId) {
    navigateTo(`/video/${videoId}/edit`)
  }
}
</script>
 
<template>
  <div class="max-w-lg mx-auto">
    <!-- 拖拽上传区域 -->
    <div
      class="border-2 border-dashed border-gray-300 rounded-xl p-12 text-center cursor-pointer hover:border-primary"
      @click="fileInput?.click()"
    >
      <input
        ref="fileInput"
        type="file"
        accept="video/*"
        class="hidden"
        @change="onFileSelect"
      />
 
      <div v-if="!selectedFile">
        <p class="text-gray-500">点击或拖拽视频文件到此处</p>
        <p class="text-sm text-gray-400 mt-2">支持 MP4、MOV、AVI，最大 2GB</p>
      </div>
 
      <div v-else>
        <p class="font-medium">{{ selectedFile.name }}</p>
        <p class="text-sm text-gray-500">
          {{ (selectedFile.size / 1024 / 1024).toFixed(1) }} MB
        </p>
      </div>
    </div>
 
    <!-- 上传进度 -->
    <div v-if="status === 'uploading'" class="mt-6">
      <UProgress :value="progress" />
      <p class="text-sm text-gray-500 mt-2 text-center">上传中 {{ progress }}%</p>
    </div>
 
    <!-- 错误提示 -->
    <p v-if="error" class="text-red-500 text-sm mt-4">{{ error }}</p>
 
    <!-- 上传按钮 -->
    <UButton
      v-if="selectedFile && status !== 'uploading'"
      @click="handleUpload"
      class="mt-6 w-full"
    >
      开始上传
    </UButton>
  </div>
</template>

本章小结

直传架构：Presigned URL 让浏览器直传 S3/R2，服务端不经手视频文件，节省带宽
分片上传：S3 Multipart Upload，10MB 分片，失败只需重传失败分片
处理流水线：元数据提取 → HLS 转码（多码率）→ 封面截取 → 上传到对象存储
转码策略：根据原始分辨率动态选择输出码率（360p ~ 1080p），不超过源分辨率
异步编排：简单场景用后台异步，生产环境用 BullMQ 消息队列（支持重试和并发控制）
用户体验：拖拽上传、实时进度条、文件类型/大小校验