高并发WebSocket服务：基于协程与asyncio的完整实战指南

1. 为什么WebSocket服务要选协程：这笔账得算清楚

1.1 WebSocket的长连接特性彻底改变了并发模型

做WebSocket服务，第一反应是用什么并发方案？我见过不少团队纠结这个问题，最后选了线程池，每一个连接配一个线程。开发阶段没什么感觉，联调时候也能跑，等模拟了上千个在线客户端，服务端CPU就发疯了，内存也一路往上飙。

这不是线程不好，而是WebSocket这种长连接模型，天生和线程模型拧着来。HTTP请求是“请求-响应-结束”，一个请求处理完，线程就释放了，用线程池配合短连接非常合理。但WebSocket建立之后，这条连接是长时间驻留的，客户端不会频繁断开。服务器为了等一条连接上的消息，就必须一直“盯着”这条连接。如果用一连接一线程的模式，一个线上1万在线用户，服务端就要起1万条线程。这个账，光算线程栈的内存就吓人。

所以我后来把整套服务用Python协程重写了一遍，用asyncio事件循环加websockets库。一个进程稳稳扛住几千条WebSocket连接，消息转发延迟能做到毫秒级。这篇文章就围绕“基于协程的并发WebSocket通信实践”展开，把选型思路、核心代码、压测方法、坑点排查全部过一遍。适合已经会一点asyncio语法、但没系统做过长连接服务的人参考，也适合正在纠结技术选型、准备从线程模型迁过来的团队。

1.2 线程模型在长连接场景下的真实成本

线程的成本主要在三块，第一块是内存。操作系统为每个线程预留独立栈空间，默认栈大小在Linux上是8MB的虚拟内存。虽然虚拟内存不等于物理内存，但1万条线程，光栈地址空间就是80GB的虚拟地址占用，这还没算线程控制块、内核资源。连接数一旦上去，系统先扛不住的不是CPU，而是资源上限。

第二块是切换开销。线程是由操作系统内核来调度的，线程切换涉及用户态内核态切换、寄存器保存恢复、调度器逻辑。一次切换可能只有微秒级，但高并发下频繁切换，CPU时间大量花在调度而不是业务逻辑上，这是很亏的。

第三块是Python的GIL限制。Python的线程不能真正并行执行CPU密集任务。你开再多线程，同一时刻只有一个线程在跑Python字节码。WebSocket场景虽然主要是IO密集，但计算密集部分依然会被GIL卡脖子，多线程提升十分有限。

我踩过一次很深的坑：某项目用线程处理WebSocket消息，每次客户端连进来就新建一条线程，在线5000的时候，服务端直接无响应。后来查系统日志，线程数飙到6000多，上下文切换每秒几十万次，整个机器像被拖死了一样。从那以后，我对长连接服务基本只用协程方案。

1.3 协程调度：把网络等待时间“挤”出来

协程解决长连接问题的思路完全不一样。它不要求每个连接一个线程，而是只用一个线程，在这个线程里跑一个事件循环。所有连接被拆成一个个协程任务，当任务遇到网络IO需要等待时，它主动通过await挂起自己，把执行权交还给事件循环。事件循环这时候就去执行其他就绪的任务，比如处理另一条连接上到的消息。

这里的关键是：等待网络IO的时间，不再被白白浪费掉。CPU在等待期间转去干别的活了。协程间的切换是纯用户态的，函数调用级别的状态保存与恢复，不需要内核参与，切换开销可以做到极低。

用一个生活化的类比来说：线程模型像每桌客人配一个专属服务员，服务员站在旁边等客人点菜，客人不说话他就干等着。协程模型像一个服务员同时服务很多桌，点完一桌的菜，马上跑去做别的，客人下个指令再回来处理。服务员的数量没有增加，但接待能力翻了几倍。

1.4 协程、线程、进程：三张并发方案的对比表

做选型的时候，我习惯列一张对照表，把三个维度的关键参数摆出来。这里直接把我自己常用的表格放出来。

看完这张表，为什么长连接场景选协程就很清楚了。WebSocket服务是典型的IO密集、高并发、连接长驻场景，这三个特征正好都落在协程的优势区里。进程模型在Python里更适合真正需要多核并行计算的任务，线程模型适合短请求或中小规模的并发。长连接这块，协程是性价比最高的那个。

2. 动手前的准备：事件循环、async语法与websockets库选型

2.1 Python协程的最小理解框架

在写代码之前，我先把Python协程的底层框架讲明白。如果你已经吃透了asyncio，可以直接跳到2.2节，但没吃透的话，先花十分钟把这几个概念理一遍，后面写代码会顺畅很多。

