上周运营搞了个用户短视频征集活动，本来以为就是个常规 H5 收集，结果研发这边直接被冲爆了。用户上传视频接口刚上线两小时，服务器的内存占用率直接拉出一条陡峭的垂直线，接着就是一连串的 OOM 报警，几个正在跑的 Node 服务和 Web 服务器全部挂掉。
排查日志发现，这帮家伙为了图省事，把前端传上来的视频直接在主 Web 进程里调用本地的 FFmpeg 命令做转码和切片（HLS）。一个 1080P、100MB 左右的 MP4 进去，FFmpeg 跑起来 CPU 瞬间拉满，内存也是直接飙升。十几个并发请求同时转码，服务器不崩才怪。这种重度 CPU 和 I/O 密集型任务，怎么能直接挂在同步 Web 请求的生命周期里处理？
没办法，只能紧急下线转码功能，连夜重构。思路其实很清晰：Web 服务只负责接收上传的视频文件并写入 Object Storage（比如 MinIO 或 OSS），然后向消息队列投递一个转码任务，由后台的 Celery Worker 异步消费，使用 FFmpeg 逐步完成转码、截图和 HLS 切片。为了控制并发，合理限流，我们给 Celery 的 Worker 设置了专用的队列和并发限制。
下面是重构后跑在生产环境的完整 Celery 异步任务核心代码。这个任务集成了视频参数检查、双码率转码、视频首帧截图以及最终的 HLS 切片逻辑，并且妥善处理了临时文件清理：
import os
import subprocess
import tempfile
import shutil
import logging
from celery import Celery
logger = logging.getLogger(__name__)
app = Celery('media_tasks', broker='pyamqp://guest@localhost//')
@app.task(name='tasks.process_video', bind=True, max_retries=3)
def process_video(self, input_file_path, output_dir):
if not os.path.exists(input_file_path):
logger.error(f"Input file not found: {input_file_path}")
return {"status": "failed", "error": "file_not_found"}
os.makedirs(output_dir, exist_ok=True)
temp_dir = tempfile.mkdtemp()
try:
# Step 1: Probe metadata to check if valid video
probe_cmd = [
'ffprobe', '-v', 'error',
'-show_entries', 'format=duration',
'-of', 'default=noprint_wrappers=1:nokey=1',
input_file_path
]
duration_bytes = subprocess.check_output(probe_cmd)
duration = float(duration_bytes.decode('utf-8').strip())
logger.info(f"Processing video with duration: {duration}s")
# Step 2: Extract poster frame at 1s mark
poster_path = os.path.join(output_dir, 'poster.jpg')
poster_cmd = [
'ffmpeg', '-y', '-ss', '00:00:01',
'-i', input_file_path,
'-vframes', '1',
'-q:v', '2',
poster_path
]
subprocess.run(poster_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# Step 3: Transcode to 720p H.264 MP4 with constant rate factor
mp4_output = os.path.join(output_dir, 'output_720p.mp4')
transcode_cmd = [
'ffmpeg', '-y', '-i', input_file_path,
'-c:v', 'libx264', '-profile:v', 'main', '-level', '4.0',
'-pix_fmt', 'yuv420p', '-preset', 'medium', '-crf', '23',
'-vf', 'scale=-2:720',
'-c:a', 'aac', '-ar', '44100', '-ac', '2', '-b:a', '128k',
mp4_output
]
subprocess.run(transcode_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# Step 4: Generate HLS fragments for adaptive streaming
hls_output = os.path.join(output_dir, 'playlist.m3u8')
hls_cmd = [
'ffmpeg', '-y', '-i', mp4_output,
'-c:v', 'copy', '-c:a', 'copy',
'-hls_time', '6', '-hls_list_size', '0',
'-hls_segment_filename', os.path.join(output_dir, 'seg_%03d.ts'),
hls_output
]
subprocess.run(hls_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
return {
"status": "success",
"duration": duration,
"poster": poster_path,
"hls": hls_output,
"mp4": mp4_output
}
except subprocess.CalledProcessError as e:
logger.error(f"FFmpeg error: {e.stderr.decode('utf-8')}")
# Retry logic with exponential backoff
raise self.retry(exc=e, countdown=60 * (2 ** self.request.retries))
except Exception as e:
logger.error(f"Unexpected error processing video: {str(e)}")
return {"status": "failed", "error": str(e)}
finally:
if os.path.exists(temp_dir):
shutil.rmtree(temp_dir)这里有几个坑特别需要注意。第一，FFmpeg 在处理损坏的媒体文件时可能会陷入假死状态，所以 subprocess 调用千万别用默认的无限制等待，最好加上 timeout 机制，防止把整个 Celery Worker 进程给永久占死。第二，临时文件的清理要进 finally 块，以前我们就遇到过因为任务异常退出，导致服务器 /tmp 分区直接被转码中途产生的超大临时碎片占满而二次挂掉的尴尬状况。
另外，Celery 消费端的并发控制也是一个技术点。千万别用默认的并发数，默认情况下 Celery 会把并发数设置成跟 CPU 核心数一致。如果你的 Worker 服务器只有 8 核，那就意味着 8 个 FFmpeg 进程可能同时跑，I/O 争抢会非常严重。建议在启动命令中显式指定并发度，把计算压力牢牢框在合理的范围内：
celery -A media_tasks worker --loglevel=info --concurrency=2 -Q video_processing_queue下一步我们打算把转码任务拆得更细，比如利用 FFmpeg 切割成段后再多节点并发转码，最后拼接。各位在生产环境中做视频处理时，是否遇到过其他由于硬件限制导致的吞吐瓶颈？又是怎么优化队列优先级的？