Springboot集成websocket

文章目录

Springboot集成websocket

1. WebSocket概述

我们再日常的web应用开发过程中，常见的是前端向后端发起请求，但有的时候需要后端主动给前端发送消息，这个时候全双工的websocket即时通讯就闪亮登场了。

• 全双工通讯模式允许双方同时进行双向通讯，如手机通话。
• 半双工通讯模式允许双方交替发送和接收信息，但不能同时通讯，如对讲机。
• 单工通讯模式只能单向传输信息，不能回复，如广播电台。

2. WebSocket原理

WebSocket是一种在单个TCP连接上进行全双工通信的协议。它通过一个简单的握手过程来建立连接，然后在连接上进行双向数据传输。与传统的HTTP请求不同，WebSocket连接一旦建立，就可以在客户端和服务器之间保持打开状态，直到被任何一方关闭。

WebSocket协议的核心特点包括：

• 全双工通信：客户端和服务器可以同时发送和接收消息。
• 持久连接：一旦建立连接，就可以持续进行数据交换，无需像HTTP那样频繁地建立新的连接。
• 低延迟：由于连接是持久的，数据可以几乎实时地发送和接收。
• 轻量级协议：WebSocket协议的头部信息非常简单，减少了数据传输的开销。

3. Spring Boot集成WebSocket

在Spring Boot中集成WebSocket非常简单，Spring提供了对WebSocket的原生支持。以下是一个基本的集成步骤：

3.1 添加依赖

在

pom.xml

中添加Spring Boot的WebSocket依赖：

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

3.2 创建WebSocket配置类

创建一个配置类来启用和配置WebSocket：

packagecom.jiayuan.common.config;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.web.socket.server.standard.ServerEndpointExporter;importorg.springframework.web.socket.server.standard.ServletServerContainerFactoryBean;/**
 * @Title: WebSocketConfig
 * @Package: com.jiayuan.common.config
 * @Description: websocket配置
 * @Author: xmc
 * @Date: 创建时间 2024-04-24
 */@ConfigurationpublicclassWebSocketConfig{/**
     * 自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
     *
     * @return
     */@BeanpublicServerEndpointExporterserverEndpointExporter(){returnnewServerEndpointExporter();}/**
     * 通信文本消息和二进制缓存区大小
     * 避免对接 第三方 报文过大时，Websocket 1009 错误
     *
     * @return
     */@BeanpublicServletServerContainerFactoryBeancreateWebSocketContainer(){ServletServerContainerFactoryBean container =newServletServerContainerFactoryBean();// 在此处设置bufferSize
        container.setMaxTextMessageBufferSize(10240000);
        container.setMaxBinaryMessageBufferSize(10240000);
        container.setMaxSessionIdleTimeout(15*60000L);return container;}}

3.3 创建消息处理器

