FFmpeg处理管道视频流遇下载中断报错，如何实现缓冲暂停续处理？

你的核心问题是实时流直接喂FFmpeg时，短时间断流会导致FFmpeg因不完整NAL单元报错。下面从缓冲机制、FFmpeg容错配置、断流恢复三个方面给你实用方案：

一、本地缓冲队列 + 异步读写分离 #

直接实时写stdin的问题在于断流时FFmpeg立刻拿到不完整数据，通过双线程+缓冲队列可以解决：一个线程负责拉流存队列，另一个线程从队列取数据喂FFmpeg，队列空时暂停写，等数据恢复后继续。

修改后的代码示例 #

import subprocess
from functools import partial
from queue import Queue
import threading
import time

def stream_reader(stream, buffer_queue):
    # 独立线程拉取流数据，存入缓冲队列
    try:
        for chunk in iter(partial(stream.read, 1024), b''):
            # 队列满时阻塞，避免内存溢出
            buffer_queue.put(chunk)
    except Exception as e:
        print(f"拉流异常: {e}")
    finally:
        # 标记流结束
        buffer_queue.put(None)

def ffmpeg_writer(process, buffer_queue, min_buffer=128):
    # 从缓冲队列取数据喂FFmpeg，缓冲不足时等待
    temp_buffer = []
    while True:
        chunk = buffer_queue.get()
        if chunk is None:
            break
        
        temp_buffer.append(chunk)
        # 缓冲达到设定阈值（或队列空但有缓冲）时，批量写入FFmpeg
        if len(temp_buffer) >= min_buffer or (temp_buffer and buffer_queue.empty()):
            process.stdin.write(b''.join(temp_buffer))
            process.stdin.flush()
            temp_buffer = []
        
        # 队列空时短暂休眠，减少CPU空转
        if buffer_queue.empty():
            time.sleep(0.01)
    
    # 写入剩余缓冲数据
    if temp_buffer:
        process.stdin.write(b''.join(temp_buffer))
        process.stdin.flush()
    
    process.stdin.close()

# 初始化缓冲队列，最大缓冲5MB（按1024字节chunk计算）
buffer_queue = Queue(maxsize=5 * 1024)

# 启动带容错参数的FFmpeg进程
process = subprocess.Popen(
    (
        'ffmpeg -err_detect ignore_err -max_interleave_delta 1000000 '
        '-fflags +genpts+igndts -i pipe: -f flv -codec copy -listen 1 '
        '-flvflags no_duration_filesize http://127.0.0.1:8080 -re'
    ),
    stdin=subprocess.PIPE,
    # 可选：把stderr输出到文件调试，比如 stderr=open('ffmpeg.log', 'w')
    stderr=subprocess.DEVNULL
)

# 启动拉流和写FFmpeg的线程
reader_thread = threading.Thread(target=stream_reader, args=(stream, buffer_queue))
writer_thread = threading.Thread(target=ffmpeg_writer, args=(process, buffer_queue))

reader_thread.start()
writer_thread.start()

# 等待线程结束
reader_thread.join()
writer_thread.join()

process.wait()

二、FFmpeg容错参数配置 #

给FFmpeg加几个参数，提升对断流和不完整数据的容忍度：

• -err_detect ignore_err：忽略解码错误，继续处理后续数据
• -max_interleave_delta 1000000：增大交错缓存，适应流的中断延迟
• -fflags +genpts+igndts：自动生成缺失的PTS（时间戳），忽略无效的DTS
• 如果是H.264流，还可以加-h264_mp4toannexb（如果原始流是MP4封装的裸流）

这些参数能让FFmpeg在短时间断流时，尽量不崩溃，而是等待后续数据恢复。

三、断流后的FFmpeg重启与关键帧恢复 #

如果断流时间过长，缓冲耗尽导致FFmpeg崩溃，可以监听进程状态，自动重启FFmpeg，并确保新进程从关键帧开始接收数据：

1. 在拉流线程中记录最近的关键帧位置（NAL单元的开头是0x00000001或0x000001，关键帧的NAL类型是5）
2. 当检测到FFmpeg进程退出时，关闭旧进程，清空缓冲队列
3. 拉流恢复后，先跳过非关键帧数据，直到找到下一个关键帧，再启动新的FFmpeg进程并喂数据

注意事项 #

• min_buffer参数根据流码率调整：比如1Mbps的流，1秒约128KB，设为128就是1秒缓冲，可应对普通网络波动
• 队列maxsize不要设置过大，避免占用过多内存
• 可以把FFmpeg的stderr输出到文件，方便调试断流时的具体错误

内容的提问来源于stack exchange，提问作者Андрей