Spring Boot WebSocketSession 实战：从心跳检测到连接管理，一个聊天室的完整实现-程序员充电站

Spring Boot WebSocketSession 实战：从心跳检测到连接管理，一个聊天室的完整实现

在当今互联网应用中，实时交互功能已成为标配需求。无论是社交平台的即时消息、在线协作工具的协同编辑，还是金融行业的实时行情推送，背后都离不开稳定高效的实时通信机制。传统HTTP协议"一问一答"的模式显然无法满足这些场景，而WebSocket协议凭借其全双工通信特性，成为构建实时应用的首选方案。

Spring Boot作为Java生态中最流行的应用框架，为WebSocket开发提供了简洁而强大的支持。但很多开发者在掌握了基础API后，面对真实业务场景时仍会陷入困境：如何确保连接稳定性？怎样高效管理大量活跃会话？异常断开后如何优雅恢复？这些问题在官方文档中往往找不到现成答案。

本文将从一个在线聊天室的完整实现出发，深入探讨WebSocketSession在实战中的高级应用。不同于简单的"Hello World"示例，我们将重点解决以下核心问题：

连接管理：使用线程安全的CopyOnWriteArrayList维护活跃会话
心跳机制：定时检测并剔除僵尸连接
异常处理：网络波动时的自动重连策略
消息广播：高效向所有客户端推送消息

1. 项目基础搭建

1.1 初始化Spring Boot项目

首先创建基础的Spring Boot项目，添加必要的依赖：

<dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> </dependencies>

1.2 WebSocket基础配置

创建WebSocket配置类，启用WebSocket支持并定义端点：

@Configuration @EnableWebSocket public class WebSocketConfig implements WebSocketConfigurer { @Override public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { registry.addHandler(chatWebSocketHandler(), "/chat") .setAllowedOrigins("*") .addInterceptors(new HttpSessionHandshakeInterceptor()); } @Bean public WebSocketHandler chatWebSocketHandler() { return new ChatWebSocketHandler(); } }

这里我们设置了WebSocket端点路径为/chat，并添加了一个握手拦截器用于获取HTTP会话信息。

2. 核心会话管理实现

2.1 会话容器设计

使用CopyOnWriteArrayList存储活跃会话，确保线程安全：

public class ChatWebSocketHandler implements WebSocketHandler { private final List<WebSocketSession> sessions = new CopyOnWriteArrayList<>(); @Override public void afterConnectionEstablished(WebSocketSession session) { sessions.add(session); log.info("新连接加入: {}, 当前在线: {}", session.getId(), sessions.size()); } @Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { sessions.remove(session); log.info("连接关闭: {}, 原因: {}, 当前在线: {}", session.getId(), status.getReason(), sessions.size()); } }

2.2 消息处理逻辑

实现消息广播功能，同时处理文本和二进制消息：

@Override public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) { if (message instanceof TextMessage) { String payload = ((TextMessage) message).getPayload(); broadcastMessage(session, payload); } else if (message instanceof BinaryMessage) { // 处理二进制消息（如图片、文件） byte[] payload = ((BinaryMessage) message).getPayload().array(); handleBinaryMessage(session, payload); } } private void broadcastMessage(WebSocketSession sender, String message) { sessions.parallelStream() .filter(WebSocketSession::isOpen) .forEach(session -> { try { session.sendMessage(new TextMessage( String.format("[%s]: %s", sender.getId().substring(0, 8), message))); } catch (IOException e) { log.error("消息发送失败", e); } }); }

3. 心跳检测机制实现

3.1 心跳协议设计

定义心跳消息格式和响应机制：

// 心跳请求格式 private static final String HEARTBEAT_REQUEST = "HB_REQUEST"; // 心跳响应格式 private static final String HEARTBEAT_RESPONSE = "HB_RESPONSE"; @Override public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) { if (message instanceof TextMessage) { String payload = ((TextMessage) message).getPayload(); if (HEARTBEAT_REQUEST.equals(payload)) { handleHeartbeat(session); return; } // 正常消息处理... } } private void handleHeartbeat(WebSocketSession session) { try { session.sendMessage(new TextMessage(HEARTBEAT_RESPONSE)); session.getAttributes().put("lastHeartbeat", System.currentTimeMillis()); } catch (IOException e) { log.error("心跳响应失败", e); } }

3.2 定时检测与清理

配置定时任务检查心跳状态：

@Scheduled(fixedRate = 30000) public void checkHeartbeats() { long currentTime = System.currentTimeMillis(); sessions.removeIf(session -> { Long lastHeartbeat = (Long) session.getAttributes().get("lastHeartbeat"); if (lastHeartbeat == null || currentTime - lastHeartbeat > 45000) { try { session.close(CloseStatus.SESSION_NOT_RELIABLE); } catch (IOException e) { log.error("关闭失效连接失败", e); } return true; } return false; }); }

4. 高级特性与异常处理

4.1 断线重连策略

客户端实现指数退避重连机制：

let reconnectAttempts = 0; const maxReconnectAttempts = 5; const baseDelay = 1000; function connectWebSocket() { const socket = new WebSocket('ws://localhost:8080/chat'); socket.onclose = function(e) { if (reconnectAttempts < maxReconnectAttempts) { const delay = Math.min(baseDelay * Math.pow(2, reconnectAttempts), 30000); reconnectAttempts++; setTimeout(connectWebSocket, delay); } }; // 其他事件处理... }

4.2 消息可靠性保障

服务端实现消息确认机制：

// 消息格式设计 public class ChatMessage { private String messageId; private String content; private long timestamp; // getters & setters } // 处理确认消息 private void handleAck(WebSocketSession session, String messageId) { // 从待确认队列移除消息 pendingMessages.remove(messageId); } // 消息重发机制 private void resendPendingMessages(WebSocketSession session) { pendingMessages.entrySet().stream() .filter(entry -> entry.getValue().equals(session)) .forEach(entry -> { try { session.sendMessage(new TextMessage(entry.getKey())); } catch (IOException e) { log.error("消息重发失败", e); } }); }

5. 性能优化与扩展

5.1 会话属性优化

合理设置WebSocket会话参数：

@Bean public ServletServerContainerFactoryBean createWebSocketContainer() { ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean(); container.setMaxTextMessageBufferSize(8192); container.setMaxBinaryMessageBufferSize(8192); container.setMaxSessionIdleTimeout(300000L); return container; }

5.2 分布式扩展方案

使用Redis实现跨节点会话管理：

@Configuration public class RedisConfig { @Bean public RedisTemplate<String, WebSocketSession> redisTemplate(RedisConnectionFactory factory) { RedisTemplate<String, WebSocketSession> template = new RedisTemplate<>(); template.setConnectionFactory(factory); template.setKeySerializer(new StringRedisSerializer()); template.setValueSerializer(new Jackson2JsonRedisSerializer<>(WebSocketSession.class)); return template; } } public class DistributedSessionManager { private final RedisTemplate<String, WebSocketSession> redisTemplate; public void addSession(WebSocketSession session) { redisTemplate.opsForValue().set(session.getId(), session); } public void broadcastMessage(String message) { Set<String> keys = redisTemplate.keys("*"); keys.forEach(key -> { WebSocketSession session = redisTemplate.opsForValue().get(key); if (session != null && session.isOpen()) { try { session.sendMessage(new TextMessage(message)); } catch (IOException e) { redisTemplate.delete(key); } } }); } }

在实现过程中，有几个关键点需要特别注意：