Python协程的语法核心就是async def和await。你用async def定义一个函数，这个函数调用时不会立即执行，而是返回一个协程对象。真正执行它需要把它交给事件循环，或者在其他协程里用await驱动。await的作用是挂起当前协程，直到等待的异步操作完成。这里有一个新手最容易卡住的点：await不是“等一会再继续”的意思，它是“先暂停我自己，把控制权还回去，等结果准备好了再回来继续”。事件循环会负责在未来的某个时刻恢复这个协程。

事件循环是整个异步系统的调度中枢。它维护一个任务队列，循环往复地检查哪些任务有结果了、哪些任务可以恢复执行。asyncio库把这个调度器实现了，你不需要自己造轮子，只需要把协程任务交给它。一个task就是事件循环里的一个执行单元，等于一条连接、一个后台任务之类的东西。事件循环每迭代一轮，会处理一批已经就绪的事件回调，然后继续循环。

还有个概念值得一提：asyncio.Queue和asyncio.Semaphore这些同步原语。很多新手以为协程是单线程的就不需要并发控制了，这个说法对了一半。单线程模型下，你不需要像多线程那样担心两个线程同时改一个变量，因为同一时刻只有一个协程在运行。但在一个协程内部await挂起之后，其他协程会插进来执行，如果多个协程都在往同一个列表、字典里写数据，依然可能出现逻辑上的交错问题。后面第5章我会专门讨论这个问题怎么处理。

2.2 websockets库为什么是首选

Python做WebSocket的库其实不少，常用的有websockets、aiohttp、FastAPI自带的WebSocket支持、还有gunicorn搭daphne跑ASGI这一套。我的经验是，如果目的是搞清楚并发WebSocket的核心机制，做底层可控的项目，websockets库是最直接的。

原因有三个：第一，它本身构建在asyncio之上，天生就是用协程写业务逻辑，不需要做什么适配；第二，API面向WebSocket协议本身做了精简，握手、帧解析、Ping/Pong这些都由库处理，你只需要关心连接建立后的收发逻辑；第三，它的并发模型默认就是“一个连接一个协程任务”，这正好满足我们这篇文章的主题。

aiohttp也能做WebSocket，但它是个完整HTTP框架，你只是想要一个WebSocket长连接服务，引入它的额外复杂度不值得。FastAPI就更不用说了，它适合的是快速开发带HTTP接口加WebSocket的业务系统，如果用它的WebSocket，底层还是调用的websockets库，只是多了层Starlette封装。学透底层的websockets库，再去看FastAPI的WebSocket接口，几乎是零成本。

2.3 项目目录与依赖

我建议你按下面的结构建工程，后面所有代码都围绕这个目录来写：

async_ws_demo/ ├── server.py # WebSocket服务端入口 ├── client.py # 测试客户端 ├── connection_manager.py # 连接管理器 ├── requirements.txt └── benchmark.py # 压测脚本

依赖其实就一个：websockets库。安装命令我直接给出来。

pip install websockets

当前我用的是websockets 12.x版本，API和早期版本有一点差别，主要体现在websockets.serve的返回对象类型上。新版更推荐用async with websockets.serve(...) as server:这样的上下文管理器写法，但直接调用也没问题。装完之后，用下面这段验证一下版本，同时确认基础环境可用：

import websockets import asyncio print(websockets.__version__) print(asyncio.get_event_loop_policy())

3. 并发WebSocket服务的完整落地实现

3.1 服务端基础骨架：注册连接与消息循环

先写一个最简单的服务端，功能是：客户端连上来之后发一条消息作为用户名，服务端接收后打印，然后把消息广播给所有在线客户端。这个骨架是所有更复杂功能的基础。

import asyncio import websockets async def handler(websocket): username = await websocket.recv() print(f"{username} 已连接") async for message in websocket: print(f"[{username}] {message}") await websocket.send(f"服务器收到: {message}") async def main(): async with websockets.serve(handler, "0.0.0.0", 8765): print("WebSocket服务已启动: ws://0.0.0.0:8765") await asyncio.Future() if __name__ == "__main__": asyncio.run(main())

这段代码里最关键的是async for message in websocket这个循环。它是websockets库提供的最常用的消息接收方式，相当于一个无限循环，每来一条消息，就进入循环体处理一次。websocket.send()是异步方法，发送大消息时需要等待写入完成，所以调用的时候和接收消息一样，得用await。

