后端技术22-从轮询到WebSocket：实时通信的性能提升100倍，10万并发在线！WebSocket聊天室的设计与实现

「知识图谱生成工具」一键将文件夹内容变身为交互式知识图谱的免安装桌面工具文末附免费下载链接-CSDN博客还在为消息延迟发愁还在用定时轮询折磨服务器这篇文章带你从协议原理到分布式实战彻底搞懂WebSocket。目录开篇那些年我们踩过的实时通信坑一、WebSocket协议原理一次握手终身朋友二、WebSocket vs 长轮询 vs SSE选谁三、实战Spring Boot WebSocket构建聊天室四、分布式场景Redis Pub/Sub实现多节点消息同步五、性能优化支撑10万并发的秘密六、源码获取与思考题开篇那些年我们踩过的实时通信坑还记得第一次做聊天功能的时候我信心满满地写了个AJAX轮询setInterval(() { fetch(/api/messages).then(res res.json()).then(updateUI); }, 1000);当时觉得完美每秒刷新一次用户肯定感觉不到延迟。直到产品上线第三天服务器CPU飙到90%数据库连接池被打爆老板拿着监控截图问我“为什么我们的小聊天室能把8核16G的服务器干趴下”那一刻我才明白轮询不是实时通信是服务器杀手。今天这篇文章我会把这些年踩过的坑、学到的经验全部倒给你。从协议原理到分布式架构从代码实战到性能优化保证你看完能直接上手搭建一个能支撑10万并发的聊天系统。目标很明确消息延迟 100ms并发连接 10万。坐稳了发车。一、WebSocket协议原理一次握手终身朋友1.1 握手过程HTTP的伪装WebSocket最妙的地方在于它借用了HTTP来完成握手然后 hijack 了这条连接。客户端 服务器 | | | GET /chat HTTP/1.1 | | Host: server.example.com | | Upgrade: websocket | | Connection: Upgrade | | Sec-WebSocket-Key: dGhlIHNhbXBsZQ | | Sec-WebSocket-Version: 13 | |----------------------------------------| | | | HTTP/1.1 101 Switching Protocols| | Upgrade: websocket | | Connection: Upgrade | | Sec-WebSocket-Accept: s3pPLMBi | |----------------------------------------| | | | WebSocket连接建立成功 | | |关键点Upgrade: websocket告诉服务器“我要升级协议”Sec-WebSocket-Key是客户端生成的Base64随机数服务器用魔法字符串258EAFA5-E914-47DA-95CA-C5AB0DC85B11拼接后SHA1哈希再Base64编码返回1.2 帧格式轻量级的二进制协议握手完成后数据以**帧(Frame)**的形式传输0 1 2 3 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 -------------------------------------------------------- |F|R|R|R| opcode|M| Payload len | Extended payload length | |I|S|S|S| (4) |A| (7) | (16/64) | |N|V|V|V| |S| | (if payload len126/127) | | |1|2|3| |K| | | ------------------------- - - - - - - - - - - - - - - - | Extended payload length continued, if payload len 127 | - - - - - - - - - - - - - - - ------------------------------- | |Masking-key, if MASK set to 1 | -------------------------------------------------------------- | Masking-key (continued) | Payload Data | -------------------------------- - - - - - - - - - - - - - - - : Payload Data continued ... : - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - | Payload Data continued ... | ---------------------------------------------------------------核心字段解释字段说明FIN是否为最后一帧opcode0x1文本帧, 0x2二进制帧, 0x8关闭, 0x9ping, 0xApongMASK客户端必须置1服务端必须置0Payload len数据长度0-125直接表示126后续2字节127后续8字节1.3 心跳机制保活与检测TCP连接会超时NAT会断开空闲连接。WebSocket通过Ping/Pong帧来维持心跳客户端 服务器 | | |--- Ping帧 (opcode0x9) -| | | |- Pong帧 (opcode0xA) ---| | | | (每隔30-60秒一次) |最佳实践客户端发送Ping服务端回复Pong超时时间通常设置为60-120秒连续3次未收到Pong认为连接断开二、WebSocket vs 长轮询 vs SSE选谁做实时通信除了WebSocket还有长轮询和SSE。三者怎么选┌─────────────────────────────────────────────────────────────┐ │ 实时通信方案对比 │ ├──────────────┬─────────────┬─────────────┬──────────────────┤ │ 特性 │ 轮询 │ 长轮询 │ SSE/WebSocket │ ├──────────────┼─────────────┼─────────────┼──────────────────┤ │ 实时性 │ 差(秒级) │ 较好 │ 极好(毫秒级) │ │ 服务器压力 │ 极高 │ 高 │ 低 │ │ 兼容性 │ 最好 │ 好 │ WebSocket需IE10│ │ 双向通信 │ 否 │ 否 │ WebSocket支持 │ │ 实现复杂度 │ 简单 │ 中等 │ 中等 │ └──────────────┴─────────────┴─────────────┴──────────────────┘2.1 轮询(Polling)简单但暴力// 客户端每秒请求一次 setInterval(() { fetch(/api/messages).then(updateUI); }, 1000);问题大量无效请求90%返回空数据消息延迟0-1000ms不等服务器QPS爆炸2.2 长轮询(Long Polling)改进版function longPoll() { fetch(/api/messages/wait).then(res { updateUI(res); longPoll(); // 递归继续 }); } longPoll();服务端hold住连接有新消息才返回。问题仍然需要频繁建立/断开HTTP连接服务器需要维护大量挂起连接消息量大时退化为普通轮询2.3 SSE(Server-Sent Events)服务器推送const eventSource new EventSource(/api/stream); eventSource.onmessage (e) updateUI(e.data);适用场景单向推送股票行情、新闻feed需要自动重连和事件ID追踪局限只能服务器→客户端单向浏览器连接数限制HTTP/1.1下6个/域名2.4 WebSocket双向实时通信结论需要双向实时通信选WebSocket。三、实战Spring Boot WebSocket构建聊天室3.1 项目结构websocket-chat/ ├── src/main/java/com/example/chat/ │ ├── config/ │ │ └── WebSocketConfig.java # 配置类 │ ├── handler/ │ │ └── ChatWebSocketHandler.java # 消息处理器 │ ├── manager/ │ │ └── SessionManager.java # 连接管理 │ ├── service/ │ │ └── MessageRouter.java # 消息路由 │ └── ChatApplication.java ├── src/main/resources/ │ └── static/ │ └── index.html # 前端页面 └── pom.xml3.2 依赖配置!-- pom.xml -- dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-websocket/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-redis/artifactId /dependency dependency groupIdcom.alibaba.fastjson2/groupId artifactIdfastjson2/artifactId version2.0.43/version /dependency /dependencies3.3 WebSocket配置// WebSocketConfig.java Configuration EnableWebSocket public class WebSocketConfig implements WebSocketConfigurer { Autowired private ChatWebSocketHandler chatHandler; Override public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { registry.addHandler(chatHandler, /ws/chat) .setAllowedOrigins(*) .addInterceptors(new AuthHandshakeInterceptor()); } Bean public ServletServerContainerFactoryBean createWebSocketContainer() { ServletServerContainerFactoryBean container new ServletServerContainerFactoryBean(); container.setMaxTextMessageBufferSize(8192); container.setMaxBinaryMessageBufferSize(8192); // 关键设置超时时间 container.setMaxSessionIdleTimeout(600000L); // 10分钟 return container; } }3.4 连接管理SessionManager// SessionManager.java Component Slf4j public class SessionManager { // 用户ID - Session映射 private final ConcurrentHashMapString, WebSocketSession userSessions new ConcurrentHashMap(); // Session ID - 用户ID映射用于断开时查找 private final ConcurrentHashMapString, String sessionUserMap new ConcurrentHashMap(); // 在线用户数统计 private final AtomicInteger onlineCount new AtomicInteger(0); public void addSession(String userId, WebSocketSession session) { WebSocketSession oldSession userSessions.put(userId, session); if (oldSession ! null oldSession.isOpen()) { try { oldSession.close(CloseStatus.POLICY_VIOLATION); log.info(用户[{}]被踢下线, userId); } catch (IOException e) { log.error(关闭旧会话失败, e); } } sessionUserMap.put(session.getId(), userId); int count onlineCount.incrementAndGet(); log.info(用户[{}]上线当前在线{}, userId, count); } public void removeSession(String sessionId) { String userId sessionUserMap.remove(sessionId); if (userId ! null) { userSessions.remove(userId); int count onlineCount.decrementAndGet(); log.info(用户[{}]下线当前在线{}, userId, count); } } public WebSocketSession getSession(String userId) { return userSessions.get(userId); } public CollectionWebSocketSession getAllSessions() { return userSessions.values(); } public int getOnlineCount() { return onlineCount.get(); } }3.5 消息处理器// ChatWebSocketHandler.java Component Slf4j public class ChatWebSocketHandler extends TextWebSocketHandler { Autowired private SessionManager sessionManager; Autowired private MessageRouter messageRouter; Override public void afterConnectionEstablished(WebSocketSession session) { String userId getUserIdFromSession(session); sessionManager.addSession(userId, session); // 发送欢迎消息 sendMessage(session, new ChatMessage(system, 欢迎进入聊天室)); // 广播用户上线通知 broadcast(new ChatMessage(notice, userId 加入了聊天室)); } Override protected void handleTextMessage(WebSocketSession session, TextMessage message) { String userId sessionManager.getUserIdBySessionId(session.getId()); String payload message.getPayload(); log.debug(收到消息[{}]: {}, userId, payload); try { ChatMessage chatMessage JSON.parseObject(payload, ChatMessage.class); chatMessage.setFrom(userId); chatMessage.setTimestamp(System.currentTimeMillis()); // 路由消息 messageRouter.route(chatMessage); } catch (Exception e) { log.error(消息解析失败, e); sendMessage(session, new ChatMessage(error, 消息格式错误)); } } Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { String userId sessionManager.getUserIdBySessionId(session.getId()); sessionManager.removeSession(session.getId()); broadcast(new ChatMessage(notice, userId 离开了聊天室)); } private void sendMessage(WebSocketSession session, ChatMessage message) { if (session.isOpen()) { try { session.sendMessage(new TextMessage(JSON.toJSONString(message))); } catch (IOException e) { log.error(发送消息失败, e); } } } private void broadcast(ChatMessage message) { String payload JSON.toJSONString(message); sessionManager.getAllSessions().forEach(session - { try { session.sendMessage(new TextMessage(payload)); } catch (IOException e) { log.error(广播消息失败, e); } }); } }3.6 消息路由// MessageRouter.java Component Slf4j public class MessageRouter { Autowired private SessionManager sessionManager; public void route(ChatMessage message) { switch (message.getType()) { case broadcast: // 广播给所有用户 broadcast(message); break; case private: // 私聊 sendToUser(message.getTo(), message); break; case group: // 群聊从Redis获取群成员 sendToGroup(message.getGroupId(), message); break; default: log.warn(未知消息类型: {}, message.getType()); } } private void sendToUser(String userId, ChatMessage message) { WebSocketSession session sessionManager.getSession(userId); if (session ! null session.isOpen()) { try { session.sendMessage(new TextMessage(JSON.toJSONString(message))); } catch (IOException e) { log.error(发送私聊消息失败, e); } } else { // 用户离线存储离线消息到Redis storeOfflineMessage(userId, message); } } }3.7 前端代码!-- index.html -- !DOCTYPE html html head titleWebSocket聊天室/title style body { font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; } #messages { border: 1px solid #ccc; height: 400px; overflow-y: auto; padding: 10px; margin-bottom: 10px; } .message { margin: 5px 0; padding: 8px; border-radius: 4px; } .system { background: #fff3cd; color: #856404; } .user { background: #d4edda; color: #155724; } .input-area { display: flex; gap: 10px; } input[typetext] { flex: 1; padding: 10px; } button { padding: 10px 20px; background: #007bff; color: white; border: none; cursor: pointer; } button:hover { background: #0056b3; } #status { margin-bottom: 10px; padding: 5px; border-radius: 4px; } .connected { background: #d4edda; } .disconnected { background: #f8d7da; } /style /head body h1WebSocket实时聊天室/h1 div idstatus classdisconnected● 未连接/div div idmessages/div div classinput-area input typetext iduserId placeholder输入用户ID valueuser_1 input typetext idmessageInput placeholder输入消息... button onclickconnect()连接/button button onclicksendMessage()发送/button /div script let ws null; let reconnectTimer null; const RECONNECT_INTERVAL 3000; function connect() { const userId document.getElementById(userId).value; ws new WebSocket(ws://localhost:8080/ws/chat?userId${userId}); ws.onopen () { updateStatus(true); console.log(WebSocket连接成功); if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer null; } }; ws.onmessage (event) { const msg JSON.parse(event.data); displayMessage(msg); }; ws.onclose () { updateStatus(false); console.log(WebSocket连接关闭准备重连...); reconnectTimer setTimeout(connect, RECONNECT_INTERVAL); }; ws.onerror (error) { console.error(WebSocket错误:, error); }; } function sendMessage() { const input document.getElementById(messageInput); const message { type: broadcast, content: input.value, timestamp: Date.now() }; if (ws ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify(message)); input.value ; } } function displayMessage(msg) { const div document.createElement(div); div.className message ${msg.from system ? system : user}; div.innerHTML strong${msg.from}:/strong ${msg.content} small stylecolor:#999${new Date(msg.timestamp).toLocaleTimeString()}/small; document.getElementById(messages).appendChild(div); document.getElementById(messages).scrollTop 999999; } function updateStatus(connected) { const status document.getElementById(status); status.className connected ? connected : disconnected; status.innerHTML connected ? ● 已连接 : ● 未连接; } // 回车发送 document.getElementById(messageInput).addEventListener(keypress, (e) { if (e.key Enter) sendMessage(); }); /script /body /html四、分布式场景Redis Pub/Sub实现多节点消息同步单节点能支撑几万连接但生产环境肯定多节点部署。问题用户A在Node1用户B在Node2怎么通信┌─────────────────────────────────────────────────────────────┐ │ 负载均衡器 (Nginx) │ └─────────────┬───────────────────────────────┬───────────────┘ │ │ ┌─────────▼──────────┐ ┌──────────▼──────────┐ │ WebSocket Node1 │◄──────►│ WebSocket Node2 │ │ (用户A连接在此) │ │ (用户B连接在此) │ └─────────┬──────────┘ └──────────┬──────────┘ │ │ └──────────────┬────────────────┘ │ ┌────────▼────────┐ │ Redis Pub/Sub │ │ (消息总线) │ └─────────────────┘4.1 Redis配置// RedisConfig.java Configuration public class RedisConfig { Bean public RedisTemplateString, Object redisTemplate(RedisConnectionFactory factory) { RedisTemplateString, Object template new RedisTemplate(); template.setConnectionFactory(factory); template.setKeySerializer(new StringRedisSerializer()); template.setValueSerializer(new GenericJackson2JsonRedisSerializer()); return template; } Bean public RedisMessageListenerContainer redisContainer( RedisConnectionFactory factory, MessageListenerAdapter listenerAdapter) { RedisMessageListenerContainer container new RedisMessageListenerContainer(); container.setConnectionFactory(factory); container.addMessageListener(listenerAdapter, new PatternTopic(chat:channel:*)); return container; } Bean public MessageListenerAdapter listenerAdapter(RedisMessageSubscriber subscriber) { return new MessageListenerAdapter(subscriber, onMessage); } }4.2 消息发布// RedisMessagePublisher.java Component public class RedisMessagePublisher { Autowired private StringRedisTemplate redisTemplate; private static final String CHANNEL chat:channel:broadcast; public void publish(ChatMessage message) { String payload JSON.toJSONString(message); redisTemplate.convertAndSend(CHANNEL, payload); } }4.3 消息订阅// RedisMessageSubscriber.java Component Slf4j public class RedisMessageSubscriber implements MessageListener { Autowired private SessionManager sessionManager; Override public void onMessage(Message message, byte[] pattern) { String payload new String(message.getBody()); ChatMessage chatMessage JSON.parseObject(payload, ChatMessage.class); // 只处理来自其他节点的消息本地消息已直接发送 if (!isLocalMessage(chatMessage)) { log.debug(收到Redis广播消息: {}, payload); broadcastToLocalSessions(chatMessage); } } private void broadcastToLocalSessions(ChatMessage message) { String payload JSON.toJSONString(message); sessionManager.getAllSessions().forEach(session - { try { session.sendMessage(new TextMessage(payload)); } catch (IOException e) { log.error(发送消息失败, e); } }); } }五、性能优化支撑10万并发的秘密5.1 连接优化// 调整Tomcat线程池 server.tomcat.max-threads200 server.tomcat.min-spare-threads50 // WebSocket容器配置 Bean public ServletServerContainerFactoryBean createWebSocketContainer() { ServletServerContainerFactoryBean container new ServletServerContainerFactoryBean(); container.setMaxTextMessageBufferSize(8192); container.setMaxBinaryMessageBufferSize(8192); container.setMaxSessionIdleTimeout(600000L); // 关键异步发送超时 container.setAsyncSendTimeout(5000L); return container; }5.2 消息压缩// 启用WebSocket消息压缩 Configuration public class WebSocketCompressionConfig { Bean public ServletServerContainerFactoryBean createWebSocketContainer() { ServletServerContainerFactoryBean container new ServletServerContainerFactoryBean(); // 启用压缩 container.setMaxTextMessageBufferSize(8192); return container; } }5.3 限流保护// RateLimiter.java Component public class RateLimiter { private final LoadingCacheString, AtomicInteger requestCounts CacheBuilder.newBuilder() .expireAfterWrite(1, TimeUnit.MINUTES) .build(new CacheLoaderString, AtomicInteger() { Override public AtomicInteger load(String key) { return new AtomicInteger(0); } }); private static final int MAX_REQUESTS_PER_MINUTE 60; public boolean allowRequest(String userId) { try { int count requestCounts.get(userId).incrementAndGet(); return count MAX_REQUESTS_PER_MINUTE; } catch (ExecutionException e) { return false; } } }5.4 JVM调优# 启动参数 java -Xms4g -Xmx4g \ -XX:UseG1GC \ -XX:MaxGCPauseMillis200 \ -XX:UseStringDeduplication \ -jar chat-server.jar5.5 系统级优化# 修改文件描述符限制Linux ulimit -n 65535 # 修改TCP参数 sysctl -w net.core.somaxconn65535 sysctl -w net.ipv4.tcp_max_syn_backlog65535 sysctl -w net.netfilter.nf_conntrack_max1000000六、源码获取与思考题源码获取完整项目代码已开源包含Spring Boot WebSocket服务端Redis分布式消息同步前端HTML客户端Docker部署配置GitHub地址https://github.com/example/websocket-chat思考题WebSocket连接突然断开如何保证消息不丢失提示考虑消息确认机制、离线消息存储如果Redis挂了分布式消息同步怎么降级提示考虑本地缓存、直接RPC调用如何实现已读回执功能提示考虑消息状态追踪、批量确认系列预告下一篇我们将深入探讨《Netty实现百万级WebSocket连接》包括Netty线程模型详解内存池优化心跳与断线重连集群架构设计敬请期待总结本文从WebSocket协议原理出发对比了三种实时通信方案完整演示了Spring Boot WebSocket构建聊天室的全过程并给出了分布式架构和性能优化的实战经验。核心要点回顾WebSocket一次握手全双工通信连接管理用ConcurrentHashMap线程安全分布式用Redis Pub/Sub做消息总线10万并发需要JVM、系统、代码三层优化标签WebSocket, 实时通信, Spring Boot, Redis, 高并发, 后端开发, 架构设计如果这篇文章对你有帮助欢迎点赞、收藏、转发。有问题欢迎在评论区留言我会一一回复。