简介：这份资源提供基于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 库而不是原生 socket
2.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, sender=websocket) 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_exceptions=True ) for ws, res in zip(targets, results): if isinstance(res, Exception): connected_clients.pop(ws, None)这段代码里，connected_clients 用 websocket 对象本身做 key，简单直接。broadcast 里先过滤掉发送者，再用 gather 并发发送。return_exceptions=True 是关键，它让某个连接发送失败时不会抛异常中断整个 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( format=FORMAT, channels=CHANNELS, rate=RATE, input=True, frames_per_buffer=CHUNK ) 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( format=FORMAT, channels=CHANNELS, rate=RATE, output=True, frames_per_buffer=CHUNK ) 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?rate=16000&channels=1，服务端在 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, sender=websocket) 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), timeout=1.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_size：
async def main(): async with websockets.serve( handler, "0.0.0.0", 8765, max_size=2**20, # 单条消息最大 1MB ping_interval=20, # 每 20 秒发一次 ping 保活 ping_timeout=20 # 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_interval=20，客户端 websockets.connect 时也设置 ping_interval=20。如果中间有反向代理，还要确认代理的超时时间大于 ping 间隔。
5.4 现象：广播时某个客户端断开导致其他人也收不到
这是异常没捕获干净。broadcast 里如果直接 await ws.send(data)，某个连接抛 ConnectionClosed 异常会中断整个循环，后面的连接就收不到数据了。必须用 gather 加 return_exceptions=True，或者每个 send 单独 try/except。
5.5 现象：内存持续增长
connected_clients 字典只增不减。检查 handler 的 finally 块是否真的执行了。如果 handler 里有 while True 循环且没有退出条件，异常发生时可能不会走到 finally。另外确认没有在别处重复注册同一个连接，比如重连时旧连接没清理就注册了新连接。
6. 二次开发改造：从能跑到好用需要动哪些地方
这份代码跑通不难，但要用在真实项目里，有几个地方值得动手改。第一个是加参数协商。现在采样率、声道数是写死的，改成客户端连接时通过 URL 参数或首条控制消息上报，服务端记录每个连接的参数，转发时按目标连接的参数做重采样。重采样可以用 audioop.ratecv，Python 标准库自带，不用额外装包。
第二个是加音频编码。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_retries=5): for attempt in range(max_retries): try: return await websockets.connect(uri, ping_interval=20) 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 相关的代码，都强制自己先跑一遍断线重连和并发广播的测试，确认异常路径都走通了再提交。希望帮到你。
本文还有配套的精品资源，点击获取