asyncio.Future()在这里的作用是让事件循环永远跑下去，不会因为main协程结束就退出。如果你跑起来发现程序闪退，大概率就是这里写错了——直接写await asyncio.sleep(9999)虽然也能凑合，但远不如挂一个永不完成的Future干净。

3.2 连接管理器：集中管理所有在线连接

基础骨架只能做点对点通信，要实现“一个人发言，所有人可见”的广播效果，必须先有一个地方记录当前所有在线的WebSocket连接。我习惯抽一个ConnectionManager类来做这件事。

# connection_manager.py import websockets class ConnectionManager: def __init__(self): self.connections = {} async def connect(self, websocket, username): self.connections[username] = websocket await self.broadcast(f"{username} 加入了聊天室") def disconnect(self, username): if username in self.connections: del self.connections[username] print(f"{username} 已断开，当前在线: {len(self.connections)}人") async def broadcast(self, message): for ws in list(self.connections.values()): try: await ws.send(message) except websockets.ConnectionClosed: pass

这个类在多处代码里都用得上。这些地方可以重点看看：connect里先把自己注册进去，再广播一条加入消息，其他人立刻能看到这个用户上线了，注意广播的时候要先把values转成list，否则往dict里增删元素会报RuntimeError: dictionary changed size during iteration；disconnect里用del而不是pop，逻辑更明确；broadcast里捕获ConnectionClosed异常，某个连接断了不至于让整个广播流程崩溃。

服务端入口和这个管理器配合起来就是这样：

# server.py import asyncio import websockets from connection_manager import ConnectionManager manager = ConnectionManager() async def handler(websocket): username = await websocket.recv() await manager.connect(websocket, username) try: async for message in websocket: print(f"[{username}] {message}") await manager.broadcast(f"[{username}] {message}") except websockets.ConnectionClosed: pass finally: manager.disconnect(username) await manager.broadcast(f"{username} 离开了聊天室") async def main(): server = await websockets.serve( handler, "0.0.0.0", 8765, ping_interval=20, ping_timeout=20, max_size=1_048_576, ) print("WebSocket服务已启动: ws://0.0.0.0:8765") await server.wait_closed() if __name__ == "__main__": asyncio.run(main())

这里有个很容易踩的坑：websocket.recv()如果客户端一直没有发消息，这个调用会一直挂起，但ping_interval会自动在后台发送Ping帧。我设置ping_interval=20和ping_timeout=20，意思是每20秒发一次Ping，如果20秒内没收到Pong，就判定连接已断开。这个参数在后面心跳章节还会详细聊。

3.3 广播消息的实现与慢连接问题

上面那个简单的broadcast实现，有一个隐藏的性能问题：如果其中一个客户端网络很差，await ws.send()可能会卡很久，广播就被这个慢客户端拖住了。

一个简单的改进办法是用并发发送代替逐个等待：

import asyncio import websockets async def broadcast(self, message): if not self.connections: return tasks = [] for ws in list(self.connections.values()): tasks.append(asyncio.create_task(self._safe_send(ws, message))) await asyncio.gather(*tasks, return_exceptions=True) async def _safe_send(self, ws, message): try: await ws.send(message) except websockets.ConnectionClosed: self.disconnect_from_ws(ws)

这样每条消息的发送都是独立任务，快连接不会被慢连接阻塞。不过也要注意，这不是银弹。如果每发一条广播消息就创建了一万个协程任务，任务本身的创建和销毁也有开销。消息频率高的时候，可以进一步改成每个连接一个发送队列，广播只往队列里丢消息，每个连接由独立的发送协程消费自己的队列。这样广播操作本身是O(1)的。如果你自己写的广播服务到了每秒几千条消息的阶段，我会建议往这个方向改造；如果是几百人的聊天室，用gather的方式已经绰绰有余。

3.4 心跳保活与断线检测

做WebSocket服务最怕的一种情况是：客户端断电、网络断了，但TCP连接没有正常关闭，服务端傻等半天也不知道对方已经没了。WebSocket协议提供了Ping/Pong帧来解决这个问题，websockets库也把这一层封装好了。

开启ping_interval之后，库会在后台自动发送Ping帧。如果超过ping_timeout还没收到Pong，库会直接抛出一个ConnectionClosedError异常，你只需要在handler里捕获异常，然后做清理工作就行。这个机制极大减少了僵尸连接的数量。

如果你希望业务层也能检测到断线，比如记录用户最后一次活跃时间，可以在业务逻辑里自己维护一个心跳。做法并不复杂：用一个后台任务定时发送自定义的心跳消息，另一端收到后回复。这样做的代价是消息量会变大，但如果业务上需要感知在线状态，这依然是个好用且简单的方案。

