WebSocket流式音频FFmpeg处理：首次成功后报Invalid frame size错误求助

问题背景 #

用Python构建基于WebSocket的流式音频处理应用，服务器端通过FFmpeg处理音频字节数据后传入机器学习算法。首次调用FFmpeg处理正常，但第二次调用时出现以下错误：

[mp3 @ 0x1208051e0] Invalid frame size (352): Could not seek to 363.
[in#0 @ 0x600002214200] Error opening input: Invalid argument
Error opening input file pipe:0.

原处理代码：

async def ffmpeg_read(bpayload: bytes, sampling_rate: int = 16000) -> np.array:
    ar = f"{sampling_rate}"
    ac = "1"
    format_for_conversion = "f32le"
    ffmpeg_command = [
        "ffmpeg",
        "-i", "pipe:0",
        "-ac", ac,
        "-acodec", f"pcm_{format_for_conversion}",
        "-ar", ar,
        "-f", format_for_conversion,
        "pipe:1"]
    try:
        process = await asyncio.create_subprocess_exec(
            *ffmpeg_command,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE)

        process.stdin.write(bpayload)
        await process.stdin.drain() 
        process.stdin.close()

        out_bytes = await process.stdout.read(8000)  # Read asynchronously
        audio = np.frombuffer(out_bytes, np.float32)

        if audio.shape[0] == 0:
            raise ValueError("Malformed soundfile")
        return audio

    except FileNotFoundError:
        raise ValueError(
            "ffmpeg was not found but is required to load audio files from filename")

问题原因 #

1. 进程复用缺失：原代码每次调用都新建FFmpeg进程，处理完就关闭stdin并让进程退出。对于流式音频的分块数据（如MP3帧），单块数据可能不是完整的音频帧，FFmpeg无法独立解析这类不完整数据。
2. 帧结构解析失败：第二次处理时，FFmpeg尝试解析不完整的音频帧（或残留的不完整数据），导致帧大小校验失败，触发输入错误。

解决方案 #

核心思路是复用FFmpeg进程，保持stdin/stdout的长连接，让FFmpeg持续处理流式输入，而非每次新建进程。同时添加线程安全控制和进程异常重启机制。

修改后的代码 #

import asyncio
import numpy as np

class FFmpegAudioProcessor:
    def __init__(self, sampling_rate: int = 16000):
        self.sampling_rate = sampling_rate
        self.channels = "1"
        self.target_format = "f32le"
        self.process = None
        self._lock = asyncio.Lock()

    async def _init_process(self):
        """初始化FFmpeg长连接进程"""
        ffmpeg_cmd = [
            "ffmpeg",
            "-i", "pipe:0",
            "-ac", self.channels,
            "-acodec", f"pcm_{self.target_format}",
            "-ar", str(self.sampling_rate),
            "-f", self.target_format,
            "-loglevel", "error",  # 关闭冗余日志，如需调试可改为info
            "pipe:1"
        ]
        self.process = await asyncio.create_subprocess_exec(
            *ffmpeg_cmd,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        # 后台读取stderr，避免缓冲区满导致进程阻塞
        asyncio.create_task(self._read_stderr())

    async def _read_stderr(self):
        """后台处理FFmpeg的错误输出"""
        if not self.process:
            return
        while True:
            line = await self.process.stderr.readline()
            if not line:
                break
            # 可根据需求将日志写入文件或监控系统
            print(f"FFmpeg Error: {line.decode().strip()}")

    async def process_audio(self, payload: bytes) -> np.array:
        """处理单块音频字节数据"""
        # 进程异常退出时自动重启
        if not self.process or self.process.returncode is not None:
            await self._init_process()

        async with self._lock:
            # 写入音频数据到FFmpeg标准输入
            self.process.stdin.write(payload)
            await self.process.stdin.drain()

            # 读取处理后的PCM数据（8000字节对应2000个float32采样点，可根据需求调整）
            output_bytes = await self.process.stdout.read(8000)
            audio_array = np.frombuffer(output_bytes, np.float32)

            if len(audio_array) == 0:
                raise ValueError("处理后无有效音频数据")
            return audio_array

    async def shutdown(self):
        """关闭FFmpeg进程"""
        if self.process:
            self.process.stdin.close()
            await self.process.wait()

使用示例 #

# 初始化处理器
processor = FFmpegAudioProcessor(sampling_rate=16000)
await processor._init_process()

# 处理WebSocket传来的音频数据
async def handle_websocket_message(message):
    audio_data = await processor.process_audio(message.payload)
    # 将audio_data传入机器学习算法处理

关键改动说明 #

• 进程复用：FFmpeg进程仅初始化一次，持续处理流式输入，避免每次创建进程的开销和帧解析问题。
• 线程安全：用asyncio.Lock保证多WebSocket连接并发处理时的输入输出安全。
• 异常处理：检测进程状态，异常退出时自动重启，提升稳定性。
• stderr处理：后台读取错误日志，防止缓冲区满导致进程阻塞。

额外建议 #

• 如果输入是MP3等带帧结构的编码格式，可在FFmpeg命令中添加-fflags +genpts参数，帮助FFmpeg更好地处理不完整的流式数据。
• 调整read()的字节数，确保与WebSocket传输的音频块时长匹配（如16000采样率下，1秒音频对应64000字节float32数据）。

内容的提问来源于stack exchange，提问作者Nimrod Sadeh