简介：这份资源提供基于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( format=FORMAT, channels=CHANNELS, rate=RATE, input=True, frames_per_buffer=CHUNK )这段代码初始化了 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_exceptions=True ) 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_exceptions=True保证某个客户端发送失败不会影响其他人。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-dev，macOS 上需要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，就是版本不匹配。我一般会锁定websockets>=11.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( format=FORMAT, channels=CHANNELS, rate=RATE, output=True, frames_per_buffer=CHUNK, output_buffer_size=CHUNK * 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. 二次开发：从能跑到好用，我踩过的那些坑
这套代码跑通之后，离真正能用还有一段距离。我第一次拿它做语音房的时候，以为改改端口就能上线，结果被现实狠狠教育了一顿。下面是我在二次开发中总结的几个关键改造点，按优先级从高到低排。
第一，加房间隔离。原始代码是所有客户端都在一个全局集合里，一个人说话所有人都能听到。实际场景肯定要分房间。改造方法是在服务端维护一个dict，key 是房间 ID，value 是该房间的客户端集合。客户端连接时先发一条文本消息声明房间号，服务端解析后把连接放到对应房间的集合里。转发时只遍历同房间的集合。这里要注意，客户端断开时要从房间集合里移除，如果房间空了就把 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_exceptions=True ) finally: if room_id and room_id in rooms: rooms[room_id].discard(websocket) if not rooms[room_id]: del rooms[room_id]第二，加音频编码压缩。原始代码传的是裸 PCM，16k 单声道 16 位下每秒 32KB，10 个人同时说话就是 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 收发框架跑一遍，确认基础链路没问题，再往上叠业务逻辑。这样即使后面出问题，也能快速定位是传输层还是业务层。希望这套代码和上面的经验能帮到你，少走一些我当年走过的弯路。
本文还有配套的精品资源，点击获取
