WebRTC+Aiortc音频流故障：杂音与MediaStreamError报错排查

问题描述 #

通过JavaScript WebRTC将音频流推送到Python aiortc服务器，测试播放音频时仅能听到杂音，几秒后程序崩溃并抛出MediaStreamError异常：

future: <Task finished name='Task-13' coro=<websocket_handler.<locals>.on_track.<locals>.process_audio() done, defined at ~/audio_webrtc/stream.py:52> exception=MediaStreamError()> Traceback (most recent call last): File "~/audio_webrtc/stream.py", line 55, in process_audio frame = await track.recv() File "~/venv/lib/python3.10/site-packages/aiortc/rtcrtpreceiver.py", line 206, in recv raise MediaStreamError aiortc.mediastreams.MediaStreamError

错误原因分析 #

1. 音频格式不匹配：WebRTC默认采集48000Hz采样率的音频，而服务器端PyAudio播放流设置为16000Hz，格式差异导致杂音。
2. 未处理Track关闭异常：process_audio协程未捕获MediaStreamError，音频Track关闭时track.recv()直接抛出异常，导致程序崩溃。
3. 阻塞事件循环：PyAudio的stream.write()是同步阻塞方法，在asyncio协程中调用会阻塞事件循环，引发资源处理异常。
4. 资源泄漏：服务器端peer_connections字典未在WebSocket关闭时移除过期连接，导致资源无法释放。

修复方案 #

1. 统一客户端与服务器音频格式 #

客户端采集音频时指定与服务器一致的采样率、声道参数。

2. 捕获Track异常并清理资源 #

在音频处理协程中添加异常捕获，处理MediaStreamError并停止播放流。

3. 异步处理音频写入 #

将PyAudio的同步写入操作放到线程中执行，避免阻塞asyncio事件循环。

4. 清理PeerConnection资源 #

WebSocket关闭时，从字典中移除对应连接，释放资源。

修改后的代码 #

client.js（关键修改） #

在getUserMedia中指定音频参数：

mediaStream = await navigator.mediaDevices.getUserMedia({ 
  audio: {
    sampleRate: 16000,
    channelCount: 1,
    echoCancellation: false
  }, 
  video: false 
});

server.py（关键修改） #

import pyaudio
import asyncio
import numpy as np
from aiortc import RTCPeerConnection, RTCSessionDescription, RTCIceCandidate
from aiortc.sdp import candidate_from_sdp
from aiohttp import web, WSMsgType
import json
import uuid
from pathlib import Path
routes = web.RouteTableDef()

peer_connections = {}  # dictionary to store RTCPeerConnection instances

# Audio parameters
CHANNELS = 1
RATE = 16000

# Create PyAudio instance
audio = pyaudio.PyAudio()

@routes.get('/ws')
async def websocket_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    pc_id = None  # this will store the PeerConnection ID
    audio_stream = None

    async for msg in ws:
        if msg.type == WSMsgType.TEXT:
            data = json.loads(msg.data)
            message_type = data.get("message_type")
            content = data.get("content")
            pc_id = data.get("id")

            pc = peer_connections.get(pc_id)
            if pc is None:
                # create a new PeerConnection if it does not exist
                pc = RTCPeerConnection()
                peer_connections[pc_id] = pc

                @pc.on("track")
                def on_track(track):
                    nonlocal audio_stream
                    if track.kind == "audio":
                        print("Track %s received" % track.kind)
                        # 创建音频播放流
                        audio_stream = audio.open(format=pyaudio.paInt16,
                                                channels=CHANNELS,
                                                rate=RATE,
                                                output=True)

                        async def process_audio():
                            try:
                                while True:
                                    frame = await track.recv()
                                    pcm_data = frame.to_ndarray().tobytes()
                                    # 异步执行音频写入，避免阻塞事件循环
                                    await asyncio.to_thread(audio_stream.write, pcm_data)
                            except Exception as e:
                                print("Error processing audio:", str(e))
                            finally:
                                if audio_stream:
                                    audio_stream.stop_stream()
                                    audio_stream.close()
                                    print("Audio stream closed")

                        asyncio.create_task(process_audio())
                    else:
                        print("Ignoring track of kind:", track.kind)

            if message_type == "SDP":
                sdp = RTCSessionDescription(**content)
                await pc.setRemoteDescription(sdp)
                answer = await pc.createAnswer()
                await pc.setLocalDescription(answer)
                await ws.send_json({
                    "message_type": "SDP",
                    "content": {
                        "type": "answer",
                        "sdp": pc.localDescription.sdp
                    },
                })
            elif message_type == "CANDIDATE":
                candidate_sdp = content['candidate']
                candidate_obj = candidate_from_sdp(candidate_sdp)
                candidate_obj.sdpMid = content['sdpMid']
                candidate_obj.sdpMLineIndex = content['sdpMLineIndex']
                candidate = RTCIceCandidate(**candidate_obj.__dict__)
                await pc.addIceCandidate(candidate)

        elif msg.type == WSMsgType.CLOSE:
            pc = peer_connections.get(pc_id)
            if pc:
                await pc.close()
                del peer_connections[pc_id]  # 移除过期连接
            if audio_stream:
                audio_stream.stop_stream()
                audio_stream.close()
            break

    return ws


app = web.Application()
app.router.add_routes(routes)
app.router.add_static('/static/', path=str(Path(__file__).parent / 'static'), name='static')
web.run_app(app)

额外注意事项 #

• 若部分浏览器不支持16000Hz采样率，可添加 fallback 逻辑，使用服务器端格式转换工具（如pydub）处理音频。
• 测试时关闭浏览器回声消除功能，避免影响音频质量。

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