FFmpeg等待管道关闭才处理输入 如何配置实现实时转码推流

FFmpeg需要等stdin管道关闭才启动处理，完全是配置错误导致的，核心诱因有3个：

• 输入探测阈值设置过大：你配置的-analyzeduration 2147483647单位为微秒，换算后接近36分钟，-probesize 2147483647单位为字节，换算后为2GB。FFmpeg启动后会先读取输入做格式探测，默认要攒够设置的探测数据量、或者等到输入EOF（即管道关闭）才会结束探测进入转码流程，两个值设到GB级、几十分钟级，自然会一直攒数据直到管道关闭才开始工作。
• 多层缓存未主动刷新：Python侧subprocess管道、FFmpeg内部IO层默认都会攒够一定数据量才向下一层传递，没有配置即时刷缓存规则的话，写入的数据会卡在缓存层，到不了FFmpeg的处理逻辑。
• 编码封装层默认攒帧：libx264默认会缓存数十帧做画质、码率优化，FLV封装也会攒包到固定大小才向外发送，没有开低延迟配置的话，就算数据送到处理层，也会攒一段才输出。

完全可以实现收到数据就立刻处理、实时输出推流的效果，需要从FFmpeg参数、代码逻辑两侧同步调整：

FFmpeg参数调整规则 #

• 缩小输入探测阈值：针对MJPEG格式的管道输入，设置100ms探测时长、100KB探测数据量足够完成格式识别，不需要配置GB级的探测规则。
• 开启输入低延迟模式：给输入加nobuffer、low_delay标记，关闭FFmpeg内部的输入缓存。
• 关闭编码层攒帧逻辑：libx264编码添加-tune zerolatency参数，禁用所有需要缓存多帧的优化项，实现编码完一帧就立刻输出。
• 开启输出即时刷新：输出侧加-flush_packets 1，每处理完一个音视频包就立刻刷缓存推到RTMP服务器，不要攒包。
• 修正-re参数位置：该参数是输入选项，要放在对应管道输入的前面，作用是按设定帧率控制输入速度，避免一开始就把所有帧一次性塞给FFmpeg。

Python代码调整规则 #

• 每次向stdin写完一帧数据后，必须调用p.stdin.flush()强制刷新Python进程内的管道缓存，否则数据会卡在Python侧缓存里，根本没发送给FFmpeg。
• 不需要读取FFmpeg输出时，不要把stdout设为subprocess.PIPE，否则FFmpeg输出的日志、运行信息会把管道缓存写满，直接导致进程卡住，直接重定向到subprocess.DEVNULL即可，需要调试时可以重定向到本地日志文件。
• 删掉循环里不合理的time.sleep(2)，该逻辑会导致每读一帧就卡2秒，完全不符合实时推流要求，要按设定帧率调整间隔，比如15fps的话每帧间隔约66ms。

import subprocess
import cv2
import os
import time

p = subprocess.Popen(
    [
        'ffmpeg',
        '-y',
        # 全局低延迟配置
        '-fflags', 'nobuffer',
        '-flags', 'low_delay',
        # 第一个输入：MJPEG图片管道，-re放在输入前控制读取速率
        '-re',
        '-f', 'image2pipe',
        '-vcodec', 'mjpeg',
        '-framerate', '15',
        # 缩小探测阈值，避免长时间攒数据
        '-analyzeduration', '100000',
        '-probesize', '102400',
        '-i', '-',
        # 第二个输入：本地音频文件
        '-vn', '-i', audio_path,
        # 视频编码配置：零延迟模式
        '-c:v', 'libx264',
        '-pix_fmt', 'yuv420p',
        '-preset', 'ultrafast',
        '-tune', 'zerolatency',
        # 音频编码配置
        '-c:a', 'aac',
        # 输出配置：每包立即刷新推送
        '-flush_packets', '1',
        '-f', 'flv',
        rtmp_url
    ],
    stdin=subprocess.PIPE,
    stdout=subprocess.DEVNULL, # 不需要读输出直接丢空，避免缓存阻塞
    stderr=subprocess.DEVNULL # 调试时可替换为日志文件路径
)

i = 0
while True:
    current_images = [name for name in os.listdir(img_dir) if os.path.isfile(os.path.join(img_dir, name))]
    current_images = sorted(current_images)
    if i > fnum-1:
        print("frame read failed")
        break
    try:
        img = cv2.imread(os.path.join(img_dir, current_images[i]))
        success, buffer = cv2.imencode('.jpg', img)
        frame = buffer.tobytes()
        p.stdin.write(frame)
        p.stdin.flush() # 关键：每次写完帧强制刷管道缓存
        time.sleep(1/15) # 按15fps控制帧间隔
        i += 1
    except Exception:
        time.sleep(0.1)
        pass

p.stdin.close()
p.wait()

内容的提问来源于stack exchange，提问作者Mahmoud Ferig