packagecom.jiayuan.common.config;importcn.hutool.core.collection.CollectionUtil;importcn.hutool.extra.spring.SpringUtil;importcn.hutool.json.JSONUtil;importcom.alibaba.fastjson2.JSON;importcom.alibaba.fastjson2.JSONObject;importcom.baomidou.mybatisplus.core.toolkit.StringUtils;importcom.jiayuan.common.redis.RedisCache;importcom.jiayuan.modules.critical.dto.CvRecordExtraDTO;importcom.jiayuan.modules.critical.dto.SyncRecordDTO;importcom.jiayuan.modules.critical.service.CvRecordService;importlombok.RequiredArgsConstructor;importlombok.extern.slf4j.Slf4j;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Component;importjavax.annotation.Resource;importjavax.websocket.*;importjavax.websocket.server.PathParam;importjavax.websocket.server.ServerEndpoint;importjava.io.IOException;importjava.util.List;importjava.util.concurrent.ConcurrentHashMap;/**
 * @Title: WebSocketServer
 * @Package: com.jiayuan.common.config
 * @Description: websocket的服务端
 * @Author: xmc
 * @Date: 创建时间 2024-04-24
 */@Component@Slf4j@ServerEndpoint("/api/pushMessage/{userId}")publicclassWebSocketServer{/**
     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
     */privatestaticint onlineCount =0;/**
     * concurrent包的线程安全Set，用来存放每个客户端对应的WebSocket对象。
     */privatestaticConcurrentHashMap<String,WebSocketServer> webSocketMap =newConcurrentHashMap<>();/**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */privateSession session;/**
     * 接收userId
     */privateString userId ="";/**
     * 连接建立成
     * 功调用的方法
     */@OnOpenpublicvoidonOpen(Session session,@PathParam("userId")String userId){this.session = session;this.userId = userId;if(webSocketMap.containsKey(userId)){
            webSocketMap.remove(userId);//加入set中
            webSocketMap.put(userId,this);}else{//加入set中
            webSocketMap.put(userId,this);//在线数加1addOnlineCount();}
        log.info("用户连接:"+ userId +",当前在线人数为:"+getOnlineCount());sendMessage("连接成功");}/**
     * 连接关闭
     * 调用的方法
     */@OnClosepublicvoidonClose(){if(webSocketMap.containsKey(userId)){
            webSocketMap.remove(userId);//从set中删除subOnlineCount();}
        log.info("用户退出:"+ userId +",当前在线人数为:"+getOnlineCount());}/**
     * 收到客户端消
     * 息后调用的方法
     *
     * @param message 客户端发送过来的消息
     **/@OnMessagepublicvoidonMessage(String message,Session session){
        log.info("用户消息:"+ userId +",报文:"+ message);//可以群发消息//消息保存到数据库、redisif(StringUtils.isNotBlank(message)){try{//解析发送的报文JSONObject jsonObject =JSON.parseObject(message);//追加发送人(防止串改)
                jsonObject.put("fromUserId",this.userId);String toUserId = jsonObject.getString("toUserId");//传送给对应toUserId用户的websocketif(StringUtils.isNotBlank(toUserId)&& webSocketMap.containsKey(toUserId)){
                    webSocketMap.get(toUserId).sendMessage(message);}else{//否则不在这个服务器上，发送到mysql或者redis
                    log.error("请求的userId:"+ toUserId +"不在该服务器上");}}catch(Exception e){
                e.printStackTrace();}}}/**
     * @param session
     * @param error
     */@OnErrorpublicvoidonError(Session session,Throwable error){

        log.error("用户错误:"+this.userId +",原因:"+ error.getMessage());
        error.printStackTrace();}/**
     * 实现服务
     * 器主动推送
     */publicvoidsendMessage(String message){try{this.session.getBasicRemote().sendText(message);}catch(IOException e){
            e.printStackTrace();}}/**
     * 发送自定
     * 义消息
     **/publicstaticvoidsendInfo(String message,String userId){
        log.info("发送消息到:"+ userId +"，报文:"+ message);if(StringUtils.isNotBlank(userId)&& webSocketMap.containsKey(userId)){
            webSocketMap.get(userId).sendMessage(message);}else{
            log.error("用户"+ userId +",不在线！");}}/**
     * 获得此时的
     * 在线人数
     *
     * @return
     */publicstaticsynchronizedintgetOnlineCount(){return onlineCount;}/**
     * 在线人
     * 数加1
     */publicstaticsynchronizedvoidaddOnlineCount(){WebSocketServer.onlineCount++;}/**
     * 在线人
     * 数减1
     */publicstaticsynchronizedvoidsubOnlineCount(){WebSocketServer.onlineCount--;}}

上述代码可能会有疑问：

session

和

userId

两个字段是不是安全的？多人连接，后面的会不会覆盖掉前面

session

和

userId

。

答案是不会的，关于处理器

WebSocketServer

我们要明确以下几点：

1. websocket是原型模式，@ServerEndpoint每次建立双向通信的时候都会创建一个实例
2. 为什么每次都@OnOpen都要检查webSocketMap.containsKey(userId) ，实际使用的时候发现偶尔会出现重连失败或者其他原因导致之前的session还存在，这时候就需要一个清除动作

3.4 服务器主动给客户端发送消息

 WebSocketServer.sendInfo("服务器主动给客户端发送消息test", "zhangsan");

4. 使用ApiPost测试WebSocket

以下是如何使用ApiPost进行测试的步骤：

1. 新建一个websocket测试

1. 填写URLws://localhost:8080/cvms-api/api/pushMessage/3

注意以下几点：

• 协议是ws，加密方式请选择wss
• 选择Raw
• URL的拼接公式如下

# servlet.context-path 这个是application.yml中的配置
ws://ip:port//${servlet.context-path}/注解@ServerEndpoint的值

• 有权限验证的，比如说shiro权限验证，URL就需要加入白名单

 filterMap.put("/api/pushMessage/*", "anon");

5. 参考和感谢

SpringBoot2.0集成WebSocket，实现后台向前端推送信息