async def heartbeat(websocket, interval=30): while True: try: await websocket.send("__ping__") await asyncio.sleep(interval) except websockets.ConnectionClosed: break

用的时候在handler中创建这个后台任务：

async def handler(websocket): hb_task = asyncio.create_task(heartbeat(websocket)) try: async for message in websocket: ... finally: hb_task.cancel()

注意一定要在finally里取消这个后台任务，否则handler结束了，heartbeat协程还在睡眠等待，下次会给一个已关闭的WebSocket发消息，喷出一堆异常。

3.5 优雅关闭：别让连接断得随随便便

开发期可以直接Ctrl+C把服务停了，但生产环境要保证服务停止时能先通知所有客户端，再关闭连接，最后清理资源。websockets库提供了优雅关闭的接口。

async def main(): stop_event = asyncio.Event() server = await websockets.serve( handler, "0.0.0.0", 8765, ping_interval=20, ping_timeout=20, ) loop = asyncio.get_running_loop() loop.add_signal_handler(signal.SIGINT, stop_event.set) loop.add_signal_handler(signal.SIGTERM, stop_event.set) await stop_event.wait() server.close() await server.wait_closed()

add_signal_handler这个API只能在主线程的事件循环中使用，它把信号注册到事件循环上，而不是用传统的signal信号处理函数。这样收到SIGINT或SIGTERM后，stop_event被设置，主协程从await stop_event.wait()返回，紧接着调server.close()开始关闭流程。wait_closed()等待所有连接的处理协程收尾。这样能确保不会出现连接被半路硬断的情况。

4. 并发压测与性能调优实录

4.1 用协程客户端模拟高并发连接

写完服务端，必须验证一下它到底能扛多少连接。我写了一个简单的压测脚本，它用协程客户端发起大量连接，每个客户端发送固定条数的消息，统计总耗时。

import asyncio import time import websockets async def one_client(index, sem): async with sem: ws = await websockets.connect("ws://127.0.0.1:8765") await ws.send(f"user_{index}") await ws.recv() start = time.time() for _ in range(100): await ws.send(f"hello from {index}") await ws.recv() elapsed = time.time() - start await ws.close() return elapsed async def main(): total = 200 sem = asyncio.Semaphore(200) start = time.time() results = await asyncio.gather( *(one_client(i, sem) for i in range(total)) ) total_time = time.time() - start print(f"总耗时: {total_time:.2f}s") print(f"平均单客户端耗时: {sum(results) / total:.3f}s") asyncio.run(main())

这里用asyncio.Semaphore控制并发创建连接的速度，避免一瞬间把所有连接都压上去。实际压测的时候，我建议你先从100个并发开始，慢慢往上加，一点一点摸出服务端的能力上限。

4.2 文件描述符限制与系统参数调整

压测到一个临界点，你大概率会看到这么个错误：

OSError: [Errno 24] Too many open files

这是操作系统的文件描述符上限被触发了。Linux下每个连接都会占用一个fd，单个进程默认的fd上限通常是1024，你开一千多个连接就直接爆了。解决方法不复杂，一个是改系统级的limits.conf，一个是给进程设置更高的ulimit。

ulimit -n 65535

如果想让某个用户的终端默认生效，可以在/etc/security/limits.conf里加配置。改完这个，压测脚本和服务端进程都需要重新启动。另外一个和WebSocket连接数强相关的系统参数是fs.file-max，它决定整个系统能打开的最大文件数。如果连ulimit都调上去了还是报错，检查一下这个参数。

4.3 uvloop：给事件循环加速

websockets库本身基于asyncio，而asyncio的事件循环是纯Python实现的。有一个库叫uvloop，它把事件循环底层换成了libuv实现，也就是Node.js使用的那个事件循环库。换装很简单：

pip install uvloop

然后在入口处加两行：

import uvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

在Linux上，uvloop对事件循环的吞吐提升是很明显的，尤其在大量连接、大量事件回调的场景下，实测下来性能提升通常在20%到50%之间。不过要注意，uvloop目前不支持Windows，Windows上需要用asyncio自带的ProactorEventLoop。如果服务部署在Linux服务器上，强烈建议加上。

4.4 消息量与缓冲区控制

压测过程中我发现一个有意思的现象：当服务端广播大量消息时，内存会快速上涨。原因在于websockets库内部为每条发送消息维护了一个队列，如果某个客户端消费跟不上，消息就会堆积在内存里。

