对象存储+消息队列+FFmpeg：小视频处理全链路实现

小视频类产品的技术难点，往往不在“拍视频”，而在拍完之后的那条链路：一个用户把原始视频传到服务器，另一个用户要能顺畅播放出来，中间隔着上传、转码、存储、分发、播放器兼容好几道坎。很多内容团队花大力气做选题和剪辑，等真要发布一个能上线的版本时，才发现整条链路到处是问题。

这篇文章针对“唐人小视频”这样的内容型小视频项目，把一条最常见的视频处理链路完整拆开：客户端上传原始视频，服务端把它转成 HLS 流并存储，播放端通过 HTTP 加载视频。先建立全局认知，再给出可运行的参考实现。

文章提供的代码是完整的，不依赖某个特定商业平台，核心组件只是对象存储、消息队列和 FFmpeg。你跑通这个最小工程后，再往短视频完整产品迁移，技术上会顺手很多。

1. 这篇文章真正要解决的问题

先说结论：一个看起来只是“上传视频给别人看”的功能，如果要稳定运行，至少要解决四个问题。

第一，大文件怎么传。一个几分钟的原始视频可能上百 MB，不能靠一个普通 HTTP 接口慢慢接收，服务端很容易超时和占满连接。更合理的做法是让客户端直接上传到对象存储，服务端只负责生成凭证和记录任务。

第二，格式怎么统一。移动端录出来的视频可能是 MP4、MOV、MKV，编码可能是 H.264、H.265、VP9。如果所有终端都直接播放原始文件，老旧的播放器、低端手机、部分浏览器会出现有声音没画面，或者干脆不支持。所以服务端要有一个可靠的转码环节。

第三，转码怎么异步化。FFmpeg 转一次视频要几秒到几十秒，绝对不能放在 HTTP 请求里同步等待。需要把任务丢进队列，由后台 worker 慢慢消费。

第四，播放怎么兼容。如果直接用<video>标签播放原始 MP4，浏览器要等下载到一定进度才能播，拖动进度也不流畅。实际工程里更常见的方案是转成 HLS，让播放器连续拉取多个小分片，起播更快、拖动更顺畅。

如果你是后端工程师，这篇文章能让你理解对象存储、消息队列、FFmpeg 三类组件是怎么组合成一条视频链路的；如果你是客户端工程师，这篇文章能让你知道上传接口设计逻辑是什么，播放地址又是怎么产生的；如果你是准备做小视频产品的新手，这篇文章可以直接当作第一个能跑通的技术原型。

2. 核心概念：对象存储、异步任务与 HLS

先解释几个反复出现的概念，避免后文代码和术语混淆。

2.1 对象存储：不再自己管理视频文件

对象存储可以理解成“一个支持 DELETE 和 GET 的网盘服务”。它不依赖传统服务器磁盘，不受单机容量限制，通常通过 HTTP 接口访问。MinIO 是兼容 S3 协议的常见实现，适合本地开发和学习。

对象存储解决的核心问题是：小视频应用不能把视频直接存在应用服务器的本地目录里。应用服务器可能被重启、扩容、缩容，本地文件会丢，也搬不动。对象存储把文件变成对象，只要给足权限，应用服务器和客户端都能通过 URL 上传下载。

2.2 预签名 URL：把上传压力从应用服务器挪走

预签名 URL 可以这样理解：服务端生成一个“一次性临时上传链接”，客户端拿着这个链接直接把文件上传到对象存储。应用服务器不用接收大头文件，只负责业务逻辑，例如生成任务 ID、记录元数据、判断任务状态。

这个机制对自建小视频服务尤其重要，因为它能把带宽成本分散到对象存储层，而不是集中在一台后端机器上。

2.3 异步任务与消息队列：不要把 FFmpeg 卡在接口里

FFmpeg 是一个强大的音视频处理工具，可以完成转码、转封装、截图、抽帧等操作。但它处理视频很耗时，如果让 HTTP 请求一直等 FFmpeg 跑完，用户请求就会超时，服务器并发也会被拖垮。

所以正确的流程是：

1. 客户端上传完成；
2. 服务端把任务写入消息队列；
3. worker 从队列里读任务；
4. worker 调用 FFmpeg 转码；
5. 转码结果写回对象存储。

Redis 的 List 可以当成一个最基础的队列来演示，生产环境建议替换为 RabbitMQ、Kafka 或云厂商提供的消息服务，核心思路是一致的。

2.4 HLS：切分小文件，播放更顺滑

HLS 是苹果主导的流媒体协议。用 FFmpeg 把原始视频切成一个个时长 6 秒的小 TS 文件，并生成一个 index.m3u8 索引文件。播放器先加载索引文件，再按顺序拉取 TS 分片，达到按需加载的效果。

