WebSocket 长连接千万级并发下的内存与连接管理一、一台 32GB 的服务器只能维护 20 万 WebSocket 连接按照每连接 50KB 内存的计算32GB 理论上能支撑 60 万连接。但实际压测中20 万连接时系统就开始 OOM Kill。差距从哪来不是 WebSocket 连接本身而是连接关联的上下文对象用户信息、订阅列表、心跳定时器、未发送消息队列。这些在实现时往往没有做内存预算一行this.users.set(userId, userObject)就悄悄复制了一份全量用户数据给每个连接。千万级并发 WebSocket 的工程挑战本质是每连接成本控制。如果能将每连接内存降到 10KB单机可以支撑 200 万连接。如果再引入多节点水平扩展千万级就可达成。二、WebSocket 连接的内存模型优化graph TD A[WebSocket 连接br/建立] -- B[轻量连接对象br/( 2KB)] B -- C[共享用户数据br/(不复制)] B -- D[会话状态br/(状态机, 500B)] B -- E{内存优化策略} E -- E1[延迟加载br/按需读取用户信息] E -- E2[对象池复用br/ByteBuffer 预分配] E -- E3[缓冲区零拷贝br/direct buffer] E -- E4[消息队列br/背压控制] F[用户数据存储] -- C G[Redis / Memcached] -- C style E fill:#4A90D9,color:#fff style E1 fill:#50B86C,color:#fff style E2 fill:#50B86C,color:#fff style E3 fill:#50B86C,color:#fff style E4 fill:#50B86C,color:#fff四个核心优化策略共享而非复制连接对象不持有用户对象的完整副本。用户信息存在 Redis 或本地 LRU Cache 中连接对象只保存userId引用。需要时通过userId查一次不需要时不占用内存。消息发送缓冲区池化WebSocket 写操作需要 ByteBuffer。如果每个连接都allocate(4096)10 万连接就是 400MB 仅用于空缓冲区。使用 Netty 的 PooledByteBufAllocator内存池化管理实际内存占用减少 80%。零拷贝转发消息从用户 A 的 WebSocket 到用户 B 的 WebSocket不要在应用层做byte[]拷贝。使用 Netty 的CompositeByteBuf或直接传递 ByteBuf 引用。背压控制如果消费者WebSocket 输出速度落后于生产者业务逻辑需要积累的消息会撑爆内存。用 FlushConsolidation 批量发送超过水位线则丢弃或降级。三、生产级高并发 WebSocket 实现连接管理器/** * WebSocket 连接管理器 * 千万级并发的核心最小化每连接内存占用 */ public class WebSocketConnectionManager { // 使用 Netty 的 FastThreadLocal 替代 ThreadLocal // Netty 的 FastThreadLocal 使用数组索引替代 HashMap性能更好 // 在高并发下减少 GC 压力 private static final FastThreadLocalbyte[] LOCAL_BUFFER new FastThreadLocal() { Override protected byte[] initialValue() { return new byte[512]; // 预分配小缓冲区 } }; // 连接存储ConcurrentHashMap 分段锁 // 为什么用 ConcurrentHashMap 而非 HashMap synchronized // ConcurrentHashMap 内部按 segment 分段锁 // 16 个 segment 意味着 16 个连接可以同时注册互不阻塞 private final ConcurrentHashMapString, WebSocketSession connections new ConcurrentHashMap(65536); // 预分配容量 // 连接数监控 private final AtomicLong connectionCount new AtomicLong(0); private final AtomicLong messageCount new AtomicLong(0); // 每个 IP 的最大连接数防滥用 private static final int MAX_CONNECTIONS_PER_IP 50; private final ConcurrentHashMapString, AtomicInteger ipConnectionCounts new ConcurrentHashMap(); /** * 注册新连接 * * 为什么返回 boolean 而非 void * 注册可能因各种原因失败(超限/重复) * 需要让调用方知道结果做相应处理 */ public boolean register(String userId, String ip, WebSocketSession session) { // 1. IP 级别限流 // 为什么要做 IP 限流而非仅用户级 // 防止一个 IP 下的恶意脚本创建大量连接耗尽服务器资源 AtomicInteger ipCount ipConnectionCounts.computeIfAbsent( ip, k - new AtomicInteger(0) ); if (ipCount.incrementAndGet() MAX_CONNECTIONS_PER_IP) { ipCount.decrementAndGet(); closeQuietly(session, IP 连接数超限); return false; } // 2. 去重检查同一个 userId 不能有多个连接 // 先检查如果存在旧连接关闭旧连接 WebSocketSession oldSession connections.get(userId); if (oldSession ! null oldSession.isOpen()) { // 关闭旧连接可能是之前的连接没有正常断开 closeQuietly(oldSession, 新连接替换); } // 3. 注册连接 connections.put(userId, session); connectionCount.incrementAndGet(); // 4. 绑定 session 关闭回调 // 为什么不在外部手动 remove // WebSocket 可能因网络断开、客户端关闭等多种原因断开 // 绑定 closeFuture 回调确保无论何种原因都能清理 session.closeFuture().addListener(future - { unregister(userId, ip, session); }); return true; } /** * 注销连接 */ private void unregister(String userId, String ip, WebSocketSession session) { connections.remove(userId, session); connectionCount.decrementAndGet(); // 释放 IP 配额 AtomicInteger ipCount ipConnectionCounts.get(ip); if (ipCount ! null) { ipCount.decrementAndGet(); // 清理如果计数归零移除 Map 的 key // 为什么需要清理 key // 长期运行后 ipConnectionCounts 会积累大量值为 0 的 key // 导致内存缓慢泄漏 ipConnectionCounts.remove(ip, 0); } } /** * 按用户 ID 发送消息 * * 为什么用异步发送 * 同步 send 在高并发下会阻塞 Netty 的 EventLoop 线程 * 导致该线程上的所有连接都受影响 */ public CompletableFutureVoid sendToUser( String userId, String message ) { WebSocketSession session connections.get(userId); if (session null || !session.isOpen()) { return CompletableFuture.failedFuture( new RuntimeException(用户不在线: userId) ); } // Netty 的 TextWebSocketFrame 是引用计数对象 // release() 时必须调用否则内存泄漏 TextWebSocketFrame frame new TextWebSocketFrame(message); // writeAndFlush 返回 ChannelFuture包装为 CompletableFuture ChannelFuture future session.writeAndFlush(frame); CompletableFutureVoid result new CompletableFuture(); future.addListener((ChannelFutureListener) f - { if (f.isSuccess()) { messageCount.incrementAndGet(); result.complete(null); } else { result.completeExceptionally(f.cause()); } }); return result; } /** * 广播给所有在线用户 * * 注意千万级连接下不要做广播 * 千万次 writeAndFlush 即使是空消息也会挤满 Netty 的 IO 线程。 * 广播应限制在直播间群组等小范围场景 */ public void broadcast(String message, int maxConnections) { int count 0; for (Map.EntryString, WebSocketSession entry : connections.entrySet()) { if (count maxConnections) break; WebSocketSession session entry.getValue(); if (session.isOpen()) { // 使用 write 而非 writeAndFlush // 批量 write 后统一 flush减少系统调用次数 session.write(new TextWebSocketFrame(message)); count; } } // 统一 flush 所有缓冲区 // FlushConsolidationThresholdNetty 默认 256 次 write 才 flush 一次 // 减少 flush 的系统调用次数但增加延迟 for (WebSocketSession session : connections.values()) { if (session.isOpen()) { session.flush(); } } } private void closeQuietly(WebSocketSession session, String reason) { try { if (session.isOpen()) { session.close( new CloseStatus(4000, reason) ); } } catch (Exception ignored) { } } public long getConnectionCount() { return connectionCount.get(); } }多节点扩展一致性哈希路由/** * 基于一致性哈希的多节点 WebSocket 集群 * * 为什么用一致性哈希而非简单取模 * 取模路由在节点增减时需要全量重新计算 * 一致性哈希只影响相邻节点减少连接迁移量 */ public class ConsistentHashRouter { private final TreeMapInteger, String ring new TreeMap(); private final HashFunction hashFunction Hashing.murmur3_32(); private final int virtualNodes; // 每个物理节点的虚拟节点数 public ConsistentHashRouter(ListString nodes, int virtualNodes) { this.virtualNodes virtualNodes; for (String node : nodes) { addNode(node); } } /** * 根据 userId 路由到目标节点 * * 为什么用 userId 而非随机 * 同一用户的连接必须固定路由到同一节点 * 否则消息投递需要通过节点间转发增加延迟 */ public String route(String userId) { int hash hashFunction.hashString( userId, StandardCharsets.UTF_8 ).asInt(); // 顺时针查找第一个 hash 的虚拟节点 Map.EntryInteger, String entry ring.ceilingEntry(hash); if (entry null) { // 没找到hash 超过了所有节点回到环的起点 entry ring.firstEntry(); } return entry.getValue(); } public void addNode(String node) { for (int i 0; i virtualNodes; i) { String virtualNode node # i; int hash hashFunction.hashString( virtualNode, StandardCharsets.UTF_8 ).asInt(); ring.put(hash, node); } } public void removeNode(String node) { for (int i 0; i virtualNodes; i) { String virtualNode node # i; int hash hashFunction.hashString( virtualNode, StandardCharsets.UTF_8 ).asInt(); ring.remove(hash); } } }四、高并发 WebSocket 的架构边界缺点广播操作是性能杀手在千万级连接下广播一条消息意味着千万次系统调用。必须限制广播范围如分组、频道而非全局广播。节点间消息转发是瓶颈用户 A 在 Node1用户 B 在 Node2A 发给 B 的消息需要节点间转发。转发链路的吞吐和延迟需要专门的内部消息总线。连接迁移的瞬态丢消息节点缩容时一致性哈希将连接重新分配。迁移期间正在投递的消息可能丢失。需要引入消息确认和重试机制。禁用场景消息可靠性要求极高如支付通知WebSocket 的 TCP 连接断开时缓冲区中未发送的消息会丢失。必须配合消息队列做持久化兜底。需要长时间离线消息保持用户离线一天后上线需要补推这期间的消息。这不是 WebSocket 的事是消息存储和推送的事。五、总结千万级 WebSocket 并发核心是四个控制每连接内存控制共享数据引用而非复制、消息缓冲区控制池化 零拷贝、发送背压控制批量 flush 水位线、连接路由控制一致性哈希避免全量迁移。单机瓶颈在内存和 CPU突破单机后瓶颈在节点间转发。不必追求全局广播消息投递应限定在群组房间等小范围用发布-订阅模式替代遍历循环。