简介这是一份基于WebSocket实现语音实时收发功能的Python服务器与客户端代码主要面向需要构建长连接双向通信服务的开发人员可用于语音对讲、实时语音识别、在线会议等场景。压缩包共3个文件核心为服务器脚本与客户端脚本各一个另附一个备份文件整体体积仅9KB结构精简便于阅读和二次开发。这套代码已有3108人学习下载尤其适合具备Python基础、希望快速理解WebSocket全双工通信及语音数据传输流程的中级开发者。代码分别基于Flask-SocketIO与websocket-client库编写服务器端通过事件装饰器接收客户端消息客户端在连接建立后发送音频数据清晰展示了从建立连接、事件监听、数据传送到断开的完整链路同时备份文件可供对比修改方便调试。读者可直接运行示例验证效果并在此基础上扩展用户认证、音频编解码、实时识别或流式转发等功能为实际项目提供可落地的参考实现。 最近把WebSocket语音通道从零到一完整撸了一遍服务器端和客户端都用Python实现最终能在两台机器之间实时传语音。这套东西做完之后拿去接语音对讲、语音助手前端采集、甚至给GB28181那套对讲通道做语音转发都通用。整个链路不算复杂但里面坑确实不少比如PyAudio阻塞导致的事件循环卡死、音频格式不匹配产生杂音、网络抖动带来的延迟堆积这些我在后面都会逐个说清楚。如果你正好在纠结怎么给项目加一个实时语音收发的通道或者想搞一个局域网语音对讲Demo这篇可以直接作为参考。1. 为什么选WebSocket做语音实时传输1.1 先对比几种常见方案做语音传输之前我其实先在HTTP轮询、TCP裸Socket和WebSocket三者之间纠结过。HTTP轮询的思路是最直接的客户端隔几百毫秒去拉一次音频数据但这么做有两个硬伤。一个是实时性差轮询间隔再短也有延迟而且服务器只能被动等请求上行方向你得自己想办法推数据另一个是头部开销太大HTTP请求头动不动几百字节音频数据本身才几十字节传输效率低得没法看。做实时语音基本可以直接排除。TCP裸Socket则走向另一个极端你确实能拿到最高性能和最低延迟但得自己处理粘包、半包、心跳保活、连接状态维护这些事。Python里用socket实现一套可靠的消息帧协议工程量不小而且如果以后要让浏览器直接连TCP裸Socket完全行不通。WebSocket是中间的平衡点。它底层基于TCP天然全双工客户端和服务端可以同时收发数据协议层面每个消息都有明确的帧边界音频这种二进制流直接作为二进制帧塞进去不用担心粘包问题Python生态里有成熟的websockets库浏览器端也有原生WebSocket对象将来想从Python客户端换成Web页面客户端几乎不用改服务端逻辑。1.2 WebSocket的二进制帧优势语音数据本身就是二进制流WebSocket对二进制消息的支持是它最适合音频传输的核心原因。用WebSocket发音频核心就一行代码await websocket.send(audio_bytes)。底层会自动把字节流封装成二进制帧带上长度和类型信息发到对端。对端收到后也不用像TCP那样自己拼包、判断边界直接拿到的就是一个完整的音频块。另一个优势是文本帧和二进制帧可以混用且互不干扰。我在这套设计里就用文本消息传控制指令、二进制消息传音频数据。服务端只需判断消息类型就能决定是执行信令逻辑还是转发音频流。提示WebSocket毕竟是在应用层做转发每一条消息都会有一些帧头开销。实测下来对于语音这种实时业务把单帧音频控制在0.05到0.2秒是最合理的区间帧太短则网络开销占比太高帧太长则延迟感明显。2. 整体架构与音频数据协议2.1 架构与角色划分整个系统采用星型架构一个WebSocket服务器作为语音转发中心多个客户端连接上来后服务器把任何一个客户端发来的音频转发给其他所有客户端。这样天然支持点对点语音对讲和多方语音群聊。链路如下客户端A的麦克风采集音频 - A通过WebSocket二进制帧发送 - 服务器收到后选择其他客户端转发 - 客户端B的扬声器播放音频。反向同理B的声音也能传到A。服务器本身不碰音频内容纯转发好处是逻辑简单也能在某些场景下做集中式音频分析比如录制、转写、VAD检测等。2.2 音频格式怎么定收发双方必须提前约定音频格式这一步没做好后面全是杂音。我采用的是最通用的PCM格式具体参数如下参数设置说明编码格式PCM 16bit每个采样点2字节声道数1单声道语音不需要立体声采样率16000Hz语音识别常用的标准采样率单帧样本数1600对应0.1秒音频采样率选择16kHz是有讲究的。从信号采样定理来说采样率至少要是信号最高频率的两倍。人声的主要能量集中在300到3400Hz理论上8kHz采样率就够电话级通话了但16kHz已经是公认的“宽带语音”标准音节清晰度更高而且大量语音识别引擎、离线TTS引擎都推荐16kHz输入后续接识别模型省掉重采样这一步。单帧样本选1600意味着每帧时长是1600 / 16000 0.1秒。这个值我试过很多次0.1秒一帧是延迟和开销的平衡点网络里每秒钟只发10个包每个包约3200字节1600样本 * 2字节带宽占用才32KB/s基本不占什么资源。2.3 文本指令与音频数据的区分这套协议里我直接用消息类型区分指令和音频。WebSocket的文本消息走控制指令比如{cmd: hello, client: device-001}二进制消息一律当作音频帧。服务器端的判断逻辑很简单if isinstance(message, bytes): # 二进制消息按音频数据处理 await forward_audio(message) else: # 文本消息按信令处理 await handle_command(message)这种约定简单粗暴但非常实用。实际项目中你还可以扩展比如二进制消息的前四个字节存一个时间戳或序列号后面对齐网络抖动、做缓存缓冲时非常有用。我这次是先跑通基础版没加但架构上预留了扩展空间。3. 开发环境与依赖安装3.1 基础环境要求开发环境要求不高我用的是Python 3.10理论上Python 3.8以上都能跑。核心依赖就两个库websockets负责WebSocket网络通信pyaudio负责音频采集和播放。3.2 安装依赖pip install websockets pyaudio如果运气好一条命令就能装完。但如果是在Windows或者某些Linux发行版上pyaudio极大概率会安装失败原因几乎都是portaudio这个C库缺失。Windows环境下的解决办法是去PyAudio官网下载对应Python版本的whl文件或者直接用conda安装conda install pyaudioLinux环境先装系统依赖再装Python包sudo apt-get install portaudio19-dev python3-pyaudio pip install pyaudio安装成功后可以先跑一段代码确认设备可用import pyaudio p pyaudio.PyAudio() print(可用输入设备数量:, p.get_device_count()) for i in range(p.get_device_count()): info p.get_device_info_by_index(i) print(i, info.get(name), int(info.get(maxInputChannels)), 输入通道)这个输出能让你确认麦克风设备是否被识别输出设备编号也方便后面对多设备做选择。4. 服务端完整实现4.1 服务端核心逻辑服务器核心就三件事维护客户端连接集合、接收音频数据、广播转发给其他客户端。另外也需要处理客户端连接和断开时的状态清理。连接管理我用了一个set集合每个接入的WebSocket连接对象都放进去。为什么用set而不是list是为了断开时discard操作效率更高而且不会重复添加连接。广播转发时注意一个关键点不能把音频回传给发送者自己不然客户端会听到自己的声音形成正反馈啸叫。所以转发循环里要跳过来源websocket。4.2 服务端代码import asyncio import websockets # 保存所有活跃的WebSocket连接 connected_clients set() async def audio_server(websocket, pathNone): # 新客户端接入 connected_clients.add(websocket) addr websocket.remote_address print(f[连接] 客户端加入: {addr}, 当前连接数: {len(connected_clients)}) try: async for message in websocket: # 二进制消息 音频数据转发给其他客户端 if isinstance(message, (bytes, bytearray)): if not message: continue print(f[音频] 收到 {len(message)} 字节转发给其他客户端) for client in connected_clients: if client is not websocket: try: await client.send(message) except Exception as e: print(f[转发失败] {e}) # 文本消息 控制信令这里先打印 elif isinstance(message, str): print(f[信令] {message}) except websockets.exceptions.ConnectionClosed as e: print(f[断开] 客户端连接关闭: {addr}, code{e.code}, reason{e.reason}) except Exception as e: print(f[错误] {type(e).__name__}: {e}) finally: # 无论什么原因退出都要清理连接 connected_clients.discard(websocket) print(f[清理] 客户端 {addr} 移除, 当前连接数: {len(connected_clients)}) async def main(): # 绑定0.0.0.0允许局域网内其他机器连接 server await websockets.serve(audio_server, 0.0.0.0, 8765) print(WebSocket语音服务器已启动监听端口 8765) await server.wait_closed() if __name__ __main__: asyncio.run(main())4.3 服务端设计说明监听地址用0.0.0.0而不是127.0.0.1是为了让局域网内的其他设备也能连接。如果只用本机调试改回127.0.0.1就行。async for message in websocket是websockets库提供的最舒服的写法它内部帮你处理了消息帧的接收和拆包还会在连接断开时抛出ConnectionClosed异常。这个异常一定要捕获否则服务端进程容易崩。转发时逐客户端await client.send()这里有个隐藏问题如果某个客户端网络很差send会长时间阻塞拖慢整个转发循环。生产环境可以用asyncio.create_task做异步发送每个客户端一个发送任务。基础版为了代码可读性先同步转发我在开发时也是先跑通再优化避免一开始就陷入并发细节。5. 客户端完整实现5.1 音频采集与播放的细节客户端比服务端复杂因为它要同时处理两件事从麦克风采集音频并通过WebSocket发出去同时从WebSocket接收音频数据并通过扬声器播放。PyAudio是同步阻塞库采集和播放都是阻塞调用。最简单的写法是在两个异步协程里直接调用stream.read()和stream.write()但这会有大坑PyAudio的阻塞调用会卡住整个asyncio事件循环导致发数据和收数据互相阻塞整个链路瘫痪。我踩过这个坑之后找到了Python 3.9以上的标准解法——用asyncio.to_thread把阻塞的音频调用丢到线程池去执行这样事件循环始终不会被卡住。5.2 客户端代码import asyncio import websockets import pyaudio # 音频参数必须和服务端约定一致 FORMAT pyaudio.paInt16 # 16位PCM CHANNELS 1 # 单声道 RATE 16000 # 采样率 16kHz CHUNK 1600 # 每帧样本数0.1秒 # 建立采样流和播放流注意分别创建 p pyaudio.PyAudio() stream_in p.open( formatFORMAT, channelsCHANNELS, rateRATE, inputTrue, frames_per_bufferCHUNK ) stream_out p.open( formatFORMAT, channelsCHANNELS, rateRATE, outputTrue, frames_per_bufferCHUNK ) async def send_audio(websocket): print([采集] 开始采集麦克风音频并发送...) while True: # to_thread避免阻塞事件循环 data await asyncio.to_thread(stream_in.read, CHUNK) await websocket.send(data) async def receive_audio(websocket): print([播放] 开始接收服务器音频并播放...) async for message in websocket: if isinstance(message, (bytes, bytearray)): # to_thread避免播放阻塞接收循环 await asyncio.to_thread(stream_out.write, message) async def main(): uri ws://127.0.0.1:8765 print(f[连接] 正在连接服务器 {uri} ...) async with websockets.connect(uri) as websocket: print([连接] 已连接到服务器) # 并发执行发送和接收两个任务 send_task asyncio.create_task(send_audio(websocket)) recv_task asyncio.create_task(receive_audio(websocket)) await asyncio.gather(send_task, recv_task) if __name__ __main__: try: asyncio.run(main()) except KeyboardInterrupt: print([退出] 用户中断) finally: # 清理资源 stream_in.stop_stream() stream_in.close() stream_out.stop_stream() stream_out.close() p.terminate() print([清理] 音频资源已释放)5.3 为什么必须用to_thread处理阻塞调用这一步值得展开讲。asyncio是单线程事件循环同一时刻只能执行一个协程。如果send_audio里的stream_in.read()直接阻塞300毫秒等待麦克风数据那么整个事件循环就被卡住了receive_audio协程根本没有机会运行服务器发过来的音频全部积压在缓冲区里表现就是自己说话别人能听到但完全听不到对方说话。用asyncio.to_thread(stream_in.read, CHUNK)之后read()被放到线程池运行事件循环继续畅通两个任务才能真正并发。这是整个客户端实现里最核心的一个改动。如果你的Python版本低于3.9用loop.run_in_executor(None, stream_in.read, CHUNK)替代即可。6. 联调运行与实测记录6.1 本机回环测试先启动服务器python server.py正常输出如下WebSocket语音服务器已启动监听端口 8765再开一个终端启动客户端python client.py看到以下输出说明连接成功[连接] 正在连接服务器 ws://127.0.0.1:8765 ... [连接] 已连接到服务器 [采集] 开始采集麦克风音频并发送... [播放] 开始接收服务器音频并播放...此时服务器端会打印[连接] 客户端加入: (127.0.0.1, 56789), 当前连接数: 1 [音频] 收到 3200 字节转发给其他客户端因为只有一台客户端服务器不会把它自己的音频回传给它所以麦克风说话时不会在扬声器听到回声这是符合预期的。6.2 双机对讲测试用两台机器或者同一台机器开两个客户端进程连到同一个服务器。服务器收到任意一方的音频后会转发给另一方。实测在局域网环境下口语交流的延迟体感很低基本接近对讲机效果。注意同一台机器跑两个客户端时会产生严重的回声啸叫因为扬声器声音会再次被麦克风采集。建议测试时戴耳机或者在代码里让服务器只做单点转发避免形成音频环。6.3 局域网连接注意客户端连接服务器时uri要写成ws://服务器局域网IP:8765而不是ws://127.0.0.1:8765。服务器需要放行8765端口Windows和Linux防火墙都可能拦截非本机连接。最简单的方式是临时关闭防火墙测试验证通了之后再添加放行规则。7. 常见问题与排查技巧7.1 PyAudio安装失败现象pip install pyaudio报错编译C扩展失败。原因系统缺少portaudio C库。Windows解决下载对应Python版本的whl安装或者用conda装。Linux解决先apt-get install portaudio19-dev python3-pyaudio再装。7.2 能连接但说话没声音现象WebSocket连接正常服务器也能看到音频数据进入但对方听不到。可能原因麦克风设备默认音量被调成0或者输入的音频参数与播放端不一致。排查先用系统的录音软件测试麦克风是否正常再打印两端的RATE、CHANNELS、FORMAT确认一致最后检查播放端设备是否选了正确的输出设备。7.3 有声音但全是刺耳杂音现象音频能传过去但播放出来是持续的嘶嘶声或爆音。原因采样率不匹配比如一个用44100录、一个用16000播、位深不匹配、或者单双声道不一致。解决把两端音频参数强制统一尤其是采样率和声道数。我在开发时曾经因为一个客户端用了默认设备的44100Hz另一个用16000Hz导致对面全是变调的娃娃音排查了半天才发现是参数不一致。7.4 延迟越来越大语音越来越卡现象刚开始对讲正常一分钟后延迟越来越明显。原因播放端stream_out.write()写入速度跟不上网络到达速度音频帧在缓冲区里堆积。解决一是调大播放端的frames_per_buffer提高播放缓冲二是在receive_audio里加一个简单的丢弃逻辑比如如果发现每次循环拿到的消息已经积压就只播放最新的几帧。实时语音系统里丢帧永远比延迟堆积更可取。7.5 连接后立刻断开现象客户端连接上服务器后几秒内断线服务器打印ConnectionClosed。原因典型原因是网络不稳定、服务器负载高、或者触发了WebSocket的ping/pong超时机制。解决在客户端和服务端之间加心跳。websockets库有内置的ping_interval和ping_timeout参数websockets.connect(uri, ping_interval20, ping_timeout20)服务器端也设置对应的值可以有效检测死连接。7.6 事件循环卡死收发不同步现象客户端要么只能发不能收要么只能收不能发。原因在协程里直接调用了阻塞的stream.read()或stream.write()卡住事件循环。解决所有PyAudio的阻塞操作都用asyncio.to_thread或run_in_executor包一层这是整个系统能否并发工作的关键。最后再分享一点个人体会。这套代码跑通以后我最大的感觉是实时语音系统的难点从来不在语法层面而在工程细节。音频参数的定义、阻塞调用的处理、缓冲的堆积这些坑不实际跑一遍根本发现不了。如果让我重新做一次我可能会在一开始就加上时间戳语音帧的设计这样后面做音频对齐、录制回放、VAD检测都会省很多事。另外就是测试环境一定不要只在本机跑拿到真实的局域网甚至无线网络里测一轮延迟和丢包的表现会给你很多预料之外的收获。本文还有配套的精品资源点击获取
