对象存储+消息队列+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 跑完用户请求就会超时服务器并发也会被拖垮。所以正确的流程是客户端上传完成服务端把任务写入消息队列worker 从队列里读任务worker 调用 FFmpeg 转码转码结果写回对象存储。Redis 的 List 可以当成一个最基础的队列来演示生产环境建议替换为 RabbitMQ、Kafka 或云厂商提供的消息服务核心思路是一致的。2.4 HLS切分小文件播放更顺滑HLS 是苹果主导的流媒体协议。用 FFmpeg 把原始视频切成一个个时长 6 秒的小 TS 文件并生成一个 index.m3u8 索引文件。播放器先加载索引文件再按顺序拉取 TS 分片达到按需加载的效果。对比项直接播放原始 MP4转成 HLS 分发起播速度需要下载更多数据才能播放加载第一个小分片就能播拖动进度必须跳到对应偏移量下载播放器按索引分段拉取清晰度切换通常不支持可以按码率切换编码兼容依赖浏览器和客户端支持服务端统一压制从工程角度看HLS 把“大文件分发”变成了“小文件分发”更适合 CDN 缓存和弱网播放。这也是很多小视频项目会默认采用转 HLS 方案的原因。3. 环境准备与前置条件为了阅读体验不要刻意省略环境信息。以下版本不是苛刻要求你只要保证组件可用版本略有差异也能跑通。推荐准备这些工具Docker用于启动 Redis 和 MinIOPython 3.10 或 3.11用于编写上传服务和转码 workerFFmpeg本机需要能执行ffmpeg命令一个测试视频文件例如本地用手机录一段 MP4 即可。版本方面本文演示不绑定精确版本。如果你从零开始建议尽量使用较新的稳定版本。FFmpeg 版本偏低时可能缺少 H.264 编码器这会导致后面的转码命令报错。先创建项目目录mkdir chv-demo cd chv-demo我优先使用 Docker Compose 启动 Redis 和 MinIO。这样可以省去本地安装软件的麻烦也让环境更接近生产形态。创建docker-compose.ymlservices: 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_urlhttp://127.0.0.1:9000, aws_access_key_idminioadmin, aws_secret_access_keyminioadmin, region_nameus-east-1, configConfig(signature_versions3v4), ) bucket videos try: client.head_bucket(Bucketbucket) print(bucket exists:, bucket) except Exception: client.create_bucket(Bucketbucket) print(bucket created:, bucket) # 演示环境允许公开读取视频生产环境不要这样配置 policy { Version: 2012-10-17, Statement: [ { Effect: Allow, Principal: {AWS: [*]}, Action: [s3:GetObject], Resource: [farn:aws:s3:::{bucket}/*], } ], } client.put_bucket_policy(Bucketbucket, Policyjson.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_responsesTrue) def get_s3_client(): return boto3.client( s3, endpoint_urlS3_ENDPOINT, aws_access_key_idACCESS_KEY, aws_secret_access_keySECRET_KEY, region_nameus-east-1, configConfig(signature_versions3v4), ) 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 fraw/{upload_id}.mp4 s3 get_s3_client() upload_url s3.generate_presigned_url( put_object, Params{Bucket: BUCKET, Key: object_key, ContentType: video/mp4}, ExpiresIn3600, ) # 这里先不写数据库只把元数据放在一个简单结构里打印 # 生产环境需要把 upload_id、title、status 落到数据库 print(fcreate 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 fraw/{video_id}.mp4 try: s3.head_object(BucketBUCKET, Keyobject_key) except Exception: raise HTTPException(status_code404, detailraw 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.mp4curl -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 上传注意复制你实际拿到的完整 URLcurl -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 是链路里最核心的工程部分。它做四件事从 Redis 队列里取出任务从 MinIO 下载原始视频用 FFmpeg 转成 HLS 分片把 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_urlS3_ENDPOINT, aws_access_key_idACCESS_KEY, aws_secret_access_keySECRET_KEY, region_nameus-east-1, configConfig(signature_versions3v4), ) def download_raw(s3, video_id): raw_key fraw/{video_id}.mp4 local_path os.path.join(WORK_DIR, video_id, source.mp4) os.makedirs(os.path.dirname(local_path), exist_okTrue) s3.download_file(BUCKET, raw_key, local_path) return local_path def transcode_to_hls(source_path, out_dir): os.makedirs(out_dir, exist_okTrue) 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, checkTrue) def upload_hls_dir(s3, video_id, local_dir): prefix fhls/{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(fsuccess: {video_id} - hls/{video_id}/index.m3u8) finally: # 本地临时文件在确认上传成功后全部清理 shutil.rmtree(local_dir, ignore_errorsTrue) def main(): r redis.Redis.from_url(REDIS_URL, decode_responsesTrue) print(worker started, waiting for video_jobs ...) while True: item r.brpop(video_jobs, timeout5) 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(ffailed: {video_id}, error{e}, moved to video_dead_letter) if __name__ __main__: main()这段代码有几个关键点。第一FFmpeg 命令传参用列表而不是字符串拼接。谁都不应该用 shell 去拼一个带有用户可控文件名的命令列表传参会去掉 shell 层避免命令注入。第二视频 ID 全局使用 UUIDworker 不信任来自用户上传的原始文件名。这是刻意为之。一旦把用户提供的文件名拼到路径或命令里会引发路径穿越问题。第三转码失败后会把消息移到video_dead_letter死信队列。直接brpop消费消息后如果程序中途崩溃队列消息已经丢失所以失败至少要保留痕迹。启动 workerpython 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 langzh-CN head meta charsetUTF-8 / title唐人小视频-播放验证/title script srchttps://cdn.jsdelivr.net/npm/hls.js1/script /head body h3HLS 播放验证/h3 video idvideo controls stylewidth: 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 分发。验证是否播放成功可以看两个指标播放器出现画面时间可以自由拖动浏览器 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. 常见问题与排查思路把上面这套流程跑下来大概率会遇到下面几类问题。我把高频现象整理成一张排查表。问题现象可能原因排查方式解决方案上传 URL 无法访问MinIO 未启动或访问端口不对curl -I http://127.0.0.1:9000/minio/health/live检查 docker compose 是否正常运行上传后 complete 返回 404视频 raw 对象不存在登录 MinIO 控制台查看raw/前缀确认 PUT 请求成功不要直接把浏览器返回内容当上传成功worker 一直等不到任务video_jobs队列名不统一redis-cli llen video_jobs检查 main.py 和 worker.py 的队列名是否一致FFmpeg 报 unknown encoder libx264本地 FFmpeg 没有编译 H.264 编码器ffmpeg -encoders | grep 264更换 FFmpeg 版本或安装带 x264 的构建版本播放器加载 m3u8 失败m3u8 的 Content-Type 不对curl -I查看响应头上传 m3u8 时指定application/vnd.apple.mpegurl播放黑屏无声音转码命令或音频编码不支持单独在本地执行 FFmpeg 命令测试输出文件确认输出文件是独立可播放的 HLS 文件播放时有声音没画面视频编码 H.265 不被浏览器支持用 ffprobe 查看原始视频编码统一转码为 H.264确保终端兼容性worker 处理失败