Python长时间运行FFmpeg子进程的事件驱动输出捕获问题

我之前在处理长期运行的FFmpeg转码任务时，也踩过PIPE输出积压导致内存溢出的坑，结合你提到的事件驱动、按需获取最新输出的需求，给你梳理几个实用的解决方案：

首先得明确问题根源：subprocess.PIPE是有缓冲区上限的，如果Python端不及时读取，子进程写满缓冲区会被阻塞，同时未读取的输出会积压在内存里导致溢出。我们需要实时消费输出，但又要避免存储全部历史内容，只保留最新的输出片段供按需获取。

你提到的LifoQueue（后进先出队列）非常适合这个场景——它天然适合存储最新内容，我们只需要限制队列的最大容量，让它永远只保留最新的N条输出（甚至只保留1条），这样内存占用就完全可控。

import subprocess
import threading
from queue import LifoQueue, Empty

def read_stream(stream, queue, max_queue_size=1):
    # 按行读取流，避免一次性读入大量数据
    for line in iter(stream.readline, b''):
        try:
            # 队列满时先弹出最旧的内容，再放入新内容
            if queue.full():
                queue.get_nowait()
            # 转码为字符串，忽略编码错误（FFmpeg可能输出非UTF-8字符）
            queue.put(line.decode('utf-8', errors='ignore'))
        except Empty:
            pass
    stream.close()

# 示例FFmpeg命令（根据你的实际需求调整）
ffmpeg_cmd = [
    'ffmpeg', '-i', 'input.mp4', '-c:v', 'libx264', '-crf', '23', 'output.mp4'
]
# 启动子进程：设置行缓冲，让FFmpeg输出一行就立刻flush
process = subprocess.Popen(
    ffmpeg_cmd,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    stdin=subprocess.DEVNULL,
    bufsize=1
)

# 创建LifoQueue，只保留最新1条输出（可根据需求调整maxsize）
output_queue = LifoQueue(maxsize=1)

# 启动线程读取stdout和stderr（注意：FFmpeg进度信息通常输出到stderr）
stdout_thread = threading.Thread(target=read_stream, args=(process.stdout, output_queue))
stderr_thread = threading.Thread(target=read_stream, args=(process.stderr, output_queue))
# 设置守护线程，主程序退出时自动结束线程
stdout_thread.daemon = True
stderr_thread.daemon = True
stdout_thread.start()
stderr_thread.start()

# 按需获取最新输出的函数
def get_latest_ffmpeg_output():
    try:
        # LifoQueue的get()会直接拿到最新的那条内容
        return output_queue.get_nowait()
    except Empty:
        return None

# 模拟事件驱动场景：比如每隔5秒主动获取一次最新状态
import time
while process.poll() is None:
    latest_output = get_latest_ffmpeg_output()
    if latest_output:
        print(f"FFmpeg最新输出: {latest_output.strip()}")
    time.sleep(5)

# 等待子进程结束
process.wait()

这个方案的优势：

• 线程实时读取输出，避免PIPE缓冲区满导致子进程阻塞
• LifoQueue限制容量，完全不会出现内存溢出问题
• 按需调用get_latest_ffmpeg_output()即可获取最新内容，符合事件驱动的按需解析需求

如果你的程序本身就基于事件循环（比如GUI、asyncio应用），可以用selectors模块实现非阻塞读取，不需要额外启动线程，更贴合事件驱动的设计。

import subprocess
import selectors

sel = selectors.DefaultSelector()

ffmpeg_cmd = [
    'ffmpeg', '-i', 'input.mp4', '-c:v', 'libx264', '-crf', '23', 'output.mp4'
]
process = subprocess.Popen(
    ffmpeg_cmd,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    stdin=subprocess.DEVNULL,
    bufsize=0  # 无缓冲，实时性最高
)

# 注册stdout和stderr到选择器，监听可读事件
sel.register(process.stdout, selectors.EVENT_READ)
sel.register(process.stderr, selectors.EVENT_READ)

latest_output = ""

# 事件循环（可以嵌入到你现有的主事件循环中）
while process.poll() is None:
    # 超时1秒，避免无限阻塞
    events = sel.select(timeout=1)
    for key, mask in events:
        stream = key.fileobj
        # 按行读取输出
        data = stream.readline()
        if data:
            # 更新最新输出内容
            latest_output = data.decode('utf-8', errors='ignore')
        else:
            # 流关闭，取消注册
            sel.unregister(stream)
            stream.close()

    # 此处可以根据你的事件触发逻辑，按需获取latest_output
    # 比如：当用户点击"查看状态"按钮时，打印latest_output
    # if user_triggered_event:
    #     print(latest_output.strip())

process.wait()

这个方案的优势：

• 无线程开销，完全基于事件驱动模型
• 内存中只保留最新的输出片段，内存占用极低
• 适配asyncio等事件循环框架

• FFmpeg的输出位置：FFmpeg的进度、日志信息通常输出到stderr而非stdout，所以不要只监听stdout
• 缓冲区设置：bufsize=1（行缓冲）适合文本输出，bufsize=0（无缓冲）实时性最高
• 编码错误处理：FFmpeg可能输出非UTF-8字符，解码时用errors='ignore'避免程序崩溃
• 禁用communicate()：绝对不要用subprocess.communicate()，它会等待子进程结束才读取全部输出，直接导致内存溢出

内容的提问来源于stack exchange，提问作者Robert Nagtegaal