简介这份资源提供基于WebSocket协议实现语音收发的Python服务器与客户端代码面向具备一定Python网络编程基础、希望实践实时音频传输与语音处理的开发者。包内共3个文件以2个py脚本和1个bak备份文件为主分别对应服务端与客户端逻辑压缩包约9KB体量轻便便于直接阅读与二次修改。代码演示了使用Flask-SocketIO搭建WebSocket服务端、通过websocket-client建立客户端连接并以JSON格式传输语音数据的基本流程同时提及可结合Kaldi-Gstreamer-Server进行解码与语音识别。已有3113人学习下载读者可据此理解长连接通信、事件回调与音频数据收发机制并在此基础上调整消息处理逻辑、增加认证或优化语音处理流程适合作为实时语音通信项目的入门参考与改造起点。1. 从一次语音房调试说起这套 WebSocket 收发语音的 Python 代码到底能干什么去年帮一个做在线协作工具的小团队排查问题他们的语音房在弱网下频繁断连日志里全是 WebSocket 握手成功后又莫名关闭的记录。翻完代码发现问题不在业务逻辑而在音频数据的分包和发送节奏上——他们用 WebSocket 传 PCM 裸流却按文本消息的节奏在发缓冲区一满就崩。那次之后我意识到WebSocket 传语音这件事看着简单真要在项目里跑稳收发两端的数据格式、分片策略、异常重连都得抠清楚。这份 Python 代码解决的正是这个场景用 WebSocket 协议搭一对服务端和客户端服务端负责接收客户端推上来的语音数据并转发给其他在线连接客户端负责采集音频、编码、发送同时接收远端语音并播放。它适合两类人一是想给自己的桌面工具或 Web 应用加实时语音对讲功能的开发者二是正在学网络编程、想找一个能跑通的 WebSocket 双向流式传输案例的新手。代码不依赖重型框架核心逻辑集中在连接管理和音频帧的收发上拿下来改改就能嵌进自己的项目里。2. 技术选型与整体架构为什么用 websockets 库而不是原生 socket2.1 服务端与客户端的职责划分这套代码的架构是典型的「中心转发」模式。服务端不处理音频编解码只做三件事维护所有活跃连接的列表、接收某个客户端发来的音频帧、把帧广播给除发送者之外的其他连接。客户端则承担采集、编码、发送、接收、解码、播放六个环节。把编解码放在客户端而不是服务端好处是服务端压力小坏处是不同客户端的音频参数必须提前约定一致否则收到的数据没法直接播放。常见做法是服务端用一个字典存连接对象和对应的用户标识广播时遍历字典逐个发送。这里要注意WebSocket 的 send 是异步操作遍历时不能阻塞否则一个慢连接会拖垮整个广播循环。代码里用 asyncio.gather 并发发送并对每个发送做异常捕获某个连接断了不影响其他人。2.2 为什么选 websockets 库Python 里做 WebSocket 服务端有好几个选择websockets、aiohttp、Flask-SocketIO。这份代码用的是 websockets原因是它足够轻API 就是 async def handler(websocket) 这种直白的形式没有额外的路由和中间件概念。对于语音传输这种需要长连接、高频小包、低延迟的场景websockets 基于 asyncio 的实现能保证每个连接独立协程处理不会因为某个客户端的网络抖动影响其他连接。import asyncio import websockets # 存储所有活跃连接key 为 websocket 对象value 为用户标识 connected_clients {} async def handler(websocket): # 新连接进来先注册 client_id id(websocket) connected_clients[websocket] client_id try: async for message in websocket: # 收到音频帧后广播给其他客户端 await broadcast(message, senderwebsocket) except websockets.ConnectionClosed: pass finally: # 连接断开必须清理否则广播时会向死连接发数据 connected_clients.pop(websocket, None) async def broadcast(data, sender): targets [ws for ws in connected_clients if ws ! sender] if not targets: return # 并发发送单个失败不影响整体 results await asyncio.gather( *[ws.send(data) for ws in targets], return_exceptionsTrue ) for ws, res in zip(targets, results): if isinstance(res, Exception): connected_clients.pop(ws, None)这段代码里connected_clients 用 websocket 对象本身做 key简单直接。broadcast 里先过滤掉发送者再用 gather 并发发送。return_exceptionsTrue 是关键它让某个连接发送失败时不会抛异常中断整个 gather而是把异常作为结果返回后面统一清理失效连接。参数 data 是二进制音频帧sender 是发送方的 websocket 对象。2.3 音频数据的传输格式约定WebSocket 支持文本帧和二进制帧传语音必须用二进制帧否则 base64 编码会带来 33% 的体积膨胀和额外的编解码开销。代码里客户端发送时用 websocket.send(bytes_data)服务端 async for message in websocket 收到的就是 bytes 类型。采样率、位深、声道数这些参数不在协议层约定而是靠两端代码里写死的常量保持一致。常见做法是 16kHz 采样、16 位深、单声道每帧 20ms这样一帧是 640 字节既能保证实时性又不会因为包太小导致网络开销占比过高。3. 客户端实现拆解从麦克风采集到 WebSocket 发送的完整链路3.1 音频采集与分帧客户端采集音频用 pyaudio 库它是对 PortAudio 的 Python 封装跨平台支持好。打开音频流时需要指定采样率、帧大小、声道数。帧大小按 20ms 算16kHz 采样率下就是 320 个采样点每个采样点 2 字节一帧 640 字节。import pyaudio import asyncio import websockets CHUNK 320 # 每帧采样点数16kHz 下对应 20ms FORMAT pyaudio.paInt16 CHANNELS 1 RATE 16000 audio pyaudio.PyAudio() stream audio.open( formatFORMAT, channelsCHANNELS, rateRATE, inputTrue, frames_per_bufferCHUNK ) async def send_audio(websocket): loop asyncio.get_event_loop() while True: # pyaudio 的 read 是阻塞的放到线程池里避免卡住事件循环 data await loop.run_in_executor(None, stream.read, CHUNK) await websocket.send(data)这里有个容易翻车的点stream.read 是同步阻塞调用直接放在 async 函数里会阻塞整个事件循环导致接收远端音频的协程也卡住。用 run_in_executor 把它丢到默认线程池执行事件循环就能继续处理其他任务。参数 CHUNK 决定每帧时长改小延迟低但包多改大延迟高但网络开销小20ms 是语音场景的常用折中。3.2 接收与播放的并发处理客户端需要同时做发送和接收两件事用 asyncio.create_task 起两个协程一个跑 send_audio一个跑 receive_audio。接收端收到二进制帧后直接写入输出流播放。async def receive_audio(websocket): loop asyncio.get_event_loop() out_stream audio.open( formatFORMAT, channelsCHANNELS, rateRATE, outputTrue, frames_per_bufferCHUNK ) async for message in websocket: if isinstance(message, bytes): # 写入播放流同样是阻塞操作走线程池 await loop.run_in_executor(None, out_stream.write, message) async def main(): uri ws://localhost:8765 async with websockets.connect(uri) as websocket: # 发送和接收并发执行 await asyncio.gather( send_audio(websocket), receive_audio(websocket) )接收端判断 isinstance(message, bytes) 是为了过滤掉可能的文本控制消息。播放流的 frames_per_buffer 要和采集端一致否则播放节奏会对不上。asyncio.gather 让两个协程并发跑任何一个抛异常都会取消另一个实际项目里通常会在外层加 while True 重连逻辑这里为了代码清晰没加。3.3 连接建立与参数协商客户端连接时可以在 URL 里带查询参数比如 ws://host:port/audio?rate16000channels1服务端在 handler 里通过 websocket.path 解析。这样不同客户端可以用不同参数服务端转发时按目标连接的参数做重采样。不过这份代码为了简单参数是两端写死的二次开发时如果要做多端兼容建议把参数协商加上。4. 服务端进阶连接管理、广播策略与异常处理4.1 连接生命周期管理服务端的连接管理核心是「注册-使用-清理」三步。注册在 handler 入口清理在 finally 块。中间任何环节抛异常finally 都能保证连接被移除。常见错误是只在正常关闭时清理遇到网络异常断开时忘了 pop导致 connected_clients 里堆积死连接广播时不断向这些连接发数据既浪费资源又拖慢速度。async def handler(websocket): client_id id(websocket) connected_clients[websocket] client_id try: async for message in websocket: await broadcast(message, senderwebsocket) except websockets.ConnectionClosed as e: print(f连接关闭: code{e.code}, reason{e.reason}) except Exception as e: print(f未预期异常: {type(e).__name__}: {e}) finally: connected_clients.pop(websocket, None) print(f客户端 {client_id} 已清理当前在线 {len(connected_clients)})ConnectionClosed 异常里带 code 和 reason调试时很有用。code 1000 是正常关闭1006 是异常断开1009 是消息过大。根据 code 可以判断是客户端主动退出还是网络问题方便做针对性的重连策略。4.2 广播策略的优化基础版广播是遍历所有连接逐个 send连接数少时没问题上百个连接时串行发送的延迟就很明显了。优化方向有两个一是用 asyncio.gather 并发发送二是对慢连接做超时控制。async def broadcast(data, sender): targets [ws for ws in connected_clients if ws ! sender] if not targets: return # 给每个发送加 1 秒超时防止慢连接拖垮广播 async def safe_send(ws): try: await asyncio.wait_for(ws.send(data), timeout1.0) except (asyncio.TimeoutError, websockets.ConnectionClosed): connected_clients.pop(ws, None) await asyncio.gather(*[safe_send(ws) for ws in targets])asyncio.wait_for 给每个 send 加了 1 秒超时超时就认为该连接有问题直接清理。这样即使某个客户端网络卡顿也不会影响其他客户端收到音频。超时时间根据实际网络情况调整局域网可以设 0.5 秒公网建议 2 到 3 秒。4.3 消息大小限制与分片websockets 库默认单条消息最大 1MB音频帧通常远小于这个值不用改。但如果客户端把多帧攒在一起发或者传的是未压缩的高采样率音频就可能超限。服务端启动时可以显式设置 max_sizeasync def main(): async with websockets.serve( handler, 0.0.0.0, 8765, max_size2**20, # 单条消息最大 1MB ping_interval20, # 每 20 秒发一次 ping 保活 ping_timeout20 # ping 后 20 秒没响应就断开 ): await asyncio.Future() # 永久运行ping_interval 和 ping_timeout 是保活关键。WebSocket 连接经过 NAT 网关时长时间没数据会被网关断开定期 ping 能维持连接。20 秒是常用值太短浪费流量太长起不到保活作用。5. 避坑与排查语音传输中常见的五类问题5.1 现象客户端能连上但听不到声音原因通常是音频参数不一致。发送端用 16kHz 采集接收端用 44.1kHz 播放数据长度对不上播放出来是噪音或者直接没声音。排查时先确认两端的 RATE、CHANNELS、FORMAT 三个常量是否完全一致。另一个可能是接收端没判断 bytes 类型把二进制帧当文本处理了。5.2 现象声音断断续续有卡顿一般是发送节奏和采集节奏不匹配。stream.read 是阻塞的如果放在事件循环里直接调用每次读取都会卡住整个循环导致接收协程得不到执行。解决就是用 run_in_executor 把阻塞调用丢到线程池。另外检查 CHUNK 是否设得过大20ms 对应 320 采样点设成 1024 就是 64ms 一帧延迟明显增加。5.3 现象连接几分钟后自动断开NAT 网关会清理长时间无数据的连接。WebSocket 协议层有 ping/pong 机制但需要显式开启。服务端 serve 时设置 ping_interval20客户端 websockets.connect 时也设置 ping_interval20。如果中间有反向代理还要确认代理的超时时间大于 ping 间隔。5.4 现象广播时某个客户端断开导致其他人也收不到这是异常没捕获干净。broadcast 里如果直接 await ws.send(data)某个连接抛 ConnectionClosed 异常会中断整个循环后面的连接就收不到数据了。必须用 gather 加 return_exceptionsTrue或者每个 send 单独 try/except。5.5 现象内存持续增长connected_clients 字典只增不减。检查 handler 的 finally 块是否真的执行了。如果 handler 里有 while True 循环且没有退出条件异常发生时可能不会走到 finally。另外确认没有在别处重复注册同一个连接比如重连时旧连接没清理就注册了新连接。6. 二次开发改造从能跑到好用需要动哪些地方这份代码跑通不难但要用在真实项目里有几个地方值得动手改。第一个是加参数协商。现在采样率、声道数是写死的改成客户端连接时通过 URL 参数或首条控制消息上报服务端记录每个连接的参数转发时按目标连接的参数做重采样。重采样可以用 audioop.ratecvPython 标准库自带不用额外装包。第二个是加音频编码。PCM 裸流带宽占用大16kHz 16bit 单声道一路就是 256kbps十路并发就把上行跑满了。改成 Opus 编码能压到 24kbps 左右音质还过得去。Python 里可以用 opuslib 或者 pyogg编码后的帧更小WebSocket 传输压力也小。注意编码后每帧大小不固定接收端解码时要按 Opus 的帧边界处理不能像 PCM 那样按固定字节数切。第三个是加房间概念。现在所有连接都在一个广播组里A 说话 B、C、D 都能听到。实际场景通常需要分房间在 connected_clients 里加一层房间 ID 的映射广播时只发给同房间的连接。数据结构可以改成 dict[room_id, dict[websocket, client_id]]handler 里从 URL 路径解析房间号。第四个是加重连和状态恢复。客户端断线后自动重连重连时带上之前的用户标识服务端识别后恢复其房间归属。重连逻辑用指数退避第一次 1 秒第二次 2 秒最多到 30 秒避免服务端刚重启就被大量重连打满。async def connect_with_retry(uri, max_retries5): for attempt in range(max_retries): try: return await websockets.connect(uri, ping_interval20) except (OSError, websockets.InvalidURI) as e: wait min(2 ** attempt, 30) print(f连接失败{wait}秒后重试: {e}) await asyncio.sleep(wait) raise ConnectionError(重试次数耗尽)这段重连代码里2 ** attempt 实现指数退避min(..., 30) 封顶 30 秒。捕获 OSError 覆盖网络不可达的情况InvalidURI 覆盖地址写错的情况。实际项目里还可以加 jitter在等待时间上随机加减 0 到 1 秒避免多个客户端同时重连。验证改造效果时我一般会开三个终端一个跑服务端两个跑客户端然后手动拔掉其中一个客户端的网线模拟断线观察服务端日志里连接清理是否及时、另一个客户端是否还能正常收发。再开一个终端用 websocat 连上去发一段预录的 PCM 数据确认服务端广播逻辑没被改坏。这套流程走下来基本能覆盖连接管理、广播、异常处理三个核心环节。从那以后我每次改 WebSocket 相关的代码都强制自己先跑一遍断线重连和并发广播的测试确认异常路径都走通了再提交。希望帮到你。本文还有配套的精品资源点击获取