控制的办法有两个参数比较关键。一个是max_queue，它限制每个连接接收消息的队列最大长度，超过之后，新消息会被丢弃或触发流控。另一个是max_size，限制单条WebSocket消息最大字节数，防止有人故意发超大消息把内存打爆。

server = await websockets.serve( handler, "0.0.0.0", 8765, max_queue=32, max_size=2 ** 20, )

我给自己的项目设置的是max_queue=32，max_size=1MB，这两个值对一般业务足够了。如果业务里有大文件传输，可能需要调高max_size，但要谨慎，因为一个恶意的大消息请求就可能导致内存峰值飙升。

5. 常见问题排查与避坑速查

5.1 事件循环关闭时疯狂报RuntimeError

这是我被问过最多的问题。典型报错是这个：

RuntimeError: Event loop is closed

场景通常是：脚本里先调用了asyncio.run()跑了一个协程，协程结束之后，事件循环关闭了，但你之前创建的某个后台任务还在引用它。比如你把一个异步任务放在了全局变量里，第二次调用asyncio.run()时，那个任务还试图在已经关闭的事件循环上运行，于是报错。

解决思路很简单：不要在asyncio.run()之外残留任务。所有后台任务都要在同一个主协程生命周期内创建、运行、取消。如果确实需要全局生命周期管理，就把整个服务包在同一个事件循环里，比如用asyncio.run(main())这种方式，让main协程一直跑到底。

5.2 僵尸连接：连接数只增不减

我见过一些人把连接加到manager里之后，忘了在finally里移除，结果在线人数越涨越高，最后服务卡死。这里给一个排查技巧：在manager的connect和disconnect里打日志，记录当前总连接数。一旦发现连接数只增不减，基本可以确定是清理逻辑漏了。

另外要注意WebSocket断开时，async for循环可能正常退出，也可能抛异常。正常退出意味着客户端主动发了Close帧，抛异常意味着网络异常。不管哪种情况，清理逻辑都应该写在finally里，而不是让它在try或except分支里，这样最保险。

5.3 协程单线程，就没有并发问题了吗

很多人觉得协程是单线程，共享状态完全不需要保护，这个认知是错的。单个Python线程同一时刻确实只执行一个协程，但在await挂起的瞬间，控制权会切换到其他协程。如果两个协程交替修改同一个字典，可能产生逻辑交错。

举一个具体例子：两个协程同时执行connections[username] = websocket，由于dict赋值是原子的，不会出现两个线程同时写导致的数据损坏，但如果你先检查再操作，检查之后被挂起，另一个协程插入同样的key，等你恢复执行再写，就可能覆盖掉别人的数据。所以需要临界状态的时候，用asyncio.Lock来做互斥，而不是完全裸奔。

lock = asyncio.Lock() async def add_connection(username, ws): async with lock: if username in self.connections: await ws.close(code=4001, reason="用户名已存在") return self.connections[username] = ws

5.4 客户端断线重连策略

客户端不会一直在线，网络抖动或服务端重启都会导致连接断开。我习惯在客户端封装一个带指数退避的重连逻辑，避免服务端重启后所有客户端同一瞬间疯狂重连，造成连接风暴。

async def connect_with_retry(url, max_retries=5): delay = 1 for attempt in range(max_retries): try: return await websockets.connect(url) except (OSError, websockets.InvalidStatus) as exc: print(f"连接失败({attempt + 1}/{max_retries})，{delay}s后重试") await asyncio.sleep(delay) delay *= 2 raise RuntimeError(f"重试{max_retries}次后仍无法连接")

指数退避的意思是：第一次失败等1秒，第二次等2秒，第三次等4秒，以此类推。这样能快速恢复连接，又不会瞬间制造大量请求冲击服务端。

5.5 常见异常场景速查表

把这些常见场景记熟了，线上出问题能少走很多弯路。我个人的经验是，WebSocket服务排查问题，先看连接状态，再看消息队列，最后才看业务逻辑。连接状态有问题，后面所有排查都是白费功夫。

最初那套线程方案，后来我彻底没有再用过。协程方案用顺手之后，最大的感受是：代码量没有变多，但能支撑的连接数翻了几十倍。如果你正在从线程模型往协程模型迁移，不用急着一步到位，可以先写一个连接管理器，把连接注册、广播、心跳这些基础设施搭好，然后逐步把业务逻辑搬过来。踩过几次坑之后你会发现，asyncio这套模型其实非常适合WebSocket的长连接场景，迁移过程中学到的那些概念，以后写任何高并发IO服务都用得上。