Python asyncio事件循环在百万级WebSocket连接中的调度优化

在Python中处理百万级WebSocket连接时，标准的asyncio实现往往会因为调度开销、内存管理和文件描述符限制而崩溃。要达到这一量级，必须从底层的事件循环替换开始，逐步优化对象创建策略与操作系统参数。

第一阶段：替换高性能事件循环引擎

标准的CPython事件循环是基于纯Python实现的，在处理大量I/O切换时存在性能瓶颈。将其替换为基于Cython编写的uvloop是获得接近Go语言性能的第一步。

1. 执行命令安装 uvloop 库：

   在终端中运行以下命令：

   pip install uvloop

2. 修改入口文件代码以启用新引擎：

   在代码启动阶段，将默认的事件循环策略替换为uvloop。这步操作必须在任何异步操作开始之前完成。

   import asyncio
   import uvloop
   
   # 指定使用 uvloop 作为事件循环
   asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

3. 创建并运行主事件循环：

   使用标准接口获取循环并运行服务，此时底层已经由libuv驱动。

   loop = asyncio.new_event_loop()
   asyncio.set_event_loop(loop)
   
   # 假设 start_server 是你的WebSocket服务启动协程
   loop.run_until_complete(start_server())
   loop.run_forever()

第二阶段：消除内存碎片与GC压力

当连接数达到百万级时，每个连接对应的状态对象若频繁创建和销毁，会导致Python垃圾回收器（GC）不堪重负，造成服务卡顿。优化核心在于“对象复用”。

1. 定义连接类时使用 __slots__：

   默认的Python对象使用字典存储属性，内存占用大且访问慢。使用__slots__可以固定属性列表，减少约40%-60%的内存占用。

   class WebSocketConnection:
       __slots__ = ['ws', 'user_id', 'state', 'buffer']
   
       def __init__(self, ws, user_id):
           self.ws = ws
           self.user_id = user_id
           self.state = 'active'
           self.buffer = bytearray()

2. 构建简单的对象池：

   维护两个队列，一个用于存放空闲对象，一个用于存放活跃对象。当新连接建立时，从空闲队列取出；连接断开时，重置状态并回收到空闲队列。

   from collections import deque
   
   class ConnectionPool:
       def __init__(self):
           self.free_queue = deque()
           self.active_count = 0
   
       def acquire(self, ws, user_id):
           if self.free_queue:
               conn = self.free_queue.popleft()
               # 重置对象状态
               conn.ws = ws
               conn.user_id = user_id
               conn.state = 'active'
               conn.buffer.clear()
           else:
               conn = WebSocketConnection(ws, user_id)
           self.active_count += 1
           return conn
   
       def release(self, conn):
           self.active_count -= 1
           self.free_queue.append(conn)

3. 计算理论内存节省：

   假设单个对象原大小为 $1 \text{KB}$，优化后为 $0.5 \text{KB}$。对于 $N=10^6$ 个连接，内存占用变化量为：
   $$ \Delta M = N \times (Size_{old} - Size_{new}) = 10^6 \times 0.5 \text{KB} = 500 \text{MB} $$

第三阶段：优化事件循环的唤醒策略

默认情况下，asyncio每次收到数据都会唤醒对应的协程。在百万级并发下，如果频繁唤醒，CPU会消耗在上下文切换上。我们需要控制唤醒频率，实现“批量处理”。

1. 理解 recv 与 feed_data 的区别：

   大多数WebSocket库（如websockets或aiohttp）内部使用流式读取。我们需要调整流的读取缓冲区大小，减少小包唤醒次数。

2. 配置流的高水位标记：

   在传输层协议（Protocol）设置读取高水位，只有当数据积累到一定量（例如64KB）或超过特定时间时，才触发回调。

   # 示例：在自定义Protocol中设置
   class MyProtocol(asyncio.Protocol):
       def connection_made(self, transport):
           # 设置高水位，当缓冲区达到 64KB 时暂停读取
           transport.set_write_buffer_limits(high=64 * 1024)
           self.transport = transport

3. 使用 asyncio.gather 批量处理回调：

   在业务逻辑层，不要在一个连接收到消息后立即处理数据库操作。将消息放入内存队列，由单独的调度任务批量消费。

   async def batch_handler(message_queue):
       batch = []
       while True:
           # 每100条或每0.1秒处理一次
           msg = await message_queue.get()
           batch.append(msg)
           if len(batch) >= 100:
               await process_batch(batch)
               batch.clear()

第四阶段：调整操作系统内核参数

即使Python代码写得再好，如果Linux内核限制了文件描述符数量或端口复用能力，连接数依然上不去。

1. 编辑 /etc/sysctl.conf 文件：

   打开文件并添加或修改以下核心网络参数。这步操作允许系统处理更多的TCP连接和跟踪队列。

   # 允许系统打开的最大文件描述符数
   fs.file-max = 2097152
   
   # 增加TCP连接跟踪表大小，防止连接过多导致 "table full" 错误
   net.netfilter.nf_conntrack_max = 1048576
   net.nf_conntrack_max = 1048576
   
   # 优化TCP TIME_WAIT 状态，允许快速回收
   net.ipv4.tcp_tw_reuse = 1
   
   # 扩大TCP连接队列长度，防止突发流量导致丢包
   net.core.somaxconn = 65535
   net.ipv4.tcp_max_syn_backlog = 65535

2. 执行命令使配置立即生效：

   输入以下命令应用上述设置，无需重启机器。

   sysctl -p

3. 修改当前会话的文件描述符限制：

   Linux默认限制单个进程只能打开1024个文件。你需要将此限制提高到百万级。

   # 临时修改（仅当前终端有效）
   ulimit -n 1048576

   若要永久生效，编辑 /etc/security/limits.conf，添加：

   * soft nofile 1048576
   * hard nofile 1048576

优化效果对比

以下为优化前后的关键指标对比情况。

调度流程逻辑

以下展示了优化后的调度流程，核心区别在于引入了“对象池”和“批量处理”环节，减少了GC压力和CPU切换。