简介这份资源提供基于WebSocket协议实现语音收发的Python服务器与客户端代码面向具备一定Python网络编程基础、希望实践实时音频通信的开发者。内容围绕Flask-SocketIO搭建服务端事件处理、websocket-client构建客户端连接并涉及语音数据的编码传输与Kaldi-Gstreamer-Server语音识别思路可用于实时语音传输、在线识别等场景的原型验证。压缩包共3个文件包含2个py源码与1个bak备份文件整体约9KB体积轻量便于直接阅读与二次修改。目前已有3113人学习下载说明其在WebSocket语音通信方向具有一定参考价值。读者可从中获取服务端与客户端的完整通信骨架、消息回调与数据发送的写法示例并据此调整消息处理逻辑、增加认证机制或优化语音处理流程快速搭建可运行的语音收发demo。1. 从一次语音房调试说起这套 WebSocket 收发语音的 Python 代码到底能干什么去年帮一个做在线协作工具的小团队排查延迟问题他们的语音房在弱网下总是断断续续抓包一看音频数据居然走的是 HTTP 轮询每 200 毫秒发一次请求服务端光建连就快扛不住了。后来换成 WebSocket 长连接推流延迟直接从 800 毫秒掉到 120 毫秒以内。这件事让我意识到很多开发者对「WebSocket 传语音」这件事的理解还停留在「能连上就行」但真正落地时会发现音频采集、分帧、二进制传输、服务端转发、客户端播放每一步都有坑。这份 Python 代码实现的就是一套完整的 WebSocket 语音收发系统包含服务端和客户端两部分。服务端负责接收客户端发来的音频数据块再广播给同一房间的其他客户端客户端则负责采集麦克风音频、编码成可传输的格式、通过 WebSocket 发送同时接收远端音频并播放。它适合谁如果你正在做语音聊天室、在线会议、远程协作、游戏内语音或者单纯想搞明白 WebSocket 二进制传输怎么和音频处理结合这套代码就是一个能直接跑起来的起点。技术栈上服务端常见做法是用websockets库配合asyncio客户端用pyaudio采集播放音频编码走 PCM 或 Opus传输层就是 WebSocket 的二进制帧。2. 架构拆解音频从麦克风到扬声器要过几道手2.1 服务端与客户端的职责边界这套系统的核心设计思路是「服务端只做转发客户端负责采集和播放」。服务端不碰音频编解码也不做混音它只维护一个客户端连接池收到某个客户端发来的二进制音频帧后原样转发给同一房间的其他客户端。这样做的好处是服务端逻辑极简CPU 占用低扩展性强代价是客户端要承担编解码和网络抖动的处理。服务端用websockets.serve()启动每个连接进来后注册到一个全局的clients集合里。收到消息时判断类型如果是文本消息当作控制指令比如加入房间、离开房间如果是二进制消息直接遍历集合转发。这里有个细节转发时要跳过发送者自己否则会出现回声。客户端这边分两个线程一个线程负责从麦克风读数据并发送另一个线程负责接收远端数据并写入扬声器。两个线程通过 WebSocket 连接共享但读写操作要分开避免阻塞。常见做法是用asyncio的事件循环配合run_in_executor把阻塞的音频 IO 放到线程池里执行。2.2 音频采集与分帧的参数选择音频采集的参数直接决定了传输效率和音质。这份代码里采样率一般设 16000 Hz 或 44100 Hz声道数用单声道位深 16 bit。每帧的时长很关键太短会导致包太多网络开销大太长会增加延迟。我一般会设 20 毫秒一帧这样 16000 Hz 采样率下每帧就是 320 个采样点乘以 2 字节位深一帧数据 640 字节正好适合 WebSocket 传输。import pyaudio # 音频参数配置 CHUNK 320 # 每帧采样点数20ms 16000Hz FORMAT pyaudio.paInt16 # 16位深度 CHANNELS 1 # 单声道 RATE 16000 # 采样率 audio pyaudio.PyAudio() stream audio.open( formatFORMAT, channelsCHANNELS, rateRATE, inputTrue, frames_per_bufferCHUNK )这段代码初始化了 PyAudio 的输入流。CHUNK设为 320 是经过计算的不是随便填的。frames_per_buffer必须和CHUNK一致否则read()返回的数据长度会不稳定。FORMAT用paInt16是因为 16 位深度在音质和带宽之间平衡得最好8 位太糙32 位浮点又没必要。RATE选 16000 而不是 44100是因为语音场景下 16k 已经足够清晰带宽直接省一半。2.3 WebSocket 服务端的连接管理与转发逻辑服务端的核心就是一个异步的 handler 函数每个客户端连接进来都会跑一遍这个函数。里面要处理三件事连接建立时注册、收到消息时转发、连接断开时清理。import asyncio import websockets clients set() async def handler(websocket): # 新客户端接入加入集合 clients.add(websocket) try: async for message in websocket: # 二进制消息才转发文本消息可做控制指令 if isinstance(message, bytes): # 广播给其他客户端跳过发送者 await asyncio.gather( *[client.send(message) for client in clients if client ! websocket], return_exceptionsTrue ) finally: # 无论正常断开还是异常都要清理 clients.remove(websocket) async def main(): async with websockets.serve(handler, 0.0.0.0, 8765): await asyncio.Future() # 永久运行 asyncio.run(main())clients用set而不是list是因为集合的删除操作是 O(1)而且自动去重。asyncio.gather并发发送给所有客户端return_exceptionsTrue保证某个客户端发送失败不会影响其他人。finally块里的remove是必须的否则断开的连接会一直留在集合里下次广播时抛异常。这里有个容易翻车的地方如果客户端异常断开websocket对象可能已经不可用了但remove操作本身不会报错所以放在finally里最安全。2.4 客户端双线程模型与收发分离客户端要同时做两件事持续采集音频并发送持续接收音频并播放。单线程做不了因为stream.read()是阻塞的websocket.recv()也是阻塞的。常见做法是用两个线程一个跑发送循环一个跑接收循环。import threading import asyncio import websockets async def send_audio(websocket, stream): while True: # 阻塞读音频放到线程池执行避免卡事件循环 data await asyncio.get_event_loop().run_in_executor( None, stream.read, CHUNK, False ) await websocket.send(data) async def recv_audio(websocket, out_stream): async for message in websocket: if isinstance(message, bytes): # 直接写入扬声器缓冲区 out_stream.write(message) async def main(): async with websockets.connect(ws://localhost:8765) as websocket: # 初始化输入输出流... await asyncio.gather( send_audio(websocket, in_stream), recv_audio(websocket, out_stream) )run_in_executor把阻塞的stream.read丢到默认线程池这样asyncio的事件循环不会被卡住。send_audio和recv_audio用asyncio.gather并发跑任何一个抛异常都会取消另一个。stream.read的第二个参数False表示不抛溢出异常直接返回可用数据这在实时音频场景下比抛异常更合适因为偶尔丢一帧比整个程序崩掉强。3. 跑起来从零搭建环境到听到第一声回传3.1 依赖安装与版本确认这套代码依赖三个库websockets、pyaudio、asyncio标准库。pyaudio在 Windows 上可以直接pip install pyaudio但在 Linux 上需要先装portaudio19-devmacOS 上需要brew install portaudio。这是第一个容易卡住的地方很多人pip install报错就是因为系统级依赖没装。# Ubuntu/Debian 先装系统依赖 sudo apt-get install portaudio19-dev python3-pyaudio # 再装 Python 包 pip install pyaudio websockets # 确认版本 python -c import pyaudio; print(pyaudio.__version__) python -c import websockets; print(websockets.__version__)websockets库在 10.0 版本前后 API 有变化serve()的用法从websockets.serve(handler, ...)变成了websockets.serve(handler, ...)但 handler 的参数从(websocket, path)变成了(websocket)。如果你跑起来报TypeError: handler() takes 1 positional argument but 2 were given就是版本不匹配。我一般会锁定websockets11.0避免这个玄学问题。3.2 服务端启动与端口监听验证服务端代码写好后直接python server.py启动。启动后终端不会有太多输出这是正常的因为asyncio.Future()会一直挂着。要验证服务端是否真的在监听可以用netstat或ss命令看 8765 端口。# Linux/macOS 查看端口监听 ss -tlnp | grep 8765 # Windows 查看端口 netstat -an | findstr 8765如果端口没起来先检查是不是被防火墙拦了或者 8765 被其他程序占了。换端口的话服务端和客户端要同步改。另外服务端绑定0.0.0.0表示监听所有网卡如果只想本机测试改成127.0.0.1更安全。3.3 客户端连接与音频回环测试客户端启动后它会连上服务端然后开始采集和播放。但这时候如果你只开一个客户端说话是听不到回声的因为服务端会跳过发送者自己。要测试回环需要开两个客户端实例一个说话另一个就能听到。# 终端1启动服务端 python server.py # 终端2启动客户端A python client.py # 终端3启动客户端B python client.py两个客户端都连上后对着 A 的麦克风说话B 的扬声器应该能听到。如果听不到按这个顺序排查先看服务端有没有收到二进制消息加一行print(len(message))再看客户端 B 有没有收到消息最后看out_stream.write有没有被调用。这个链路里任何一环断了都会导致无声逐段打日志是最快的定位方式。3.4 延迟与音质的初步调优跑通之后你可能会觉得延迟有点大或者声音有杂音。延迟主要来自三个地方采集缓冲、网络传输、播放缓冲。采集这边CHUNK越小延迟越低但太小会导致包太碎播放这边out_stream的缓冲区大小也会影响延迟PyAudio 默认缓冲可能偏大可以在open()时指定output_buffer_size。out_stream audio.open( formatFORMAT, channelsCHANNELS, rateRATE, outputTrue, frames_per_bufferCHUNK, output_buffer_sizeCHUNK * 4 # 控制播放缓冲越小延迟越低 )output_buffer_size设成CHUNK * 4是我试过比较平衡的值再小容易爆音再大延迟明显。音质方面如果杂音多先确认采样率是否匹配客户端采集和播放的RATE必须一致否则会变调。另外如果网络抖动大可以考虑加一个简单的抖动缓冲区把收到的音频帧先存起来按固定节奏播放而不是收到就播。4. 避坑与排查那些让我熬夜的音频传输问题4.1 现象客户端连上就断服务端报 ConnectionClosedError原因通常是客户端和服务端的websockets版本不兼容或者服务端 handler 函数签名不对。旧版 handler 是(websocket, path)新版是(websocket)。解决方法是统一版本或者用websockets.serve(handler, ...)时确认 handler 只接收一个参数。我一般会在服务端加一个try/except把异常打出来不然ConnectionClosedError默认不显示堆栈很难定位。4.2 现象能连上但收不到音频服务端日志显示收到 0 字节原因是客户端stream.read()返回了空数据通常是麦克风没权限或者被其他程序占用。在 macOS 上终端需要麦克风权限在 Windows 上检查隐私设置里是否允许应用访问麦克风。另外如果CHUNK设得太大read()可能会阻塞很久才返回看起来像没数据。解决方法是先用arecord或系统录音机确认麦克风本身能工作再检查代码里的CHUNK和frames_per_buffer是否一致。4.3 现象听到的声音断断续续像机器人说话这是典型的缓冲区欠载或过载。欠载是播放缓冲区空了过载是数据积压太多。原因可能是网络抖动导致音频帧到达不均匀或者客户端发送速度和服务端转发速度不匹配。解决方法是加一个简单的抖动缓冲收到音频帧后不立即播放先存到一个队列里播放线程按固定间隔从队列取数据。队列长度可以动态调整网络差的时候多存几帧网络好的时候少存几帧。4.4 现象两个客户端互相听到回声或者自己听到自己回声的原因是服务端转发时没有跳过发送者。检查clients集合的遍历逻辑确保if client ! websocket这个条件生效。另一个可能是客户端同时开了采集和播放扬声器的声音被麦克风又采进去了这是物理回声需要戴耳机或者做回声消除。代码层面能解决的是前者后者得靠硬件或算法。4.5 现象程序跑一段时间后内存暴涨原因是clients集合里的死连接没被清理或者音频数据在某个队列里堆积。检查finally块里的remove是否执行以及有没有未消费的asyncio.Queue。另外asyncio.gather如果某个任务抛异常没被捕获可能会导致协程泄漏。我一般会在服务端加一个定时任务定期打印len(clients)观察连接数是否正常。5. 二次开发从能跑到好用我踩过的那些坑这套代码跑通之后离真正能用还有一段距离。我第一次拿它做语音房的时候以为改改端口就能上线结果被现实狠狠教育了一顿。下面是我在二次开发中总结的几个关键改造点按优先级从高到低排。第一加房间隔离。原始代码是所有客户端都在一个全局集合里一个人说话所有人都能听到。实际场景肯定要分房间。改造方法是在服务端维护一个dictkey 是房间 IDvalue 是该房间的客户端集合。客户端连接时先发一条文本消息声明房间号服务端解析后把连接放到对应房间的集合里。转发时只遍历同房间的集合。这里要注意客户端断开时要从房间集合里移除如果房间空了就把 key 也删掉不然dict会越来越大。rooms {} # {room_id: set(clients)} async def handler(websocket): room_id None try: async for message in websocket: if isinstance(message, str): # 文本消息当作加入房间指令 room_id message rooms.setdefault(room_id, set()).add(websocket) elif isinstance(message, bytes) and room_id: # 只转发给同房间的其他客户端 await asyncio.gather( *[c.send(message) for c in rooms[room_id] if c ! websocket], return_exceptionsTrue ) finally: if room_id and room_id in rooms: rooms[room_id].discard(websocket) if not rooms[room_id]: del rooms[room_id]第二加音频编码压缩。原始代码传的是裸 PCM16k 单声道 16 位下每秒 32KB10 个人同时说话就是 320KB/s带宽压力不小。实际项目里一般会换成 Opus 编码压缩率能到 10:1 甚至更高。Python 里可以用opuslib或者pyogg但要注意编码和解码要配对客户端编码、服务端透传、其他客户端解码。如果服务端要做混音那就得先解码再混音再编码复杂度会高很多。第三加心跳和重连。WebSocket 长连接在移动网络下很容易断尤其是切 WiFi 和 4G 的时候。客户端需要加心跳包每隔几秒发一个 ping服务端回 pong。如果连续几次没收到 pong就主动重连。websockets库自带 ping/pong 机制可以通过ping_interval和ping_timeout参数配置。重连逻辑要放在客户端外层断线后指数退避重试避免雪崩。第四加权限和鉴权。原始代码谁都能连实际项目肯定要验证身份。常见做法是客户端连接时在 URL 里带 token服务端在 handler 里解析并验证。验证不通过直接close()。token 可以用 JWT服务端验签后拿到用户 ID再决定能不能加入某个房间。第五加日志和监控。音频传输出问题时没有日志就是黑匣子。我一般会在服务端记录每个连接的加入、离开、收发字节数客户端记录采集帧数、发送帧数、接收帧数、播放帧数。这些计数器定期打点到监控系统一旦发现发送和接收差距变大就知道中间有丢包或阻塞。改造完之后验证方法也很重要。我习惯用三个指标来判断系统是否健康端到端延迟从说话到对方听到的时间、丢帧率发送帧数减接收帧数除以发送帧数、CPU 占用服务端和客户端分别看。延迟控制在 200ms 以内、丢帧率低于 1%、CPU 占用低于 30%基本就算能用了。从那以后我每次做实时音频项目都会先把这套 WebSocket 收发框架跑一遍确认基础链路没问题再往上叠业务逻辑。这样即使后面出问题也能快速定位是传输层还是业务层。希望这套代码和上面的经验能帮到你少走一些我当年走过的弯路。本文还有配套的精品资源点击获取 
长河君 · 2026/10/19 13:50:55 · 阅读 1.2k Python WebSocket 语音收发实战：从架构到避坑
简介这份资源提供基于WebSocket协议实现语音收发的Python服务器与客户端代码面向具备一定Python网络编程基础、希望实践实时音频通信的开发者。内容围绕Flask-SocketIO搭建服务端事件处理、websocket-client构建客户端连接并涉及语音数据的编码传输与Kaldi-Gstreamer-Server语音识别思路可用于实时语音传输、在线识别等场景的原型验证。压缩包共3个文件包含2个py源码与1个bak备份文件整体约9KB体积轻量便于直接阅读与二次修改。目前已有3113人学习下载说明其在WebSocket语音通信方向具有一定参考价值。读者可从中获取服务端与客户端的完整通信骨架、消息回调与数据发送的写法示例并据此调整消息处理逻辑、增加认证机制或优化语音处理流程快速搭建可运行的语音收发demo。1. 从一次语音房调试说起这套 WebSocket 收发语音的 Python 代码到底能干什么去年帮一个做在线协作工具的小团队排查延迟问题他们的语音房在弱网下总是断断续续抓包一看音频数据居然走的是 HTTP 轮询每 200 毫秒发一次请求服务端光建连就快扛不住了。后来换成 WebSocket 长连接推流延迟直接从 800 毫秒掉到 120 毫秒以内。这件事让我意识到很多开发者对「WebSocket 传语音」这件事的理解还停留在「能连上就行」但真正落地时会发现音频采集、分帧、二进制传输、服务端转发、客户端播放每一步都有坑。这份 Python 代码实现的就是一套完整的 WebSocket 语音收发系统包含服务端和客户端两部分。服务端负责接收客户端发来的音频数据块再广播给同一房间的其他客户端客户端则负责采集麦克风音频、编码成可传输的格式、通过 WebSocket 发送同时接收远端音频并播放。它适合谁如果你正在做语音聊天室、在线会议、远程协作、游戏内语音或者单纯想搞明白 WebSocket 二进制传输怎么和音频处理结合这套代码就是一个能直接跑起来的起点。技术栈上服务端常见做法是用websockets库配合asyncio客户端用pyaudio采集播放音频编码走 PCM 或 Opus传输层就是 WebSocket 的二进制帧。2. 架构拆解音频从麦克风到扬声器要过几道手2.1 服务端与客户端的职责边界这套系统的核心设计思路是「服务端只做转发客户端负责采集和播放」。服务端不碰音频编解码也不做混音它只维护一个客户端连接池收到某个客户端发来的二进制音频帧后原样转发给同一房间的其他客户端。这样做的好处是服务端逻辑极简CPU 占用低扩展性强代价是客户端要承担编解码和网络抖动的处理。服务端用websockets.serve()启动每个连接进来后注册到一个全局的clients集合里。收到消息时判断类型如果是文本消息当作控制指令比如加入房间、离开房间如果是二进制消息直接遍历集合转发。这里有个细节转发时要跳过发送者自己否则会出现回声。客户端这边分两个线程一个线程负责从麦克风读数据并发送另一个线程负责接收远端数据并写入扬声器。两个线程通过 WebSocket 连接共享但读写操作要分开避免阻塞。常见做法是用asyncio的事件循环配合run_in_executor把阻塞的音频 IO 放到线程池里执行。2.2 音频采集与分帧的参数选择音频采集的参数直接决定了传输效率和音质。这份代码里采样率一般设 16000 Hz 或 44100 Hz声道数用单声道位深 16 bit。每帧的时长很关键太短会导致包太多网络开销大太长会增加延迟。我一般会设 20 毫秒一帧这样 16000 Hz 采样率下每帧就是 320 个采样点乘以 2 字节位深一帧数据 640 字节正好适合 WebSocket 传输。import pyaudio # 音频参数配置 CHUNK 320 # 每帧采样点数20ms 16000Hz FORMAT pyaudio.paInt16 # 16位深度 CHANNELS 1 # 单声道 RATE 16000 # 采样率 audio pyaudio.PyAudio() stream audio.open( formatFORMAT, channelsCHANNELS, rateRATE, inputTrue, frames_per_bufferCHUNK )这段代码初始化了 PyAudio 的输入流。CHUNK设为 320 是经过计算的不是随便填的。frames_per_buffer必须和CHUNK一致否则read()返回的数据长度会不稳定。FORMAT用paInt16是因为 16 位深度在音质和带宽之间平衡得最好8 位太糙32 位浮点又没必要。RATE选 16000 而不是 44100是因为语音场景下 16k 已经足够清晰带宽直接省一半。2.3 WebSocket 服务端的连接管理与转发逻辑服务端的核心就是一个异步的 handler 函数每个客户端连接进来都会跑一遍这个函数。里面要处理三件事连接建立时注册、收到消息时转发、连接断开时清理。import asyncio import websockets clients set() async def handler(websocket): # 新客户端接入加入集合 clients.add(websocket) try: async for message in websocket: # 二进制消息才转发文本消息可做控制指令 if isinstance(message, bytes): # 广播给其他客户端跳过发送者 await asyncio.gather( *[client.send(message) for client in clients if client ! websocket], return_exceptionsTrue ) finally: # 无论正常断开还是异常都要清理 clients.remove(websocket) async def main(): async with websockets.serve(handler, 0.0.0.0, 8765): await asyncio.Future() # 永久运行 asyncio.run(main())clients用set而不是list是因为集合的删除操作是 O(1)而且自动去重。asyncio.gather并发发送给所有客户端return_exceptionsTrue保证某个客户端发送失败不会影响其他人。finally块里的remove是必须的否则断开的连接会一直留在集合里下次广播时抛异常。这里有个容易翻车的地方如果客户端异常断开websocket对象可能已经不可用了但remove操作本身不会报错所以放在finally里最安全。2.4 客户端双线程模型与收发分离客户端要同时做两件事持续采集音频并发送持续接收音频并播放。单线程做不了因为stream.read()是阻塞的websocket.recv()也是阻塞的。常见做法是用两个线程一个跑发送循环一个跑接收循环。import threading import asyncio import websockets async def send_audio(websocket, stream): while True: # 阻塞读音频放到线程池执行避免卡事件循环 data await asyncio.get_event_loop().run_in_executor( None, stream.read, CHUNK, False ) await websocket.send(data) async def recv_audio(websocket, out_stream): async for message in websocket: if isinstance(message, bytes): # 直接写入扬声器缓冲区 out_stream.write(message) async def main(): async with websockets.connect(ws://localhost:8765) as websocket: # 初始化输入输出流... await asyncio.gather( send_audio(websocket, in_stream), recv_audio(websocket, out_stream) )run_in_executor把阻塞的stream.read丢到默认线程池这样asyncio的事件循环不会被卡住。send_audio和recv_audio用asyncio.gather并发跑任何一个抛异常都会取消另一个。stream.read的第二个参数False表示不抛溢出异常直接返回可用数据这在实时音频场景下比抛异常更合适因为偶尔丢一帧比整个程序崩掉强。3. 跑起来从零搭建环境到听到第一声回传3.1 依赖安装与版本确认这套代码依赖三个库websockets、pyaudio、asyncio标准库。pyaudio在 Windows 上可以直接pip install pyaudio但在 Linux 上需要先装portaudio19-devmacOS 上需要brew install portaudio。这是第一个容易卡住的地方很多人pip install报错就是因为系统级依赖没装。# Ubuntu/Debian 先装系统依赖 sudo apt-get install portaudio19-dev python3-pyaudio # 再装 Python 包 pip install pyaudio websockets # 确认版本 python -c import pyaudio; print(pyaudio.__version__) python -c import websockets; print(websockets.__version__)websockets库在 10.0 版本前后 API 有变化serve()的用法从websockets.serve(handler, ...)变成了websockets.serve(handler, ...)但 handler 的参数从(websocket, path)变成了(websocket)。如果你跑起来报TypeError: handler() takes 1 positional argument but 2 were given就是版本不匹配。我一般会锁定websockets11.0避免这个玄学问题。3.2 服务端启动与端口监听验证服务端代码写好后直接python server.py启动。启动后终端不会有太多输出这是正常的因为asyncio.Future()会一直挂着。要验证服务端是否真的在监听可以用netstat或ss命令看 8765 端口。# Linux/macOS 查看端口监听 ss -tlnp | grep 8765 # Windows 查看端口 netstat -an | findstr 8765如果端口没起来先检查是不是被防火墙拦了或者 8765 被其他程序占了。换端口的话服务端和客户端要同步改。另外服务端绑定0.0.0.0表示监听所有网卡如果只想本机测试改成127.0.0.1更安全。3.3 客户端连接与音频回环测试客户端启动后它会连上服务端然后开始采集和播放。但这时候如果你只开一个客户端说话是听不到回声的因为服务端会跳过发送者自己。要测试回环需要开两个客户端实例一个说话另一个就能听到。# 终端1启动服务端 python server.py # 终端2启动客户端A python client.py # 终端3启动客户端B python client.py两个客户端都连上后对着 A 的麦克风说话B 的扬声器应该能听到。如果听不到按这个顺序排查先看服务端有没有收到二进制消息加一行print(len(message))再看客户端 B 有没有收到消息最后看out_stream.write有没有被调用。这个链路里任何一环断了都会导致无声逐段打日志是最快的定位方式。3.4 延迟与音质的初步调优跑通之后你可能会觉得延迟有点大或者声音有杂音。延迟主要来自三个地方采集缓冲、网络传输、播放缓冲。采集这边CHUNK越小延迟越低但太小会导致包太碎播放这边out_stream的缓冲区大小也会影响延迟PyAudio 默认缓冲可能偏大可以在open()时指定output_buffer_size。out_stream audio.open( formatFORMAT, channelsCHANNELS, rateRATE, outputTrue, frames_per_bufferCHUNK, output_buffer_sizeCHUNK * 4 # 控制播放缓冲越小延迟越低 )output_buffer_size设成CHUNK * 4是我试过比较平衡的值再小容易爆音再大延迟明显。音质方面如果杂音多先确认采样率是否匹配客户端采集和播放的RATE必须一致否则会变调。另外如果网络抖动大可以考虑加一个简单的抖动缓冲区把收到的音频帧先存起来按固定节奏播放而不是收到就播。4. 避坑与排查那些让我熬夜的音频传输问题4.1 现象客户端连上就断服务端报 ConnectionClosedError原因通常是客户端和服务端的websockets版本不兼容或者服务端 handler 函数签名不对。旧版 handler 是(websocket, path)新版是(websocket)。解决方法是统一版本或者用websockets.serve(handler, ...)时确认 handler 只接收一个参数。我一般会在服务端加一个try/except把异常打出来不然ConnectionClosedError默认不显示堆栈很难定位。4.2 现象能连上但收不到音频服务端日志显示收到 0 字节原因是客户端stream.read()返回了空数据通常是麦克风没权限或者被其他程序占用。在 macOS 上终端需要麦克风权限在 Windows 上检查隐私设置里是否允许应用访问麦克风。另外如果CHUNK设得太大read()可能会阻塞很久才返回看起来像没数据。解决方法是先用arecord或系统录音机确认麦克风本身能工作再检查代码里的CHUNK和frames_per_buffer是否一致。4.3 现象听到的声音断断续续像机器人说话这是典型的缓冲区欠载或过载。欠载是播放缓冲区空了过载是数据积压太多。原因可能是网络抖动导致音频帧到达不均匀或者客户端发送速度和服务端转发速度不匹配。解决方法是加一个简单的抖动缓冲收到音频帧后不立即播放先存到一个队列里播放线程按固定间隔从队列取数据。队列长度可以动态调整网络差的时候多存几帧网络好的时候少存几帧。4.4 现象两个客户端互相听到回声或者自己听到自己回声的原因是服务端转发时没有跳过发送者。检查clients集合的遍历逻辑确保if client ! websocket这个条件生效。另一个可能是客户端同时开了采集和播放扬声器的声音被麦克风又采进去了这是物理回声需要戴耳机或者做回声消除。代码层面能解决的是前者后者得靠硬件或算法。4.5 现象程序跑一段时间后内存暴涨原因是clients集合里的死连接没被清理或者音频数据在某个队列里堆积。检查finally块里的remove是否执行以及有没有未消费的asyncio.Queue。另外asyncio.gather如果某个任务抛异常没被捕获可能会导致协程泄漏。我一般会在服务端加一个定时任务定期打印len(clients)观察连接数是否正常。5. 二次开发从能跑到好用我踩过的那些坑这套代码跑通之后离真正能用还有一段距离。我第一次拿它做语音房的时候以为改改端口就能上线结果被现实狠狠教育了一顿。下面是我在二次开发中总结的几个关键改造点按优先级从高到低排。第一加房间隔离。原始代码是所有客户端都在一个全局集合里一个人说话所有人都能听到。实际场景肯定要分房间。改造方法是在服务端维护一个dictkey 是房间 IDvalue 是该房间的客户端集合。客户端连接时先发一条文本消息声明房间号服务端解析后把连接放到对应房间的集合里。转发时只遍历同房间的集合。这里要注意客户端断开时要从房间集合里移除如果房间空了就把 key 也删掉不然dict会越来越大。rooms {} # {room_id: set(clients)} async def handler(websocket): room_id None try: async for message in websocket: if isinstance(message, str): # 文本消息当作加入房间指令 room_id message rooms.setdefault(room_id, set()).add(websocket) elif isinstance(message, bytes) and room_id: # 只转发给同房间的其他客户端 await asyncio.gather( *[c.send(message) for c in rooms[room_id] if c ! websocket], return_exceptionsTrue ) finally: if room_id and room_id in rooms: rooms[room_id].discard(websocket) if not rooms[room_id]: del rooms[room_id]第二加音频编码压缩。原始代码传的是裸 PCM16k 单声道 16 位下每秒 32KB10 个人同时说话就是 320KB/s带宽压力不小。实际项目里一般会换成 Opus 编码压缩率能到 10:1 甚至更高。Python 里可以用opuslib或者pyogg但要注意编码和解码要配对客户端编码、服务端透传、其他客户端解码。如果服务端要做混音那就得先解码再混音再编码复杂度会高很多。第三加心跳和重连。WebSocket 长连接在移动网络下很容易断尤其是切 WiFi 和 4G 的时候。客户端需要加心跳包每隔几秒发一个 ping服务端回 pong。如果连续几次没收到 pong就主动重连。websockets库自带 ping/pong 机制可以通过ping_interval和ping_timeout参数配置。重连逻辑要放在客户端外层断线后指数退避重试避免雪崩。第四加权限和鉴权。原始代码谁都能连实际项目肯定要验证身份。常见做法是客户端连接时在 URL 里带 token服务端在 handler 里解析并验证。验证不通过直接close()。token 可以用 JWT服务端验签后拿到用户 ID再决定能不能加入某个房间。第五加日志和监控。音频传输出问题时没有日志就是黑匣子。我一般会在服务端记录每个连接的加入、离开、收发字节数客户端记录采集帧数、发送帧数、接收帧数、播放帧数。这些计数器定期打点到监控系统一旦发现发送和接收差距变大就知道中间有丢包或阻塞。改造完之后验证方法也很重要。我习惯用三个指标来判断系统是否健康端到端延迟从说话到对方听到的时间、丢帧率发送帧数减接收帧数除以发送帧数、CPU 占用服务端和客户端分别看。延迟控制在 200ms 以内、丢帧率低于 1%、CPU 占用低于 30%基本就算能用了。从那以后我每次做实时音频项目都会先把这套 WebSocket 收发框架跑一遍确认基础链路没问题再往上叠业务逻辑。这样即使后面出问题也能快速定位是传输层还是业务层。希望这套代码和上面的经验能帮到你少走一些我当年走过的弯路。本文还有配套的精品资源点击获取