首页 > 软件编程 > java > WebSocket与SpringBoot整合

WebSocket与Spring Boot整合的完整实践记录

2026-07-31 10:24:53 作者：程序员小单

WebSocket是一种在单个TCP连接上实现全双工通信的协议,解决了HTTP协议在实时通信中的局限性,下面这篇文章主要介绍了WebSocket与Spring Boot整合的完整实践记录,文中通过代码介绍的非常详细,需要的朋友可以参考下

前言

在现代 Web 应用中，实时数据展示已成为基本需求。传统 HTTP 协议在实时通信方面存在局限性，而 WebSocket 协议则为此提供了完美的解决方案。本文将深入探讨两者的区别，并分享 Spring Boot 整合 WebSocket 的完整实践。

WebSocket 与 HTTP 的核心区别

HTTP 协议的局限性

单向通信：只能由客户端发起请求，服务端响应
无状态性：每次请求都是独立的，服务端无法主动推送数据
实时性差：需要客户端轮询获取最新数据

WebSocket 协议的优势

双向通信：服务端可以主动向客户端发送数据
持久连接：建立连接后保持长时间通信
低延迟：避免了 HTTP 的请求头开销，传输效率更高

实时数据推送方案对比

方案一：客户端轮询（短轮询）

// 前端定时调用接口
setInterval(() => {
    fetch('/api/data')
        .then(response => response.json())
        .then(data => updateUI(data));
}, 1000);

缺点：

资源浪费：无数据更新时仍频繁请求
实时性差：最大延迟等于轮询间隔
会话管理复杂

方案二：WebSocket 实时推送

// 建立 WebSocket 连接
const socket = new WebSocket('ws://localhost:8080/ws/info');
socket.onmessage = (event) => {
    const data = JSON.parse(event.data);
    updateUI(data);
};

优势：

真正的实时通信
减少不必要的网络请求
服务端可控的数据推送

Spring Boot 整合 WebSocket 完整实现

1. 项目依赖配置

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

2. WebSocket 配置类

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(customWebSocketHandler(), "/ws/info")
                .setAllowedOrigins("*")
                .addInterceptors(new UserAttributeHandshakeInterceptor());
    }

    @Bean
    public CustomWebSocketHandler customWebSocketHandler() {
        return new CustomWebSocketHandler();
    }

    @Bean
    public ServletServerContainerFactoryBean createWebSocketContainer() {
        ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean();
        container.setAsyncSendTimeout(120 * 1000L);        // 异步发送超时时间
        container.setMaxTextMessageBufferSize(300 * 1024); // 文本消息缓冲区大小
        container.setMaxBinaryMessageBufferSize(300 * 1024); // 二进制消息缓冲区大小
        return container;
    }
}

3. 连接拦截器（身份验证）

@Slf4j
public class UserAttributeHandshakeInterceptor implements HandshakeInterceptor {
    
    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response,
                                   WebSocketHandler wsHandler, Map<String, Object> attributes) {
        if (request instanceof ServletServerHttpRequest) {
            ServletServerHttpRequest servletRequest = (ServletServerHttpRequest) request;
            
            // Token 验证
            String token = servletRequest.getServletRequest().getParameter("access_token");
            if (StringUtils.isEmpty(token)) {
                throw new RuntimeException("令牌不能为空");
            }
            
            // JWT 解析和验证
            Claims claims = JwtUtils.parseToken(token);
            if (claims == null) {
                throw new RuntimeException("令牌已过期或验证不正确");
            }
            
            // 获取用户信息
            String userId = JwtUtils.getUserId(claims);
            attributes.put("USER_ID", userId);
            
            // 获取业务参数
            String wellNo = servletRequest.getServletRequest().getParameter("wellNo");
            String wid = servletRequest.getServletRequest().getParameter("wid");
            String fraId = servletRequest.getServletRequest().getParameter("fraId");
            
            attributes.put("WELL_NO", StringUtils.trimToEmpty(wellNo));
            attributes.put("W_ID", StringUtils.trimToEmpty(wid));
            attributes.put("FRAD_ID", StringUtils.trimToEmpty(fraId));
        }
        return true;
    }

    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,
                               WebSocketHandler wsHandler, Exception exception) {
        // 握手后的处理逻辑
    }
}