从工程角度看，HLS 把“大文件分发”变成了“小文件分发”，更适合 CDN 缓存和弱网播放。这也是很多小视频项目会默认采用转 HLS 方案的原因。

3. 环境准备与前置条件

为了阅读体验，不要刻意省略环境信息。以下版本不是苛刻要求，你只要保证组件可用，版本略有差异也能跑通。

推荐准备这些工具：

• Docker，用于启动 Redis 和 MinIO；
• Python 3.10 或 3.11，用于编写上传服务和转码 worker；
• FFmpeg，本机需要能执行ffmpeg命令；
• 一个测试视频文件，例如本地用手机录一段 MP4 即可。

版本方面，本文演示不绑定精确版本。如果你从零开始，建议尽量使用较新的稳定版本。FFmpeg 版本偏低时，可能缺少 H.264 编码器，这会导致后面的转码命令报错。

先创建项目目录：

mkdir chv-demo && cd chv-demo

我优先使用 Docker Compose 启动 Redis 和 MinIO。这样可以省去本地安装软件的麻烦，也让环境更接近生产形态。

创建docker-compose.yml：

services: redis: image: redis:7-alpine container_name: chv-redis ports: - "6379:6379" minio: image: minio/minio:latest container_name: chv-minio command: server /data --console-address ":9001" environment: MINIO_ROOT_USER: minioadmin MINIO_ROOT_PASSWORD: minioadmin ports: - "9000:9000" - "9001:9001" volumes: - minio-data:/data volumes: minio-data:

启动依赖：

docker compose up -d redis minio

启动后，MinIO 的 API 地址是http://127.0.0.1:9000，控制台地址是http://127.0.0.1:9001，账号密码均为minioadmin。Redis 默认监听127.0.0.1:6379。

4. 初始化对象存储：创建桶与访问策略

存放视频之前需要创建一个桶。桶可以理解为对象存储里的顶层目录。整套参考实现统一使用名为videos的桶，原始视频存放在raw/前缀下，转码后的 HLS 分片存放在hls/前缀下。

在项目里创建一个init.py，它会完成桶初始化和公共读策略配置：

# 文件路径：chv-demo/init.py import json import boto3 from botocore.config import Config client = boto3.client( "s3", endpoint_url="http://127.0.0.1:9000", aws_access_key_id="minioadmin", aws_secret_access_key="minioadmin", region_name="us-east-1", config=Config(signature_version="s3v4"), ) bucket = "videos" try: client.head_bucket(Bucket=bucket) print("bucket exists:", bucket) except Exception: client.create_bucket(Bucket=bucket) print("bucket created:", bucket) # 演示环境允许公开读取视频，生产环境不要这样配置 policy = { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": {"AWS": ["*"]}, "Action": ["s3:GetObject"], "Resource": [f"arn:aws:s3:::{bucket}/*"], } ], } client.put_bucket_policy(Bucket=bucket, Policy=json.dumps(policy)) print("public read policy set")

在真实项目中，桶权限需要更细地设计。本文演示 HLS 播放，所以让videos桶支持公共读取。但这不代表原始视频也能公开访问。你可以额外限制raw/目录禁止公开 GET，或者后续引入签名播放地址和防盗链，这部分我会在最佳实践章节展开。

5. 上传服务：预签名 URL 与任务队列

上传服务的职责很清晰：创建上传任务，返回预签名 URL；客户端完成上传后，调用 confirmation 接口，把任务 ID 写入队列。

为什么不直接让客户端 POST 文件给后端？因为这样应用服务器要承担全部带宽和超时压力。更好的做法是后端只返回一个带签名的临时 URL，客户端直接把文件 PUT 到 MinIO。后端不接触视频二进制流，接口压力小很多。

在项目里创建main.py：

