一台机器转不完：把 ffmpeg 任务分发到 8 台机器的分布式转码实践

#!/bin/bash
# bench_single.sh
total_src=0
total_time=0
for f in samples/*.mp4; do
    dur=$(ffprobe -v error -show_entries format=duration -of csv=p=0 "$f")
    start=$(date +%s.%N)
    ffmpeg -y -v error -i "$f" -c:v libx265 -crf 27 -preset slow -c:a aac -b:a 128k /tmp/out.mp4
    end=$(date +%s.%N)
    t=$(awk "BEGIN{print $end-$start}")
    total_src=$(awk "BEGIN{print $total_src+$dur}")
    total_time=$(awk "BEGIN{print $total_time+$t}")
    echo "$f: ${dur}s 素材 / ${t}s 耗时"
done
echo "速度倍率: $(awk "BEGIN{printf \"%.2f\", $total_src/$total_time}")x"

总素材时长 42000 分钟 ÷ 8 = 5250 分钟 = 87.5 小时 = 3.6 天

import redis
import json
import time

r = redis.Redis(host='queue-host', port=6379, db=0)

QUEUE = 'transcode:queue'
PROCESSING = 'transcode:processing'


def claim(timeout=5):
    """原子地领取一个任务：从 QUEUE 移到 PROCESSING。"""
    item = r.brpoplpush(QUEUE, PROCESSING, timeout=timeout)
    if item:
        return json.loads(item)
    return None


def ack(item):
    """任务完成，从 PROCESSING 里删掉。"""
    r.lrem(PROCESSING, 1, json.dumps(item, sort_keys=True))


def requeue_stale(max_age=3600):
    """回收超时未完成任务（一般是 worker 挂了）。"""
    now = time.time()
    for raw in r.lrange(PROCESSING, 0, -1):
        item = json.loads(raw)
        if now - item['claimed_at'] > max_age:
            # 移回队列（用 LREM 保证原子性，避免重复入队）
            if r.lrem(PROCESSING, 1, raw):
                item['claimed_at'] = 0
                item['retries'] = item.get('retries', 0) + 1
                r.lpush(QUEUE, json.dumps(item))
                print('requeued stale task:', item['id'])

# 机器 A
celery -A proj worker -Q transcode --concurrency=8 --hostname=workerA@%h

# 机器 B、C、...
celery -A proj worker -Q transcode --concurrency=8 --hostname=workerB@%h

CELERY_TASK_ACKS_LATE = True                  # 执行完才确认，挂了会重新投递
CELERY_BROKER_TRANSPORT_OPTIONS = {'visibility_timeout': 7200}
CELERY_WORKER_PREFETCH_MULTIPLIER = 1         # 长任务必须关掉预取

本地 SSD:  处理 1080p 素材 7.8x 速度
NFS 挂载:  处理同样素材 2.1x 速度   ← 慢了近 4 倍

import shutil
import tempfile

def process(item):
    with tempfile.TemporaryDirectory(dir='/local_ssd') as tmp:
        src = os.path.join(tmp, 'in.mp4')
        dst = os.path.join(tmp, 'out.mp4')
        download(item['src_path'], src)        # 从 NFS/对象存储拉到本地
        run_ffmpeg(src, dst)
        upload(dst, item['dst_path'])          # 传回去

OSError: [Errno 116] Stale file handle: '/mnt/nfs/xxx.mp4'

MinIO (源桶)  →  下载到本地 SSD  →  ffmpeg 转码  →  上传 MinIO (目标桶)  →  删本地

#!/usr/bin/env python3
import os
import signal
import socket
import subprocess
import time
import json
import redis

HOST = socket.gethostname()
CONCURRENCY = int(os.getenv('WORKER_CONCURRENCY', '8'))
CAPABILITY = os.getenv('WORKER_CAP', 'cpu')      # cpu / gpu
_running = True


def shutdown(signum, frame):
    """收到 SIGTERM：不再领新任务，把手上的干完再退出。"""
    global _running
    print(f'[{HOST}] received SIGTERM, finishing current task...')
    _running = False


signal.signal(signal.SIGTERM, shutdown)
signal.signal(signal.SIGINT, shutdown)


def heartbeat(r, item):
    """上报心跳，让别人知道这个任务还活着。"""
    r.setex(f'task:hb:{item["id"]}', 300, json.dumps({
        'host': HOST, 'ts': time.time(), 'cap': CAPABILITY,
    }))


def main():
    r = redis.Redis(...)
    while _running:
        item = claim(timeout=5)
        if not item:
            continue
        try:
            hb_stop = start_heartbeat_thread(r, item)   # 后台线程每 30 秒上报
            process(item)
            ack(item)
        except Exception as e:
            fail(item, str(e))
        finally:
            hb_stop.set()
    print(f'[{HOST}] exited gracefully')


if __name__ == '__main__':
    main()

QUEUE = f'transcode:{CAPABILITY}'

tmpdir = tempfile.mkdtemp(prefix=f'{HOST}_{os.getpid()}_', dir='/local_ssd')

MAX_RETRY = 2

def fail(item, reason):
    retries = item.get('retries', 0)
    kind = classify(reason)

    if kind == 'corrupted' or retries >= MAX_RETRY:
        # 进死信队列，人工处理
        r.lpush('transcode:dead', json.dumps({**item, 'reason': reason}))
        r.lrem(PROCESSING, 1, json.dumps(item, sort_keys=True))
        return

    item['retries'] = retries + 1
    # 降低 preset 再试（超时场景下有效）
    if kind == 'timeout' and item.get('preset') == 'slow':
        item['preset'] = 'medium'
    r.lpush(QUEUE, json.dumps(item))
    r.lrem(PROCESSING, 1, json.dumps(item, sort_keys=True))

CORRUPT_PATTERNS = [
    'Invalid data found when processing input',
    'moov atom not found',
    'error while decoding MB',
    'Could not find codec parameters',
]

def classify(reason: str) -> str:
    for p in CORRUPT_PATTERNS:
        if p in reason:
            return 'corrupted'
    if 'timeout' in reason.lower() or 'Timeout' in reason:
        return 'timeout'
    if any(k in reason for k in ('ConnectionError', 'TimeoutError', 'S3')):
        return 'network'
    if 'No space left' in reason:
        return 'disk'
    return 'unknown'

import redis
import time

r = redis.Redis(...)

TOTAL = 9234

while True:
    pending = r.llen('transcode:queue')
    processing = r.llen('transcode:processing')
    dead = r.llen('transcode:dead')
    done = TOTAL - pending - processing - dead

    # 从心跳里统计活跃 worker
    hosts = set()
    for k in r.scan_iter('task:hb:*'):
        hb = json.loads(r.get(k) or '{}')
        if hb.get('host'):
            hosts.add(hb['host'])

    elapsed = time.time() - START
    rate = done / elapsed * 3600           # 每小时处理数
    eta = pending / rate if rate else 0

    print(f'\r进度 {done}/{TOTAL} ({done/TOTAL*100:.1f}%) | '
          f'待处理 {pending} | 进行中 {processing} | 失败 {dead} | '
          f'worker {len(hosts)} | 速率 {rate:.0f}/h | '
          f'预计剩余 {eta/3600:.1f}h', end='')
    time.sleep(30)

进度 7421/9234 (80.4%) | 待处理 1683 | 进行中 24 | 失败 106 | worker 8 | 速率 121/h | 预计剩余 13.9h

# 提交时按预估时长降序入队（长任务先跑）
tasks.sort(key=lambda t: -t['estimated_duration'])

#!/bin/bash
# auto_shutdown.sh —— 队列空了就关掉自己（省最后一小时的钱）
while true; do
    pending=$(redis-cli -h queue-host llen transcode:queue)
    processing=$(redis-cli -h queue-host llen transcode:processing)
    if [ "$pending" -eq 0 ] && [ "$processing" -eq 0 ]; then
        echo "queue empty, shutting down"
        sudo shutdown -h now
    fi
    sleep 60
done