影像技术实战20视频转码、抽帧、HLS 太慢影像处理任务队列异步化架构实战一、问题场景视频上传接口越来越慢最后直接超时影像系统里很多任务都很耗时视频转码 HLS 切片 视频抽帧 封面生成 OCR 识别 图片压缩 内容审核 AI 模型推理项目初期很多人会直接在接口里处理app.post(/upload)defupload():save_file()transcode_video()extract_frames()generate_thumbnail()return{status:ok}小文件没问题但真实上线后会出现1. 上传接口超时 2. Web 服务线程被占满 3. FFmpeg 任务互相抢 CPU 4. 用户无法查询处理进度 5. 任务失败无法重试 6. 服务重启后任务丢失 7. 同时上传多个视频直接打爆机器 8. 后台不知道任务卡在哪里本文解决的问题如何把视频转码、抽帧、封面生成等影像任务异步化设计一个可查询、可重试、可扩展的任务队列架构二、真实问题影像任务不适合同步 HTTP 接口HTTP 接口适合接收请求 保存任务记录 快速返回 task_id不适合等待 FFmpeg 跑几分钟 等待 AI 模型推理完成 等待 HLS 切片结束正确架构API 服务 ↓ 创建任务记录 ↓ 投递任务队列 ↓ Worker 消费任务 ↓ 执行影像处理 ↓ 更新任务状态 ↓ 前端轮询任务结果三、架构设计推荐最小可用架构media-task-system/ ├── api.py # 接收任务、查询状态 ├── worker.py # Worker 消费任务 ├── tasks.py # 具体影像任务 ├── store.py # 简单任务状态存储 └── media/ ├── transcode.py ├── thumbnail.py └── frames.py任务状态PENDING 等待处理 PROCESSING 正在处理 SUCCESS 成功 FAILED 失败正式系统建议状态入数据库。本文用 JSON 文件模拟方便复现。四、环境准备这里使用 Redis RQ 做轻量任务队列。mkdirmedia-task-systemcdmedia-task-system python-mvenv venv pipinstallfastapi uvicorn redis rq启动 Redisredis-server五、定义任务状态存储创建store.pyimportosimportjsonfromdatetimeimportdatetime TASK_DIRtask_storeos.makedirs(TASK_DIR,exist_okTrue)defnow():returndatetime.utcnow().isoformat()defsave_task(task_id:str,data:dict):data[updated_at]now()pathos.path.join(TASK_DIR,f{task_id}.json)withopen(path,w,encodingutf-8)asf:json.dump(data,f,ensure_asciiFalse,indent2)defload_task(task_id:str):pathos.path.join(TASK_DIR,f{task_id}.json)ifnotos.path.exists(path):returnNonewithopen(path,r,encodingutf-8)asf:returnjson.load(f)六、定义转码任务创建tasks.pyimportsubprocessimporttimefromstoreimportload_task,save_taskdeftranscode_video_task(task_id:str,input_path:str,output_path:str):taskload_task(task_id)task[status]PROCESSINGtask[started_at]time.time()save_task(task_id,task)cmd[ffmpeg,-y,-i,input_path,-c:v,libx264,-preset,veryfast,-crf,23,-c:a,aac,-b:a,128k,-pix_fmt,yuv420p,-movflags,faststart,output_path]resultsubprocess.run(cmd,stdoutsubprocess.PIPE,stderrsubprocess.PIPE,textTrue,encodingutf-8,errorsignore)taskload_task(task_id)ifresult.returncode!0:task[status]FAILEDtask[error]result.stderr[-2000:]else:task[status]SUCCESStask[output_path]output_path task[finished_at]time.time()task[ffmpeg_stderr_tail]result.stderr[-2000:]save_task(task_id,task)returntask注意任务开始、成功、失败都要更新状态。 stderr 尾部必须保存排查 FFmpeg 问题非常关键。七、API 提交任务创建api.pyimportuuidfromfastapiimportFastAPIfromredisimportRedisfromrqimportQueue,Retryfromtasksimporttranscode_video_taskfromstoreimportsave_task,load_task,now appFastAPI()redis_connRedis(hostlocalhost,port6379,db0)queueQueue(media,connectionredis_conn)app.post(/tasks/transcode)defcreate_transcode_task(input_path:str,output_path:str):task_iduuid.uuid4().hextask{task_id:task_id,task_type:transcode,input_path:input_path,output_path:output_path,status:PENDING,created_at:now(),updated_at:now(),error:None}save_task(task_id,task)jobqueue.enqueue(transcode_video_task,task_id,input_path,output_path,job_timeout3600,retryRetry(max2,interval[30,120]))task[job_id]job.idsave_task(task_id,task)return{task_id:task_id,status:PENDING}app.get(/tasks/{task_id})defget_task(task_id:str):taskload_task(task_id)iftaskisNone:return{status:NOT_FOUND}returntask启动 APIuvicorn api:app--reload八、启动 Worker创建worker.pyfromredisimportRedisfromrqimportWorker,Queue redis_connRedis(hostlocalhost,port6379,db0)queueQueue(media,connectionredis_conn)workerWorker([queue],connectionredis_conn)worker.work()启动python worker.py提交任务curl-XPOSThttp://127.0.0.1:8000/tasks/transcode?input_pathinput.mp4output_pathoutput.mp4查询任务curlhttp://127.0.0.1:8000/tasks/{task_id}九、为什么这套架构更稳同步接口的问题请求一直等待 用户体验差 Web 进程被占用 失败无法恢复 无法查询进度异步队列的优势接口快速返回 Worker 独立处理 任务状态可查 失败可重试 并发可控制 后续易扩展十、并发控制建议视频任务很重不要盲目开很多 Worker。建议单机 2 核1 个 Worker 单机 4 核1-2 个 Worker 单机 8 核2-4 个 Worker具体要看CPU 磁盘 IO 内存 视频分辨率 任务类型FFmpeg 转码通常 CPU 压力很大抽帧则 CPU 磁盘 IO 都重。十一、踩坑记录坑 1把视频二进制塞进队列错误。队列里应该传文件路径 对象存储 URL 任务 ID不要传大文件内容。坑 2任务状态只存在队列里队列不是业务数据库。正式系统必须把任务状态落库。坑 3没有失败日志没有 stderrFFmpeg 失败基本靠猜。坑 4Worker 无限扩容会打爆 CPU 和磁盘。坑 5临时文件不清理视频任务会产生大量临时文件要有生命周期管理。十二、适合收藏影像任务字段设计task_id task_type input_path output_path status progress error retry_count created_at updated_at started_at finished_at worker_id ffmpeg_stderr_tail十三、避坑清单1. 不要在上传接口里直接转码 2. 不要把大文件放进队列 3. 不要只依赖队列状态 4. 不要不保存失败日志 5. 不要无限增加 Worker 6. 不要没有任务超时 7. 不要没有失败重试 8. 不要不清理临时文件十四、总结与优化建议影像处理任务的核心不是单个 FFmpeg 命令而是任务调度和资源治理。工程建议API 只接收任务 Worker 执行重任务 任务状态持久化 失败日志可查 并发数量受控 支持重试和超时后续升级1. Celery RabbitMQ 2. 数据库任务表 3. 对象存储输入输出 4. 任务进度解析 5. GPU Worker 池 6. 任务优先级 7. Prometheus 监控只要涉及视频转码、抽帧、HLS、AI 推理就不要再把它们写在同步接口里。