# 文件路径：chv-demo/main.py import uuid import boto3 import redis from botocore.config import Config from fastapi import FastAPI, HTTPException from pydantic import BaseModel REDIS_URL = "redis://127.0.0.1:6379/0" S3_ENDPOINT = "http://127.0.0.1:9000" ACCESS_KEY = "minioadmin" SECRET_KEY = "minioadmin" BUCKET = "videos" app = FastAPI(title="唐人小视频-上传服务") redis_client = redis.Redis.from_url(REDIS_URL, decode_responses=True) def get_s3_client(): return boto3.client( "s3", endpoint_url=S3_ENDPOINT, aws_access_key_id=ACCESS_KEY, aws_secret_access_key=SECRET_KEY, region_name="us-east-1", config=Config(signature_version="s3v4"), ) class CreateVideoRequest(BaseModel): title: str file_name: str @app.post("/api/v1/videos") def create_video_task(req: CreateVideoRequest): upload_id = uuid.uuid4().hex object_key = f"raw/{upload_id}.mp4" s3 = get_s3_client() upload_url = s3.generate_presigned_url( "put_object", Params={"Bucket": BUCKET, "Key": object_key, "ContentType": "video/mp4"}, ExpiresIn=3600, ) # 这里先不写数据库，只把元数据放在一个简单结构里打印 # 生产环境需要把 upload_id、title、status 落到数据库 print(f"create task: upload_id={upload_id}, title={req.title}, file_name={req.file_name}") return { "upload_id": upload_id, "upload_url": upload_url, "status": "pending", } @app.post("/api/v1/videos/{video_id}/complete") def finish_video_upload(video_id: str): """客户端上传完成后调用，确认任务可入队。""" s3 = get_s3_client() object_key = f"raw/{video_id}.mp4" try: s3.head_object(Bucket=BUCKET, Key=object_key) except Exception: raise HTTPException(status_code=404, detail="raw video not found") redis_client.rpush("video_jobs", video_id) return {"upload_id": video_id, "status": "queued"}

这个服务里有一个值得注意的设计：create_video_task不立刻把任务入队，而是等客户端确认上传完成之后再入队。很多第一次写视频服务的人最容易在这里出错：一创建任务就通知转码 worker，结果 worker 拿到的可能是半个文件，或者对象还没有写入，最终转码失败。

先入队问题只暴露在 raw 对象存在后；如果对象不存在，complete接口会直接返回 404。

5.1 客户端模拟上传

启动上传服务前，先安装依赖：

pip install fastapi uvicorn boto3 redis

启动服务：

uvicorn main:app --host 0.0.0.0 --port 8000

现在用 curl 创建一个上传任务，假设测试视频是demo.mp4：

curl -X POST http://127.0.0.1:8000/api/v1/videos \ -H "Content-Type: application/json" \ -d '{"title": "唐风街拍 demo", "file_name": "demo.mp4"}'

响应大致是这样的字段结构：

{ "upload_id": "f3d2c1b0a1b24e6f8e0f1a2b3c4d5e6f", "upload_url": "http://127.0.0.1:9000/videos/raw/f3d2...f.mp4?X-Amz-Algorithm=...&X-Amz-Signature=...", "status": "pending" }

用返回里的upload_url执行 PUT 上传，注意复制你实际拿到的完整 URL：

curl -X PUT -H "Content-Type: video/mp4" \ --upload-file demo.mp4 \ "http://127.0.0.1:9000/videos/raw/f3d2...f.mp4?X-Amz-Algorithm=..."

上传完成后，通知后端入队：

curl -X POST http://127.0.0.1:8000/api/v1/videos/f3d2...f/complete

到这里，你的视频原始文件已经进入对象存储，任务 ID 也放到了 Redis 队列里。后面的转码 worker 会消费这个任务。

6. 转码 Worker：把 MP4 变成 HLS

转码 worker 是链路里最核心的工程部分。它做四件事：

1. 从 Redis 队列里取出任务；
2. 从 MinIO 下载原始视频；
3. 用 FFmpeg 转成 HLS 分片；
4. 把 HLS 文件回传到 MinIO。

创建worker.py：

# 文件路径：chv-demo/worker.py import os import shutil import subprocess import boto3 import redis from botocore.config import Config REDIS_URL = "redis://127.0.0.1:6379/0" S3_ENDPOINT = "http://127.0.0.1:9000" ACCESS_KEY = "minioadmin" SECRET_KEY = "minioadmin" BUCKET = "videos" WORK_DIR = "/tmp/chv_work" def get_s3_client(): return boto3.client( "s3", endpoint_url=S3_ENDPOINT, aws_access_key_id=ACCESS_KEY, aws_secret_access_key=SECRET_KEY, region_name="us-east-1", config=Config(signature_version="s3v4"), ) def download_raw(s3, video_id): raw_key = f"raw/{video_id}.mp4" local_path = os.path.join(WORK_DIR, video_id, "source.mp4") os.makedirs(os.path.dirname(local_path), exist_ok=True) s3.download_file(BUCKET, raw_key, local_path) return local_path def transcode_to_hls(source_path, out_dir): os.makedirs(out_dir, exist_ok=True) cmd = [ "ffmpeg", "-y", "-i", source_path, "-c:v", "libx264", "-preset", "veryfast", "-crf", "23", "-c:a", "aac", "-b:a", "128k", "-hls_time", "6", "-hls_list_size", "0", "-hls_segment_filename", os.path.join(out_dir, "seg_%04d.ts"), os.path.join(out_dir, "index.m3u8"), ] subprocess.run(cmd, check=True) def upload_hls_dir(s3, video_id, local_dir): prefix = f"hls/{video_id}" for file_name in os.listdir(local_dir): local_file = os.path.join(local_dir, file_name) remote_key = f"{prefix}/{file_name}" if file_name.endswith(".m3u8"): content_type = "application/vnd.apple.mpegurl" elif file_name.endswith(".ts"): content_type = "video/mp2t" else: content_type = "application/octet-stream" s3.upload_file( local_file, BUCKET, remote_key, ExtraArgs={"ContentType": content_type}, ) def process_video(video_id): s3 = get_s3_client() local_dir = os.path.join(WORK_DIR, video_id) hls_dir = os.path.join(local_dir, "hls") try: source_path = download_raw(s3, video_id) transcode_to_hls(source_path, hls_dir) upload_hls_dir(s3, video_id, hls_dir) print(f"success: {video_id} -> hls/{video_id}/index.m3u8") finally: # 本地临时文件在确认上传成功后全部清理 shutil.rmtree(local_dir, ignore_errors=True) def main(): r = redis.Redis.from_url(REDIS_URL, decode_responses=True) print("worker started, waiting for video_jobs ...") while True: item = r.brpop("video_jobs", timeout=5) if item is None: continue video_id = item[1] try: process_video(video_id) except Exception as e: # 生产环境要在这里记录完整异常并设计重试策略 # 这里先放入死信队列，避免消息丢失 r.lpush("video_dead_letter", video_id) print(f"failed: {video_id}, error={e}, moved to video_dead_letter") if __name__ == "__main__": main()