4. WebSocket 连接处理器

@Slf4j
public class CustomWebSocketHandler extends TextWebSocketHandler {
    
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        String sessionKey = generateSessionKey(session);
        WebSocketSessionHolder.addSession(sessionKey, session);
        log.info("WebSocket 连接已建立: {}", sessionKey);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {
        String sessionKey = generateSessionKey(session);
        WebSocketSessionHolder.removeSession(sessionKey);
        log.info("WebSocket 连接已关闭: {}", sessionKey);
    }

    /**
     * 生成会话唯一标识
     * 格式: 井号#井ID#段ID#用户ID#会话ID
     */
    private String generateSessionKey(WebSocketSession session) {
        Map<String, Object> attributes = session.getAttributes();
        return Stream.of(
                attributes.get("WELL_NO").toString(),
                attributes.get("W_ID").toString(),
                attributes.get("FRAD_ID").toString(),
                attributes.get("USER_ID").toString(),
                session.getId()
        ).filter(StringUtils::isNotBlank).collect(Collectors.joining("#"));
    }
}

5. 会话管理容器

@Slf4j
public final class WebSocketSessionHolder {
    
    private WebSocketSessionHolder() {}
    
    private static final Map<String, WebSocketSession> SESSION_MAP = new ConcurrentHashMap<>();

    public static void addSession(String sessionKey, WebSocketSession session) {
        // 使用装饰器增强会话功能
        WebSocketSession decoratedSession = new ConcurrentWebSocketSessionDecorator(
            session, 120 * 1000, 300 * 1024
        );
        SESSION_MAP.put(sessionKey, decoratedSession);
    }

    public static void removeSession(String sessionKey) {
        try (WebSocketSession removed = SESSION_MAP.remove(sessionKey)) {
            log.info("移除会话: {}", removed != null ? removed.getId() : sessionKey);
        } catch (Exception e) {
            log.error("移除会话异常", e);
        }
    }

    public static Collection<WebSocketSession> getSessions() {
        return SESSION_MAP.values();
    }
    
    public static Map<String, WebSocketSession> getSessionMap() {
        return SESSION_MAP;
    }
}

6. 数据推送核心服务

@Slf4j
@Service
@RequiredArgsConstructor
public class WebSocketCustomHandler {
    
    private static final ScheduledExecutorService scheduler = 
        Executors.newSingleThreadScheduledExecutor();
    
    // 数据缓存，记录各会话的最新数据时间
    private static final Map<String, LocalDateTime> realTimeCache = new ConcurrentHashMap<>();
    private static final Map<String, LocalDateTime> curveCache = new ConcurrentHashMap<>();
    
    private final Executor customWsExecutor;
    private final VarConfig varConfig;
    private final RedisService redisService;
    private final FractMapper fractMapper;
    private final WhshWsCommonService whshWsCommonService;

    /**
     * 应用启动后开始定时推送数据
     */
    @EventListener(ApplicationReadyEvent.class)
    public void startDataPushing() {
        scheduler.scheduleWithFixedDelay(() -> {
            try {
                this.pushDataToClients();
            } catch (Exception e) {
                log.error("数据推送任务执行异常", e);
            }
        }, 1, varConfig.getWsDataCycle(), TimeUnit.SECONDS);
    }

    private void pushDataToClients() {
        Map<String, WebSocketSession> sessionMap = WebSocketSessionHolder.getSessionMap();
        if (MapUtils.isEmpty(sessionMap)) {
            return;
        }

        // 清理无效缓存
        cleanupInvalidCaches(sessionMap);

        // 按业务维度分组会话
        Map<String, List<WebSocketSession>> groupedSessions = groupSessionsByBusiness(sessionMap);

        // 并行处理所有分组的数据推送
        CompletableFuture<?>[] futures = groupedSessions.entrySet().stream()
                .flatMap(entry -> entry.getValue().stream()
                        .map(session -> processAndSendData(entry.getKey(), session)))
                .toArray(CompletableFuture[]::new);

        CompletableFuture.allOf(futures)
                .thenRun(() -> log.info("本轮 WebSocket 数据推送完成"))
                .exceptionally(throwable -> {
                    log.error("数据推送异常", throwable);
                    return null;
                });
    }

