Springboot集成websocket

文章目录

• Springboot集成websocket
• • 1. WebSocket概述
  • 2. WebSocket原理
  • 3. Spring Boot集成WebSocket
  • • 3.1 添加依赖
    • 3.2 创建WebSocket配置类
    • 3.3 创建消息处理器
    • 3.4 服务器主动给客户端发送消息
  • 4. 使用ApiPost测试WebSocket
  • 5. 参考和感谢

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：

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

3.3 创建消息处理器

package com.jiayuan.common.config;  import cn.hutool.core.collection.CollectionUtil; import cn.hutool.extra.spring.SpringUtil; import cn.hutool.json.JSONUtil; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONObject; import com.baomidou.mybatisplus.core.toolkit.StringUtils; import com.jiayuan.common.redis.RedisCache; import com.jiayuan.modules.critical.dto.CvRecordExtraDTO; import com.jiayuan.modules.critical.dto.SyncRecordDTO; import com.jiayuan.modules.critical.service.CvRecordService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;  import javax.annotation.Resource; import javax.websocket.*; import javax.websocket.server.PathParam; import javax.websocket.server.ServerEndpoint; import java.io.IOException; import java.util.List; import java.util.concurrent.ConcurrentHashMap;  /**  * @Title: WebSocketServer  * @Package: com.jiayuan.common.config  * @Description: websocket的服务端  * @Author: xmc  * @Date: 创建时间 2024-04-24  */ @Component @Slf4j @ServerEndpoint("/api/pushMessage/{userId}") public class WebSocketServer {      /**      * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。      */     private static int onlineCount = 0;     /**      * concurrent包的线程安全Set，用来存放每个客户端对应的WebSocket对象。      */     private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();     /**      * 与某个客户端的连接会话，需要通过它来给客户端发送数据      */     private Session session;     /**      * 接收userId      */     private String userId = "";      /**      * 连接建立成      * 功调用的方法      */     @OnOpen     public void onOpen(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);             //在线数加1             addOnlineCount();         }         log.info("用户连接:" + userId + ",当前在线人数为:" + getOnlineCount());         sendMessage("连接成功");     }      /**      * 连接关闭      * 调用的方法      */     @OnClose     public void onClose() {         if (webSocketMap.containsKey(userId)) {             webSocketMap.remove(userId);             //从set中删除             subOnlineCount();         }         log.info("用户退出:" + userId + ",当前在线人数为:" + getOnlineCount());     }      /**      * 收到客户端消      * 息后调用的方法      *      * @param message 客户端发送过来的消息      **/     @OnMessage     public void onMessage(String message, Session session) {         log.info("用户消息:" + userId + ",报文:" + message);         //可以群发消息         //消息保存到数据库、redis         if (StringUtils.isNotBlank(message)) {             try {                 //解析发送的报文                 JSONObject jsonObject = JSON.parseObject(message);                 //追加发送人(防止串改)                 jsonObject.put("fromUserId", this.userId);                 String toUserId = jsonObject.getString("toUserId");                 //传送给对应toUserId用户的websocket                 if (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      */     @OnError     public void onError(Session session, Throwable error) {          log.error("用户错误:" + this.userId + ",原因:" + error.getMessage());         error.printStackTrace();     }      /**      * 实现服务      * 器主动推送      */     public void sendMessage(String message) {         try {             this.session.getBasicRemote().sendText(message);         } catch (IOException e) {             e.printStackTrace();         }     }      /**      * 发送自定      * 义消息      **/     public static void sendInfo(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      */     public static synchronized int getOnlineCount() {         return onlineCount;     }      /**      * 在线人      * 数加1      */     public static synchronized void addOnlineCount() {         WebSocketServer.onlineCount++;     }      /**      * 在线人      * 数减1      */     public static synchronized void subOnlineCount() {         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测试

2. 填写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，实现后台向前端推送信息