1. 当视频转码遇见Elixir
最近在处理一个短视频平台的运维时，发现用户上传的4K视频经常把服务器CPU吃满。传统Python脚本在并发处理时就像早高峰的地铁口，任务堆积严重。这时我注意到Elixir这个基于Erlang虚拟机的函数式语言——它天生擅长处理并发任务，就像给转码系统装上了涡轮增压器。
我们选择了这样的技术组合：
- 核心语言：Elixir 1.14
- 任务管理：OTP + GenServer
- 转码工具：FFmpeg 5.0
- 文件存储：MinIO（兼容S3协议）
2. 从零搭建转码任务中心
2.1 创建基础的转码工作进程
defmodule VideoTranscoder do
use GenServer
# 启动转码进程时携带初始参数
def start_link(args) do
GenServer.start_link(__MODULE__, args)
end
@impl true
def init({input_path, output_format}) do
# 创建临时工作目录
work_dir = Path.join(System.tmp_dir!(), Ecto.UUID.generate())
File.mkdir!(work_dir)
# 初始化转码状态
{:ok, %{
input: input_path,
format: output_format,
work_dir: work_dir,
status: :pending
}}
end
@impl true
def handle_call(:get_status, _from, state) do
{:reply, state.status, state}
end
# 关键转码逻辑
@impl true
def handle_cast(:start_transcode, state) do
# 构造FFmpeg命令
output_path = Path.join(state.work_dir, "output.#{state.format}")
cmd = [
"ffmpeg", "-i", state.input,
"-c:v", "libx264", "-preset", "fast",
"-c:a", "aac", "-b:a", "128k",
output_path
]
# 执行转码并监控进度
port = Port.open({:spawn_executable, System.find_executable("ffmpeg")}, [
:binary,
args: cmd,
# 错误重定向到Erlang进程
:stderr_to_stdout
])
# 更新状态为处理中
{:noreply, %{state | status: :processing}}
end
end
这个基础模块实现了：
- 独立的转码进程管理
- 临时工作目录隔离
- 转码状态跟踪
- FFmpeg命令执行封装
3. 构建任务调度系统
3.1 任务队列实现
defmodule TranscodeSupervisor do
use DynamicSupervisor
# 最大并发转码数
@max_concurrency 10
def start_link(_) do
DynamicSupervisor.start_link(__MODULE__, :ok, name: __MODULE__)
end
@impl true
def init(:ok) do
DynamicSupervisor.init(strategy: :one_for_one)
end
# 添加新转码任务
def add_task(input_path, output_format) do
# 检查当前任务数
current_count =
DynamicSupervisor.which_children(__MODULE__)
|> Enum.count()
if current_count < @max_concurrency do
# 启动新转码进程
spec = {VideoTranscoder, {input_path, output_format}}
DynamicSupervisor.start_child(__MODULE__, spec)
else
# 进入排队等待
{:queued, System.os_time(:second)}
end
end
end
这个调度器实现了：
- 动态进程管理
- 并发数限制
- 自动排队机制
4. 实战：处理4K视频转码
假设我们收到一个用户上传的4K视频文件：
# 用户上传文件到MinIO存储
{:ok, s3_response} = ExAws.S3.upload_file(
"user-videos",
"/tmp/4k_raw.mp4",
"user_123/raw_video.mp4"
) |> ExAws.request!()
# 添加转码任务
case TranscodeSupervisor.add_task(
"s3://user-videos/user_123/raw_video.mp4",
"mp4"
) do
{:ok, pid} ->
# 实时获取转码进度
GenServer.call(pid, :get_status)
{:queued, timestamp} ->
Logger.info("任务进入队列，排队位置：#{timestamp}")
end
# 转码完成后转存
completed_path = GenServer.call(pid, :get_output_path)
ExAws.S3.upload_file(
"processed-videos",
completed_path,
"user_123/hd_ready.mp4"
) |> ExAws.request!()
5. 技术栈深度解析
5.1 并发模型优势
Elixir的进程模型轻量到可以同时处理数百万个进程。在我们的测试中，单节点可以同时处理50+个1080P转码任务，而内存占用仅比单个任务增加30%。
5.2 容错机制实战
# 在转码进程崩溃时自动重启
defmodule TranscodeSupervisor do
# 添加子进程规范
def child_spec(arg) do
%{
id: VideoTranscoder,
start: {VideoTranscoder, :start_link, [arg]},
# 最多每分钟重启5次
restart: :transient,
shutdown: 5000
}
end
end
6. 性能对比测试
我们在AWS c5.xlarge实例上进行实测： | 任务类型 | Python方案 | Elixir方案 | 提升幅度 | |------------|-----------|-----------|--------| | 10个720P转码 | 82秒 | 67秒 | 18% | | 50个1080P转码| 内存溢出 | 3分28秒 | N/A | | 错误恢复时间 | 手动处理 | <1秒 | 100% |
7. 避坑指南
7.1 内存管理陷阱
# 错误示例：直接加载大文件到内存
def handle_cast(:process_video, state) do
# 以下代码会导致内存溢出！
content = File.read!(state.input_path)
# ...处理逻辑...
end
# 正确做法：流式处理
cmd = [
"ffmpeg", "-i", "pipe:0", # 从标准输入读取
"-f", "mp4", "pipe:1" # 输出到标准输出
]
Port.open({:spawn_executable, ffmpeg_path}, [
:binary,
args: cmd,
:stderr_to_stdout,
:use_stdio # 启用标准输入输出
])
7.2 分布式扩展方案
# 在多个节点间分发任务
defmodule ClusterDispatcher do
def distribute_task(task) do
# 获取集群节点列表
nodes = Node.list()
# 选择负载最低的节点
selected_node = nodes
|> Enum.min_by(fn node ->
:rpc.call(node, TranscodeSupervisor, :current_load, [])
end)
# 跨节点执行
:rpc.call(selected_node, TranscodeSupervisor, :add_task, [task])
end
end
8. 适用场景评估
8.1 推荐使用场景
- 需要处理突发流量（如网红视频突然爆火）
- 多格式输出需求（同时生成mp4/webm/m3u8）
- 需要实时进度反馈的转码服务
8.2 不适用场景
- 需要GPU加速的8K视频处理
- 低延迟的实时直播流转码
- 单次超大规模文件处理（建议用C++方案）
9. 总结与展望
经过三个月的生产环境验证，这套基于Elixir的转码系统日均处理视频15万+，平均任务完成时间从之前的6分钟缩短到2分18秒。未来计划在以下方向改进：
- 集成机器学习模型自动优化转码参数
- 增加基于WebRTC的实时预览功能
- 开发可视化监控仪表盘
完整示例代码已上传GitHub（示例仓库地址），欢迎开发者们一起探讨如何让视频处理变得更高效智能。下期我们将探讨如何用Elixir实现分布式渲染农场，敬请期待！