    /**
     * 处理并发送数据到指定会话
     */
    private CompletableFuture<Void> processAndSendData(String businessKey, WebSocketSession session) {
        return CompletableFuture.runAsync(() -> {
            try {
                FractDataVo dataVo = buildDataForSession(businessKey, session);
                if (shouldSendData(dataVo)) {
                    sendMessageToSession(session, dataVo);
                }
            } catch (Exception e) {
                log.error("处理 WebSocket 数据异常, key: {}", businessKey, e);
            }
        }, customWsExecutor);
    }

    /**
     * 构建会话所需的数据
     */
    private FractDataVo buildDataForSession(String businessKey, WebSocketSession session) {
        String[] params = businessKey.split("#");
        DiagnosisParam param = buildDiagnosisParam(params);
        FractDataVo dataVo = new FractDataVo();

        // 实时数据处理
        processRealTimeData(param, dataVo, session);
        
        // 预测数据处理
        if (isPredictionEnabled(params)) {
            processPredictionData(param, dataVo, session);
        }

        return dataVo;
    }

    private void processRealTimeData(DiagnosisParam param, FractDataVo dataVo, WebSocketSession session) {
        if (!realTimeCache.containsKey(session.getId())) {
            // 首次加载全量数据
            List<FractRealtimeVo> realtimeData = whshWsCommonService.getFractRealTime(param);
            dataVo.getRtVo().setRts(realtimeData);
            if (CollectionUtils.isNotEmpty(realtimeData)) {
                cacheLastDataTime(realTimeCache, session, realtimeData.get(realtimeData.size() - 1).getYlsj());
            }
        } else {
            // 增量数据查询
            LocalDateTime lastTime = realTimeCache.get(session.getId());
            param.setLastTime(lastTime);
            List<FractRealtimeVo> incrementalData = whshWsCommonService.getIncFractRealtime(param);
            dataVo.getRtVo().setRts(incrementalData);
            dataVo.getRtVo().setTp(1); // 标记为增量数据
            if (CollectionUtils.isNotEmpty(incrementalData)) {
                cacheLastDataTime(realTimeCache, session, incrementalData.get(incrementalData.size() - 1).getYlsj());
            }
        }
    }

    private void sendMessageToSession(WebSocketSession session, FractDataVo dataVo) {
        if (session.isOpen()) {
            try {
                logDataStatistics(dataVo); // 日志记录数据统计
                String message = JSON.toJSONString(dataVo);
                session.sendMessage(new TextMessage(message));
            } catch (Exception e) {
                log.error("WebSocket 消息发送失败", e);
            }
        }
    }

    private void logDataStatistics(FractDataVo dataVo) {
        if (dataVo.getPcVo() != null && dataVo.getPcVo().getPcs() != null) {
            log.info("发送预测曲线数据条数: {}", dataVo.getPcVo().getPcs().size());
        }
        if (dataVo.getWcVo() != null && dataVo.getWcVo().getWcs() != null) {
            log.info("发送工况数据条数: {}", dataVo.getWcVo().getWcs().size());
        }
        log.info("发送实时数据条数: {}", dataVo.getRtVo().getRts().size());
    }
}

使用工具连接验证 apifox

核心设计思路

1. 连接管理

使用拦截器进行身份认证和参数解析
通过会话持有者统一管理所有活跃连接
为每个连接生成唯一业务标识

2. 数据推送策略

增量推送：首次全量，后续只推送变化数据
缓存机制：记录各连接的最新数据时间戳
异常处理：单个连接异常不影响整体服务

3. 性能优化

使用线程池并行处理数据推送
定时清理无效连接缓存
配置合适的消息缓冲区大小

总结

WebSocket 为实时数据推送提供了高效的解决方案，特别适合数据频繁变化的业务场景。通过 Spring Boot 的优雅集成，我们可以快速构建稳定可靠的实时通信功能。本文提供的完整实现方案已在生产环境验证，可直接参考使用。

在实际项目中，还可以进一步优化：

添加连接心跳检测
实现消息重试机制
增加流量控制策略
完善监控和告警

到此这篇关于WebSocket与Spring Boot整合的完整实践记录的文章就介绍到这了,更多相关WebSocket与SpringBoot整合内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！