这段代码有几个关键点。

第一，FFmpeg 命令传参用列表而不是字符串拼接。谁都不应该用 shell 去拼一个带有用户可控文件名的命令，列表传参会去掉 shell 层，避免命令注入。

第二，视频 ID 全局使用 UUID，worker 不信任来自用户上传的原始文件名。这是刻意为之。一旦把用户提供的文件名拼到路径或命令里，会引发路径穿越问题。

第三，转码失败后会把消息移到video_dead_letter死信队列。直接brpop消费消息后，如果程序中途崩溃，队列消息已经丢失，所以失败至少要保留痕迹。

启动 worker：

python worker.py

看到输出worker started, waiting for video_jobs ...之后，worker 会一直监听 Redis 队列。

如果此时你还记得第 6 章创建的转码任务，可以再次验证。如果一切正常，日志会打印：

success: f3d2c1b0... -> hls/f3d2c1b0.../index.m3u8

打开 MinIO 控制台，进入videos桶，会看到hls/<video_id>/目录下出现了index.m3u8和多个seg_xxxx.ts文件。这就是可以被播放器连续加载的 HLS 分片。

7. H5 播放页面验证

HLS 在 iOS 原生 Safari 里可以直接播放，但在 PC Chrome 和部分 Android 浏览器里需要借助 hls.js 播放器。

创建一个简单页面index.html：

<!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8" /> <title>唐人小视频-播放验证</title> <script src="https://cdn.jsdelivr.net/npm/hls.js@1"></script> </head> <body> <h3>HLS 播放验证</h3> <video id="video" controls style="width: 640px; max-width: 100%" muted></video> <script> // 将下面的 URL 替换成 worker 日志里输出的实际地址 const videoUrl = 'http://127.0.0.1:9000/videos/hls/<VIDEO_ID>/index.m3u8'; const video = document.getElementById('video'); if (video.canPlayType('application/vnd.apple.mpegurl')) { // 原生支持 HLS 的浏览器直接播放 video.src = videoUrl; } else if (Hls.isSupported()) { const hls = new Hls(); hls.loadSource(videoUrl); hls.attachMedia(video); } else { alert('当前环境不支持 HLS 播放'); } </script> </body> </html>

我在这里默认 HLS 文件是公开可读的，所以播放器能直接拉取 m3u8 和 TS 分片。生产环境下，这些地址通常需要附加签名参数或从 CDN 分发。

验证是否播放成功，可以看两个指标：

1. 播放器出现画面，时间可以自由拖动；
2. 浏览器 Network 面板里的 TS 请求是连续出现的，而不是一次性请求一个大 MP4。

如果只有第一帧加载，拖动后立刻转圈，说明分片加载异常，要回到转码内容和播放地址层面排查。

你还可以用命令行快速检查 HLS 的元数据：

curl -I http://127.0.0.1:9000/videos/hls/<VIDEO_ID>/index.m3u8

正常响应里的Content-Type应该是：

application/vnd.apple.mpegurl

如果你的 MinIO 上传时没有设置 Content-Type，这里很可能返回binary/octet-stream，播放器会直接拒绝播放。这也是新手转 HLS 分片时最容易踩的坑。

8. 常见问题与排查思路

把上面这套流程跑下来，大概率会遇到下面几类问题。我把高频现象整理成一张排查表。