简介这份代码包围绕 WebSocket 协议实现语音双向收发面向已具备 Python 基础、希望快速搭建实时音频通信原型的开发者与学习者。包内共 3 个文件以 2 个 Python 脚本为主体另含 1 个 bak 备份文件压缩包约 9KB体量轻巧便于直接阅读、运行与二次改造。已有 3117 人学习或下载。服务端侧重维持长连接并接收语音消息客户端负责建立连接、发送数据与回调处理代码涉及消息事件注册、JSON 数据封装、异常与关闭处理等关键环节读者可据此理解 WebSocket 语音收发的整体链路对照服务端与客户端的分工快速验证音频数据的上行下行流程也为后续扩展预留空间可进一步调整消息逻辑、补充认证机制、优化音频处理或对接 Kaldi-Gstreamer-Server 等语音识别服务适合实时语音传输、识别前端接入及协议入门实践。1. 一条 20 毫秒的音频帧是怎么从麦克风跑到对面耳机里的很多人第一次接触 websocket 语音收发脑子里想的都是「录音 → 存成 wav → base64 塞进 JSON 发过去」。真跑起来就会发现对面听到的是一段一段的碎片延迟两三秒说话像在对讲机里排队。这套 Python 代码走的是另一条路麦克风每采满 20 毫秒就直接推一帧裸 PCM 二进制走 WebSocket 长连接丢给服务端服务端不做解码、不做转码原样广播给房间里的其他人。它适合两类人一类是想给自己的工具加个「实时对讲」能力又不想上 WebRTC 那套信令体系的开发者另一类是想搞清楚音频流在长连接上到底怎么切、怎么发、怎么缓冲才不炸的初学者。代码量不大服务端一个文件、客户端一个文件但把连接管理、二进制帧分流、背压丢弃、抖动缓冲这几个关键点都摆在了明面上。2. 架构与选型WebSocket 长连接加 PCM 裸流这套组合为什么够用2.1 为什么是长连接而不是 HTTP 分片上传做实时语音第一道选择题永远是传输层。HTTP 也能传音频前端定时 POST 一段录音服务端拼起来转发很多「语音消息」功能就是这么做的。问题在于它的模型是「一问一答」每次请求都要重新建连、带一遍头部、等一次响应。20 毫秒一帧的节奏下一秒就是 50 个请求光是 TCP/TLS 握手和 HTTP 头部的开销就能把带宽吃光更别说每帧都要等一个往返延迟。WebSocket 解决了这个模型问题一次握手之后连接一直挂着双方随时可以往管道里写数据没有请求-响应的配对关系。对语音这种「单向持续喷数据」的场景这是最省事的形态。它的帧头也足够小客户端发往服务端的帧要加 4 字节掩码加上 2 到 14 字节的变长头部一帧 640 字节的音频负载额外开销大概 1% 出头完全在接受范围内。WebRTC 当然是更专业的答案它自带 jitter buffer、丢包重传、回声消除、NACK/PLI 一整套。但代价是要搭信令服务器、要处理 ICE 候选、要面对 NAT 穿透失败时的一堆兜底逻辑。如果你的场景是内网工具、小团队对讲、或者只是想先跑通「能听见」WebSocket 的复杂度曲线要平缓得多。我一般的判断标准是单向、小房间、对延迟容忍在 200 毫秒以内就先用 WebSocket 把链路跑通别一上来就啃 WebRTC。2.2 文本帧管控制二进制帧管音频WebSocket 协议本身区分文本帧和二进制帧这个特性在这套代码里被用得很到位。连接建立后客户端先发一条 JSON 文本帧做握手告诉服务端「我要进哪个房间、我的采样率是多少、声道数几个、一帧多长」。服务端校验完回一条 JSON握手结束。之后所有音频数据都走二进制帧一个字节都不浪费在 JSON 的引号和花括号上。这种「控制面走文本、数据面走二进制」的分法好处是协议可读性还在。抓包或者打日志的时候你能一眼看出谁进了哪个房间、谁掉线了而不用去解析一堆裸字节。麻烦的地方在于服务端必须严格分流一旦把文本帧当成音频塞进播放缓冲PyAudio 那边立刻就是一片刺耳的噪音反过来把二进制帧丢给json.loads就是解码异常。这个坑后面第 5 章会单独讲。握手消息的字段设计我建议就按这套代码的结构来够用且不冗余字段类型说明常见取值typestring消息类型join/leave/pingroomstring房间标识任意短字符串sample_rateint采样率必须和采集端一致16000 / 48000channelsint声道数1单声道frame_msint每帧时长决定切片粒度20 / 40codecstring音频编码格式pcm_s16lesample_rate和frame_ms这两个字段是后面所有计算的基础。16000 Hz、16 bit、单声道、20 毫秒一帧算下来每帧就是16000 × 0.02 × 2 640字节。这个数字会反复出现在采集端的frames_per_buffer、服务端的帧长度校验、播放端的缓冲区计算里。只要有一处对不上声音就会变调或者卡顿这也是新手最容易翻车的地方。2.3 目录结构与依赖清单这套代码的工程结构刻意保持扁平不引入框架就是为了让人能一口气读完。我把它整理成下面这样跑起来之后你按这个结构往里加东西就行ws-voice/ ├── server.py # 服务端入口握手、房间管理、广播 ├── client.py # 客户端入口采集、发送、接收、播放 ├── audio_io.py # 采集与播放的封装隔离 PyAudio ├── protocol.py # 握手消息的编解码与字段校验 ├── requirements.txt └── tools/ └── recorder.py # 可选把收到的帧落成 wav方便回放排查依赖只有两个核心包加一个可选的分析工具# requirements.txt websockets12.0 # 服务端和异步客户端共用纯 asyncio 实现 PyAudio0.2.14 # 跨平台采集/播放底层是 PortAudio numpy1.26.4 # 可选算音量、做 VAD 时用得上websockets这个库同时提供服务端serve()和客户端connect()服务端和客户端用同一套 API心智负担小不用在websocket-client同步阻塞和websockets异步之间来回切换。PyAudio的安装是这套代码唯一的门槛Linux 上要先装portaudio19-devmacOS 用brew install portaudioWindows 上 pip 装预编译包一般能直接过。装不上的话把audio_io.py换成sounddevice也基本是改几行参数的事接口形状差不多。提示PyAudio的frames_per_buffer不要大于 1024。设成 1024 在 16 kHz 下就是 64 毫秒一帧听起来会明显发闷、有回音感。20 到 40 毫秒是比较舒服的区间。3. 服务端拆解连接管理器、房间广播与背压处理的完整链路3.1 主循环与握手先收一条 JSON再切到二进制模式服务端的入口是一个async def handler(ws)websockets.serve会给每条新连接开一个协程跑它天然就是并发的。进到 handler 里第一件事不是急着收音频而是先等一条 JSON 握手消息把这条连接的元信息房间号、采样率、帧长记下来挂到连接对象上或者存进字典。# server.py import asyncio import json import logging import websockets from websockets.server import WebSocketServerProtocol from websockets.exceptions import ConnectionClosed logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) # 房间表room_id - 该房间内所有活跃连接 ROOMS: dict[str, set[WebSocketServerProtocol]] {} # 每个连接自己的元信息key 用 id(ws) 或直接挂在 ws 上 META: dict[int, dict] {} async def handler(ws: WebSocketServerProtocol): 单条连接的生命周期握手 - 收帧 - 广播 - 清理 try: # 第一步握手。超时 5 秒没等到就断开防止空连接占资源 raw await asyncio.wait_for(ws.recv(), timeout5.0) if not isinstance(raw, str): await ws.close(code1002, reasonfirst frame must be json) return hello json.loads(raw) room_id str(hello.get(room, default)) meta { room: room_id, sample_rate: int(hello.get(sample_rate, 16000)), channels: int(hello.get(channels, 1)), frame_ms: int(hello.get(frame_ms, 20)), frames: 0, # 已转发帧数用来做统计和限流 } META[id(ws)] meta ROOMS.setdefault(room_id, set()).add(ws) # 回执告诉客户端当前房间有几个人方便它自己决定要不要开播 await ws.send(json.dumps({ type: joined, room: room_id, peers: len(ROOMS[room_id]) - 1, })) # 第二步纯数据循环。这里收到的都应该是二进制音频帧 async for message in ws: if isinstance(message, str): # 控制帧心跳、离开房间之类走文本 await handle_control(ws, meta, message) continue meta[frames] 1 await broadcast(room_id, ws, message) except asyncio.TimeoutError: logging.warning(handshake timeout: %s, ws.remote_address) except ConnectionClosed: pass finally: cleanup(ws)这段代码的逻辑分三段。第一段是握手用asyncio.wait_for兜住 5 秒超时避免有人连上来不说话把连接池占满。第二段是主循环async for message in ws是把recv()循环写得更短的语法糖它在连接关闭时会自动结束迭代不会抛异常。第三段是finally里的清理无论正常断开还是异常退出都要把连接从ROOMS和META里摘掉不然房间集合会越积越大广播时给死连接发数据。参数上有两个地方值得改。sample_rate和frame_ms目前只是记录没有做强校验如果你的客户端是外部可控的建议在这里加一条判断只允许预设的几个组合通过否则后面广播出去的数据长度五花八门对端播放缓冲很难写。frames计数器则是留着的钩子后面要做限流、做统计、做「这个连接是不是在疯狂刷帧」的判断都从它身上取。3.2 房间广播为什么不能串行 await send广播看上去就是「遍历房间里的其他连接挨个 send」但真这么写迟早出问题。await ws.send()在对方的 TCP 接收窗口满了的时候是会被挂住的你串行地 await 十个人的发送只要其中一个网络卡住后面九个人全跟着卡整个房间的延迟就被一个人拉爆了。这在实测里非常明显一个人切到弱网全房间的人开始听「卡带」。正确的做法是给每个对端的发送都开一个独立任务用asyncio.gather并发执行并且给每次发送套一个超时。超时了不是干等而是把这个对端标记成「慢消费者」考虑丢弃当前帧甚至直接断开连接。# server.py续 SEND_TIMEOUT 1.0 # 单帧下发超时超过就认为对端跟不上 async def broadcast(room_id: str, sender: WebSocketServerProtocol, payload: bytes): 把一帧音频发给房间里除发送者以外的所有人 peers ROOMS.get(room_id) if not peers: return targets [ws for ws in peers if ws is not sender and ws.state.name OPEN] if not targets: return # 并发下发每个对端独立超时return_exceptions 保证一个失败不影响其他 results await asyncio.gather( *(_safe_send(ws, payload) for ws in targets), return_exceptionsTrue, ) # 把发不动的连接清出房间避免它持续拖慢整个广播 for ws, ok in zip(targets, results): if ok is not True: logging.info(evict slow peer %s, ws.remote_address) cleanup(ws) async def _safe_send(ws: WebSocketServerProtocol, payload: bytes) - bool: try: await asyncio.wait_for(ws.send(payload), timeoutSEND_TIMEOUT) return True except (asyncio.TimeoutError, ConnectionClosed): return False_safe_send返回布尔值而不是抛异常是为了让上层的gather结果能一一对应到targets列表方便定位是哪个对端出了问题顺手把它踢出去。return_exceptionsTrue是必须的否则第一个异常会中断整个gather其他已经发出去的任务结果也拿不到。SEND_TIMEOUT设 1 秒是个经验值。一帧 20 毫秒的音频超过 1 秒还没发出去这条连接基本已经废了留着只会持续堆积任务。如果你要做「弱网降级」可以把它改成分级策略超时先丢帧连续丢 N 帧再断开。cleanup()这个函数要做得干净利落它是整套代码里最容易被写漏的地方def cleanup(ws: WebSocketServerProtocol): meta META.pop(id(ws), None) if not meta: return room ROOMS.get(meta[room]) if room is not None: room.discard(ws) # 空房间要删掉否则 ROOMS 会随房间号无限增长 if not room: ROOMS.pop(meta[room], None) logging.info(bye room%s frames%s, meta[room], meta[frames])这里用id(ws)做字典 key 有个隐患Python 的对象 id 在对象被回收后可能被复用。websockets的连接对象在 handler 的整个生命周期内都是活的所以在本例里安全但如果你的代码把连接对象存到别处又提前释放就可能出现元信息串号。稳妥一点的做法是给每个连接生成一个 uuid握手时写进 meta用 uuid 做 key。3.3 背压、心跳与启动参数websockets.serve()的调用参数决定了服务端的健壮性默认值大多偏保守需要显式调一下# server.py入口 async def main(): async with websockets.serve( handler, host0.0.0.0, port8765, ping_interval20, # 每 20 秒发一次 ping探测死连接 ping_timeout20, # 20 秒收不到 pong 就判定断开 max_size64 * 1024, # 单帧上限 64KB音频帧远小于此防恶意大包 max_queue32, # 单连接待发队列上限超过就触发背压 close_timeout5, ): logging.info(voice server listening on ws://0.0.0.0:8765) await asyncio.Future() # 永久挂起直到进程被信号杀掉 if __name__ __main__: try: asyncio.run(main()) except KeyboardInterrupt: passping_interval和ping_timeout是防「假死连接」的关键。TCP 连接在对方拔网线、切 WiFi 的情况下本地是感知不到的连接会一直挂在 ESTABLISHED 状态你以为还在广播其实数据全进了黑洞。协议层的心跳能在几十秒内把它揪出来。做反向代理的话这两个值还要和代理的空闲超时对齐——很多反代默认 60 秒没数据就断连心跳周期必须小于它。max_queue是websockets库自带的背压开关。当某个连接待发送的消息堆到这个数量库会把send()挂起等队列腾出空间。配合前面的SEND_TIMEOUT效果就是「慢消费者先被挂起超时后再被踢掉」。这个组合比自己在业务层维护一个发送队列要省事得多我一般优先用它。max_size限制的是单条消息的大小。音频帧只有几百字节设 64KB 已经很宽松主要是防止有人发个几兆的包把服务端内存打满。最后是房间人数变化的通知。上面的握手只在新人进来时回了peers房间里其他人并不知道有人加入。要让客户端能显示「当前 3 人在线」得在加入和离开的时候各推一条控制消息async def notify_room(room_id: str, event: str, count: int): 房间人数变化时给房里所有人推一条文本帧 peers ROOMS.get(room_id, set()) msg json.dumps({type: room_state, event: event, peers: count}) await asyncio.gather( *(ws.send(msg) for ws in peers if ws.state.name OPEN), return_exceptionsTrue, )在handler里ROOMS.setdefault(...).add(ws)之后调用notify_room(room_id, join, len(ROOMS[room_id]))在cleanup里调用一次leave。这条消息是文本帧客户端那边一定要和音频二进制帧分开处理别混进播放队列。4. 客户端拆解采集线程、发送协程、播放缓冲三段式落地4.1 采集PyAudio 是阻塞的必须扔进线程客户端最容易写错的地方是把stream.read()直接放进协程里调。这个调用是阻塞的它会让出 CPU 等待声卡填满缓冲区一执行就把整个事件循环卡死收发全停。正确的做法是把采集放在独立线程里跑采到的帧通过loop.call_soon_threadsafe丢回 asyncio 的事件循环。# audio_io.py import queue import threading import pyaudio SAMPLE_RATE 16000 CHANNELS 1 FRAME_MS 20 # 每帧采样点数 采样率 * 帧长(秒) CHUNK int(SAMPLE_RATE * FRAME_MS / 1000) # 16000 * 0.02 320 FORMAT pyaudio.paInt16 # 16 bit 小端每采样 2 字节 def capture_loop(out_q: queue.Queue, stop: threading.Event): 独立线程持续从麦克风读帧塞进线程安全队列 pa pyaudio.PyAudio() stream pa.open( formatFORMAT, channelsCHANNELS, rateSAMPLE_RATE, inputTrue, frames_per_bufferCHUNK, ) try: while not stop.is_set(): # exception_on_overflowFalse缓冲区溢出时返回数据而不是抛异常 data stream.read(CHUNK, exception_on_overflowFalse) if out_q.full(): # 队列满说明上游发送跟不上丢掉最旧的一帧保实时性 try: out_q.get_nowait() except queue.Empty: pass out_q.put_nowait(data) finally: stream.stop_stream() stream.close() pa.terminate()CHUNK这个值就是前面算的 320 个采样点乘以 2 字节得到 640 字节一帧和服务端握手时声明的frame_ms20严格对应。exception_on_overflowFalse这个参数很关键默认情况下缓冲区溢出会抛IOError在长时间运行的采集线程里这种偶发溢出几乎必然发生抛出来就会把线程干掉。设成 False 之后它只是丢掉那部分数据线程继续跑。out_q.full()那段是采集侧的背压。当网络卡住、发送协程消费不过来时队列会堆满。这时候有两条路阻塞等待采集线程被卡住声卡缓冲区溢出结果是丢一大段或者丢最旧的帧保持实时丢掉历史。语音场景毫无疑问选后者——宁可丢 20 毫秒的旧声音也不要让延迟越积越大。4.2 发送协程把线程队列桥接进 asyncio采集线程写的是标准库的queue.Queueasyncio 侧要消费它中间需要一个桥。最直接的办法是采集线程拿到事件循环的引用用call_soon_threadsafe把数据投递到asyncio.Queue# client.py import asyncio import threading import queue import websockets from audio_io import capture_loop, SAMPLE_RATE, CHANNELS, FRAME_MS async def sender(ws, aio_q: asyncio.Queue, stop: threading.Event): 把 asyncio.Queue 里的音频帧一帧帧推给服务端 sent 0 while not stop.is_set(): try: # 加超时避免服务端已经断开时在这里永久阻塞 frame await asyncio.wait_for(aio_q.get(), timeout1.0) except asyncio.TimeoutError: continue try: await ws.send(frame) sent 1 if sent % 500 0: # 每 10 秒打一条方便观察是否在流动 print(f[sender] {sent} frames sent) except websockets.ConnectionClosed: stop.set() break def bridge_thread_to_asyncio(loop, thread_q: queue.Queue, aio_q: asyncio.Queue, stop: threading.Event): 守护线程把 queue.Queue 的数据搬到 asyncio.Queue while not stop.is_set(): try: data thread_q.get(timeout0.2) except queue.Empty: continue # 必须用 call_soon_threadsafe跨线程直接操作 asyncio 对象是未定义行为 loop.call_soon_threadsafe(_put_drop_oldest, aio_q, data) def _put_drop_oldest(q: asyncio.Queue, data: bytes): if q.full(): try: q.get_nowait() except asyncio.QueueEmpty: pass q.put_nowait(data)loop.call_soon_threadsafe是这里唯一正确的跨线程投递方式asyncio.Queue不是线程安全的直接从别的线程put_nowait会在高并发下出现丢数据或内部状态错乱。_put_drop_oldest里又做了一次丢弃旧帧是因为采集侧和发送侧之间实际有两级队列两级都可能积压。asyncio.wait_for(aio_q.get(), timeout1.0)加超时是为了应对服务端已经断开、但stop标志还没来得及被设上的时间窗。没有这个超时协程会一直挂在get()上主协程那边等不到它自然退出进程就卡住了。4.3 接收与播放预缓冲三帧抖动才不刺耳接收侧看着简单async for收二进制帧丢给播放就完了。实际跑起来你会发现网络抖动会让帧到达的时间忽快忽慢直接一帧一写播放出来是「哒哒哒」的断续声。解决办法是在播放前加一层抖动缓冲先攒够几帧再开始播播放时如果队列空了补一帧静音而不是停下等。# audio_io.py续 def play_loop(in_q: queue.Queue, stop: threading.Event, prebuffer: int 3): 独立线程从队列取帧写声卡带预缓冲和欠载补静音 pa pyaudio.PyAudio() stream pa.open( formatFORMAT, channelsCHANNELS, rateSAMPLE_RATE, outputTrue, frames_per_bufferCHUNK, ) silence b\x00 * (CHUNK * 2) # 一帧长度的静音16bit 每采样 2 字节 started False try: while not stop.is_set(): # 预缓冲攒够 prebuffer 帧再开播压下前期的到达抖动 if not started: if in_q.qsize() prebuffer: threading.Event().wait(0.01) continue started True try: data in_q.get(timeout0.1) except queue.Empty: # 欠载补一帧静音避免声卡 underrun 导致的爆音 data silence stream.write(data, exception_on_underflowFalse) finally: stream.stop_stream() stream.close() pa.terminate()prebuffer3意味着先攒 60 毫秒的音频再开始播。这个值太小压不住抖动太大会白白增加延迟。实测下来局域网 2 到 3 帧、公网 4 到 6 帧是比较合适的起点。它本质上是用延迟换平滑度属于典型的可调参数。exception_on_underflowFalse和采集侧那个参数是一个道理播放缓冲写空的时候默认会抛异常加上这个参数后变成静默处理。配合补静音的兜底声卡不会因为偶发的数据断流就中断播放。主流程把三个部分串起来用asyncio.gather同时跑任一个退出就整体收尾# client.py主入口 async def main(room: str, url: str): loop asyncio.get_running_loop() stop threading.Event() thread_q: queue.Queue queue.Queue(maxsize50) # 采集 - 桥接 aio_q: asyncio.Queue asyncio.Queue(maxsize50) # 桥接 - 发送 play_q: queue.Queue queue.Queue(maxsize50) # 接收 - 播放 async with websockets.connect(url, max_size64 * 1024) as ws: # 握手必须第一个发服务端在等它 await ws.send(json.dumps({ type: join, room: room, sample_rate: SAMPLE_RATE, channels: CHANNELS, frame_ms: FRAME_MS, codec: pcm_s16le, })) cap threading.Thread(targetcapture_loop, args(thread_q, stop), daemonTrue) play threading.Thread(targetplay_loop, args(play_q, stop), daemonTrue) cap.start() play.start() threading.Thread(targetbridge_thread_to_asyncio, args(loop, thread_q, aio_q, stop), daemonTrue).start() async def receiver(): async for msg in ws: if isinstance(msg, str): print([ctrl], msg) # 控制帧单独处理别进播放队列 continue if play_q.full(): play_q.get_nowait() play_q.put_nowait(msg) try: await asyncio.gather(sender(ws, aio_q, stop), receiver()) except websockets.ConnectionClosed: pass finally: stop.set() cap.join(timeout1) play.join(timeout1)receiver里对文本帧的处理是硬性要求控制消息必须单独走一条路绝不能混进play_q。这一点在做房间人数通知之后会变得尤为关键混进去就是一阵刺耳的电流声。所有队列都设了maxsize配合丢弃最旧帧的策略整条链路在弱网下会自己降级而不是无限堆积内存。5. 排查清单语音流跑不通时先按这五条顺序查5.1 连接与协议层连不上、连上就断、报错在握手现象一客户端连上后 5 秒内被服务端断开日志里只有handshake timeout。原因通常是握手的时序错了。服务端一进 handler 就await ws.recv()等那条 JSON如果客户端是先启动采集线程、等采到第一帧音频才发数据第一帧就是二进制服务端拿到bytes直接按「首帧必须是 JSON」关掉连接。另一种情况是客户端用websocket-client那种同步库send的顺序和预期不一致。 解决把握手发在connect()成功之后、启动任何线程之前第一行就发。服务端侧把「收到非文本帧」的错误原因写清楚别只回一个 1002 码排查时看日志能省很多时间。现象二局域网一切正常一挂到反向代理后面60 秒左右必断断了之后重连也收不到声音。原因是代理的空闲超时。很多反向代理默认 60 秒内没有数据往来就主动断开而语音场景中没人说话的时候连接上确实一个字节都不发。 解决把ping_interval设成 20 秒让协议层的心跳持续产生流量同时确认代理配置里对 WebSocket 的Upgrade头做了透传并且proxy_read_timeout大于心跳周期。这两个条件缺一个都还是会断。现象三房间人多了之后所有人同时开始卡日志里没有异常。原因是广播用了串行await send()一个慢连接拖住了整个房间的循环。这种问题在两个人测试时完全复现不出来人一多就爆。 解决按 3.2 的写法改成asyncio.gather并发下发每路带独立超时超时的连接直接清理。改完之后你会发现即使有人故意限速其他人的声音也不受影响。5.2 音频链路能连上但声音不对现象四对面能听到声音但语速变快、音调变尖或者反过来低沉拖沓像慢放。这是采样率不匹配的典型症状。采集端按 44100 Hz 打开发送时却在握手消息里声明sample_rate: 16000对端按 16 kHz 播放 44.1 kHz 的数据音调自然被拉高。反过来就是变低。这个问题在换机器、换声卡之后特别容易出现因为不同设备的默认采样率不一样。 解决pyaudio.open()的rate、握手消息里的sample_rate、播放端的rate这三处必须是同一个常量从一个配置文件或者模块级常量里取别在三个地方各写一遍字面量。设备不支持 16 kHz 采集时少数声卡会拒绝要么在握手前做一次重采样要么整个链路统一用 48 kHz。现象五声音断断续续每隔一两秒「咔」一下跑久了内存还一直涨。断续多半是缓冲区配置的问题frames_per_buffer设得太大比如 4096一帧就是 100 多毫秒播放端的抖动缓冲根本压不住表现出来就是一顿一顿。内存涨则是队列没设上限弱网时帧在队列里无限堆积。 解决把帧长压到 20 到 40 毫秒播放端加 3 帧预缓冲和欠载补静音。所有queue.Queue和asyncio.Queue都显式指定maxsize满了就丢最旧的帧。加个每 500 帧打一次的计数日志能直观看到收发是否平衡——发送数远大于接收数就是某一段在积压。注意排查音频问题时先把收到的帧落成 wav 文件16 kHz、单声道、16 bit 直接写就行用播放器放一遍。如果 wav 里听着正常问题在播放链路如果 wav 里就已经断续问题在采集或传输。这一步能砍掉一半的排查时间。6. 二次开发把「能听见」改成「能上线」的四个改造点这套代码跑通之后第一个撞上的墙一定是带宽。16 kHz、16 bit、单声道的裸 PCM 是每秒 32 KB折合 256 kbps一个人说话就顶得上半路高清视频。加 Opus 编码是最划算的一步同样是语音质量码率能压到 16 到 24 kbps降了一个数量级方案采样率码率20 秒语音大小依赖PCM s16le 裸流16 kHz256 kbps约 640 KB无Opus 16 kbps16 kHz16 kbps约 40 KBopuslib libopusOpus 24 kbps16 kHz24 kbps约 60 KB同上接入点在sender和receiver两头改动很小。发送侧在ws.send(frame)之前编码接收侧在play_q.put_nowait(msg)之前解码# codec.py —— 编码钩子接在 sender 的 send 之前 import opuslib ENC_RATE 16000 FRAME_SAMPLES 320 # 20ms 16kHz encoder opuslib.Encoder(ENC_RATE, 1, opuslib.APPLICATION_VOIP) # VOIP 模式针对人声优化 def encode_pcm(pcm: bytes) - bytes: 640 字节 PCM - 约 40 字节 Opus 包 return encoder.encode(pcm, FRAME_SAMPLES) # 服务端不用动它只做字节转发编码格式放在握手消息的 codec 字段里即可这里有个容易忽略的点Opus 的编解码器是有状态的Encoder和Decoder实例必须每条连接各持一份不能全局共享。两个人同时说话时共用同一个 encoder解出来的声音会互相污染听起来像两个人的声音叠在一起。另外握手消息里的codec字段要跟着改成opus让对端知道该用哪条解码路径。第二个改造点是 VAD。现在不管有没有人说话采集线程都在 20 毫秒一帧地往外推安静的时候推的全是静音包。加一个基于 RMS 能量的简单静音检测低于阈值连续 N 帧就不发只在「从静音切到有声」时发一个标记帧让对端开始播放带宽能再省掉六七成。numpy算 RMS 只要一行阈值设 500 左右16 bit 满量程 32768在普通室内环境够用。第三个是鉴权。现在任何人连上来报个房间号就能进生产环境肯定不行。最省事的做法是在握手的第一条 JSON 里带token服务端校验失败直接close(code4001)。别把 token 放在 URL query 里那会进代理日志。校验逻辑放在handler的握手段在ROOMS.setdefault(...)之前没过就别让连接进房间表。第四个是转写管道。如果后面要接语音识别别在主循环里同步调接口那会把广播卡住。开一个asyncio.Queue收到帧就丢进去一个独立协程慢慢消费、攒够一段再送 ASR结果作为文本帧推回房间。这条链路的背压和音频广播是独立的互不影响。我踩过最深的一个坑是当初为了省事把 Opus 编码器写成了模块级全局变量测试时只有一个人说话一切正常上线第一天三个人同时开麦所有人都在反馈「声音像泡在水里」。查了整整一个下午最后对着opuslib的文档才反应过来状态共享的问题。从那以后我每次给音频链路加有状态的东西都强制先跑一遍「三个客户端同时说话」的用例不管功能看起来多简单。希望这套代码和这些踩坑记录能帮你在自己的项目里少走一遍同样的弯路。本文还有配套的精品资源点击获取