devbox
从前慢现在也慢
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

SpringBoot整合配置WebSocket服务代码实现_spring-boot-starter-websocket

作者：从前慢现在也慢 | 2024-06-14 08:20:33

赞

踩

spring-boot-starter-websocket

基于Javax的实现

首先，在springboot 项目中的pom文件引入WebSocket的依赖，配置如下：

  1.     <!--websocket-->
  2.         <dependency>
  3.             <groupId>org.springframework.boot</groupId>
  4.             <artifactId>spring-boot-starter-websocket</artifactId>
  5.             <exclusions>
  6.                 <exclusion>
  7.                     <artifactId>spring-boot-starter-tomcat</artifactId>
  8.                     <groupId>org.springframework.boot</groupId>
  9.                 </exclusion>
  10.             </exclusions>
  11.         </dependency>

在springboot 项目中，新建WebSocketConfig配置类，代码如下：

  1.  
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.springframework.web.socket.config.annotation.EnableWebSocket;
  6. import org.springframework.web.socket.server.standard.ServerEndpointExporter;
  7.  
  8. /**
  9.  *
  10.  *
  11.  * @version 1.0
  12.  * @since JDK1.8
  13.  * @author tarzan
  14.  * @date 2023年10月26日 15:21:09
  15.  */
  16. @Slf4j
  17. @Configuration
  18. public class WebSocketConfig{
  19.  
  20.  
  21.     @Bean
  22.     public ServerEndpointExporter serverEndpointExporter() {
  23.         return new ServerEndpointExporter();
  24.     }
  25.  
  26. }

在springboot 项目中，新建WebSocketServer 服务器服务类

完整代码如下：

  1.  
  2. import com.alibaba.fastjson.JSONObject;
  3. import lombok.EqualsAndHashCode;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springblade.core.tool.utils.CollectionUtil;
  6. import org.springblade.core.tool.utils.Func;
  7. import org.springframework.stereotype.Component;
  8.  
  9. import javax.websocket.*;
  10. import javax.websocket.server.ServerEndpoint;
  11. import java.util.List;
  12. import java.util.concurrent.CopyOnWriteArraySet;
  13.  
  14. /**
  15.  * @author tarzan
  16.  */
  17. @ServerEndpoint("/ws/automate")
  18. @Component
  19. @Slf4j
  20. @EqualsAndHashCode
  21. public class WebSocketServer {
  22.  
  23.     /** 虽然@Component默认是单例模式的，但springboot还是会为每个websocket连接初始化一个bean，所以可以用一个静态set保存起来。 */
  24.     private static final CopyOnWriteArraySet<WebSocketServer> WEBSOCKET_CLIENTS =new CopyOnWriteArraySet<>();
  25.     /** 当前session */
  26.     private Session session;
  27.     /** 当前session订阅的话题 */
  28.     private List<String> topics;
  29.  
  30.     public WebSocketServer() {
  31.     }
  32.  
  33.     @OnOpen
  34.     public void onOpen(Session session) {
  35.         this.session = session;
  36.         WEBSOCKET_CLIENTS.add(this);
  37.         log.info("WebSocketServer有新客户端连接加入：{}，当前在线人数为：{}", session.getId(),WEBSOCKET_CLIENTS.size());
  38.     }
  39.  
  40.     @OnClose
  41.     public void onClose() {
  42.         WEBSOCKET_CLIENTS.remove(this);
  43.         log.info("有一连接关闭：{}，当前在线人数为：{}", this.session.getId(), WEBSOCKET_CLIENTS.size());
  44.     }
  45.  
  46.     @OnMessage
  47.     public void onMessage(String message, Session session) {
  48.         this.topics= Func.toStrList(message);
  49.         log.info("服务端收到客户端[{}]的消息:{}", session.getId(), message);
  50.     }
  51.  
  52.     @OnError
  53.     public void onError(Session session, Throwable error) {
  54.         log.error("发生错误:"+error.getMessage());
  55.         WEBSOCKET_CLIENTS.remove(this);
  56.         log.info("有一连接异常：{}，当前在线人数为：{}", session.getId(), WEBSOCKET_CLIENTS.size());
  57.     }
  58.  
  59.     public  <T> void sendData(T data, String topicName)  {
  60.         JSONObject result=new JSONObject();
  61.         result.put("topic",topicName);
  62.         result.put("data",data);
  63.         //显示值为null的字段
  64.         //  String jsonString = JSON.toJSONString(result, SerializerFeature.WriteMapNullValue);
  65.         sendMessage(result.toString(),topicName);
  66.     }
  67.  
  68.     /**
  69.      * 发送消息广播
  70.      *
  71.      * @param message 消息文本
  72.      * @author tarzan
  73.      * @date 2022年12月08日 09:24:34
  74.      */
  75.     public  void sendMessage(String message,String topic){
  76.         try {
  77.             WEBSOCKET_CLIENTS.forEach(client ->{
  78.                 if (client.session.isOpen()&&CollectionUtil.isNotEmpty(client.topics)&&client.topics.contains(topic)) {
  79.                     client.session.getAsyncRemote().sendText(message);
  80.                 }
  81.             });
  82.         } catch (Exception e) {
  83.             log.error(e.getMessage());
  84.         }
  85.     }
  86.  
  87.     public  void sendMessageToAllUsers(String message){
  88.         try {
  89.             WEBSOCKET_CLIENTS.forEach(client -> {
  90.                 if (client.session.isOpen()) {
  91.                     client.session.getAsyncRemote().sendText(message);
  92.                 }
  93.             });
  94.         } catch (Exception e) {
  95.             log.error(e.getMessage());
  96.         }
  97.     }
  98.  
  99.  
  100. }

注意：配置号websocket服务后，访问端口和你自己web端口是一样的。 

发送使用代码示例

  1.  
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.springblade.coalface.websocket.OpcSocketServer;
  4. import org.springblade.coalface.websocket.TopicNameConstant;
  5. import org.springblade.coalface.websocket.WebSocketServer;
  6. import org.springframework.scheduling.annotation.Scheduled;
  7. import org.springframework.stereotype.Component;
  8.  
  9. import javax.annotation.Resource;
  10.  
  11. /**
  12.  * @author tarzan
  13.  */
  14. @Slf4j
  15. @Component
  16. public class WebSocketTask{
  17.  
  18.     @Resource
  19.     private WebSocketServer webSocketServer;
  20.  
  21.     @Scheduled(cron = "0/5 * * * * ?")
  22.     public void execute() {
  23.         webSocketServer.sendMessageToAllUsers(TopicNameConstant.HEARTBEAT);
  24.     }
  25. }

注：也可以把sendMessageToAllUsers方法改为静态方法，通过类去调用。 

代码是我项目中的代码，请按照自己的业务进行修改！ 

WebSocket服务器类种注入serivice服务实现

完整代码如下（这里发送的消息的方法是静态方法）：

  1.  
  2.  
  3. import lombok.EqualsAndHashCode;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springblade.coalface.calculate.enums.CollisionLevelEnum;
  6. import org.springblade.coalface.opcua.dto.CoalCutterDTO;
  7. import org.springblade.coalface.opcua.service.OpcIssueService;
  8. import org.springblade.core.tool.utils.SpringUtil;
  9. import org.springblade.core.tool.utils.StringUtil;
  10. import org.springframework.stereotype.Component;
  11.  
  12. import javax.websocket.*;
  13. import javax.websocket.server.ServerEndpoint;
  14. import java.util.concurrent.CopyOnWriteArraySet;
  15. import java.util.concurrent.atomic.AtomicReference;
  16.  
  17. /**
  18.  * @author tarzan
  19.  */
  20. @ServerEndpoint("/ws/opc/issue")
  21. @Slf4j
  22. @Component("opcWebSocketServer")
  23. @EqualsAndHashCode
  24. public class OpcSocketServer {
  25.  
  26.     /** 虽然@Component默认是单例模式的，但springboot还是会为每个websocket连接初始化一个bean，所以可以用一个静态set保存起来。 */
  27.     private static final CopyOnWriteArraySet<OpcSocketServer> WEBSOCKET_CLIENTS =new CopyOnWriteArraySet<>();
  28.     public static  final AtomicReference<CollisionLevelEnum> CMD = new AtomicReference<>(null);
  29.     /** 当前session */
  30.     private Session session;
  31.  
  32.     private final static String ALERT="alert";
  33.     private final static String CONTROL="control";
  34.  
  35.     private  final static OpcIssueService opcIssueService;
  36.  
  37.     static {
  38.         opcIssueService= SpringUtil.getBean(OpcIssueService.class);
  39.     }
  40.  
  41.     @OnOpen
  42.     public void onOpen(Session session) {
  43.         this.session = session;
  44.         WEBSOCKET_CLIENTS.add(this);
  45.         log.info("OpcSocketServer有新客户端连接加入：{}，当前在线人数为：{}", session.getId(),WEBSOCKET_CLIENTS.size());
  46.     }
  47.  
  48.     @OnClose
  49.     public void onClose() {
  50.         WEBSOCKET_CLIENTS.remove(this);
  51.         log.info("有一连接关闭：{}，当前在线人数为：{}", this.session.getId(), WEBSOCKET_CLIENTS.size());
  52.     }
  53.  
  54.     @OnMessage
  55.     public void onMessage(String message, Session session) {
  56.         log.info("服务端收到客户端[{}]的消息:{}", session.getId(), message);
  57.         String[] params=message.split(":");
  58.         if(StringUtil.isNotBlank(message)&&params.length>1){
  59.             if(ALERT.equals(params[0])){
  60.                 CollisionLevelEnum levelEnum= CollisionLevelEnum.valueOf(params[1]);
  61.                 if(CollisionLevelEnum.SAFE.equals(levelEnum)){
  62.                     manualIssue(session,levelEnum);
  63.                 }
  64.             }
  65.             if(CONTROL.equals(params[0])){
  66.                 CollisionLevelEnum levelEnum= CollisionLevelEnum.valueOf(params[1]);
  67.                 //是否需要相等
  68.                 if(levelEnum.getStatus()>= CoalCutterDTO.currentLevel.getStatus()){
  69.                     manualIssue(session,levelEnum);
  70.                 }
  71.             }
  72.         }
  73.     }
  74.  
  75.     private void  manualIssue(Session session, CollisionLevelEnum levelEnum){
  76.         CollisionLevelEnum execute=CMD.get();
  77.         if(execute==null){
  78.             verifyIssue(session,levelEnum);
  79.         }else{
  80.             if(levelEnum.getStatus()>execute.getStatus()){
  81.                 verifyIssue(session,levelEnum);
  82.             }else{
  83.                 sendMessageToUser(session,execute+"指令正在执行！请稍后点击");
  84.             }
  85.         }
  86.     }
  87.  
  88.  
  89.     private void  verifyIssue(Session session,CollisionLevelEnum levelEnum){
  90.         if(opcIssueService.manualIssue(levelEnum)){
  91.             opcIssueService.delayRecover(15,levelEnum);
  92.             sendMessageToUser(session,levelEnum+"指令下发成功！");
  93.             log.info(levelEnum+"指令下发成功！");
  94.         }else {
  95.             sendMessageToUser(session,levelEnum+"指令下发失败！");
  96.             log.error(levelEnum+"指令下发失败！");
  97.         }
  98.     }
  99.  
  100.     @OnError
  101.     public void onError(Session session, Throwable error) {
  102.         log.error("发生错误:"+error.getMessage());
  103.         WEBSOCKET_CLIENTS.remove(this);
  104.         log.info("有一连接异常：{}，当前在线人数为：{}", session.getId(), WEBSOCKET_CLIENTS.size());
  105.     }
  106.  
  107.     /**
  108.      * sendMessageToUser:发给指定用户
  109.      *
  110.      */
  111.     public static void sendMessageToUser(Session session, String message) {
  112.         if (session != null && session.isOpen()) {
  113.             session.getAsyncRemote().sendText(message);
  114.         }
  115.     }
  116.     public static void sendMessageToAllUsers(String message){
  117.         try {
  118.             WEBSOCKET_CLIENTS.forEach(client ->sendMessageToUser(client.session,message));
  119.         } catch (Exception e) {
  120.             log.error(e.getMessage());
  121.         }
  122.     }
  123.  
  124.  
  125. }

以上 通过  static {
        opcIssueService= SpringUtil.getBean(OpcIssueService.class);
    } 实现的service 服务注入，发现有时候打包的时候，会出现 getBean空指针问题。

建议改成：在方法的内使用

  1.     private void  verifyIssue(Session session,CollisionLevelEnum levelEnum){
  2.         OpcIssueService opcIssueService = SpringUtil.getBean(OpcIssueService.class);
  3.         if(opcIssueService.manualIssue(levelEnum)){
  4.             opcIssueService.delayRecover(15,levelEnum);
  5.             sendMessageToUser(session,levelEnum+"指令下发成功！");
  6.             log.info(levelEnum+"指令下发成功！");
  7.         }else {
  8.             sendMessageToUser(session,levelEnum+"指令下发失败！");
  9.             log.error(levelEnum+"指令下发失败！");
  10.         }
  11.     }

基于WebMVC的实现

首先，在springboot 项目中的pom文件引入WebSocket的依赖，配置如下：

  1.     <!--websocket-->
  2.         <dependency>
  3.             <groupId>org.springframework.boot</groupId>
  4.             <artifactId>spring-boot-starter-websocket</artifactId>
  5.             <exclusions>
  6.                 <exclusion>
  7.                     <artifactId>spring-boot-starter-tomcat</artifactId>
  8.                     <groupId>org.springframework.boot</groupId>
  9.                 </exclusion>
  10.             </exclusions>
  11.         </dependency>

在springboot 项目中，新建WebSocketConfig配置类，需要增加@EnableWebSocket注解和继承WebSocketConfigurer 类。

代码如下：

  1.  
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.springframework.web.socket.config.annotation.EnableWebSocket;
  6. import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
  7. import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
  8. import org.springframework.web.socket.server.standard.ServerEndpointExporter;
  9.  
  10. import javax.annotation.Resource;
  11.  
  12. /**
  13.  *
  14.  *
  15.  * @version 1.0
  16.  * @since JDK1.8
  17.  * @author tarzan
  18.  * @date 2023年10月26日 15:21:09
  19.  */
  20. @Slf4j
  21. @Configuration
  22. @EnableWebSocket
  23. public class WebSocketConfig implements WebSocketConfigurer {
  24.  
  25.     @Resource
  26.     private OpcSessionHandler opcSessionHandler;
  27.  
  28.     //addHandler 映射关系可以添加多个
  29.     @Override
  30.     public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
  31.         registry.addHandler(opcSessionHandler, "/ws/opc/issue").setAllowedOrigins("*");
  32.     }
  33.  
  34.     //只使用webmvc的方式这个可以省略不写
  35.     @Bean
  36.     public ServerEndpointExporter serverEndpointExporter() {
  37.         return new ServerEndpointExporter();
  38.     }
  39.  
  40. }

在springboot 项目中，新建OpcSessionHandler 会话处理类，代码如下：

  1. package org.springblade.coalface.websocket;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springblade.coalface.calculate.enums.CollisionLevelEnum;
  5. import org.springblade.coalface.opcua.dto.CoalCutterDTO;
  6. import org.springblade.coalface.opcua.service.OpcIssueService;
  7. import org.springblade.core.tool.utils.StringUtil;
  8. import org.springframework.lang.NonNull;
  9. import org.springframework.stereotype.Component;
  10. import org.springframework.web.socket.*;
  11.  
  12. import javax.annotation.Resource;
  13. import java.io.IOException;
  14. import java.util.concurrent.CopyOnWriteArraySet;
  15. import java.util.concurrent.ScheduledExecutorService;
  16. import java.util.concurrent.ScheduledThreadPoolExecutor;
  17. import java.util.concurrent.TimeUnit;
  18. import java.util.concurrent.atomic.AtomicReference;
  19.  
  20.  
  21. /**
  22.  * @author tarzan
  23.  */
  24. @Component
  25. @Slf4j
  26. public class  OpcSessionHandler implements WebSocketHandler {
  27.  
  28.     @Resource
  29.     private OpcIssueService opcIssueService;
  30.  
  31.     private static final CopyOnWriteArraySet<WebSocketSession> SESSIONS=new CopyOnWriteArraySet<>();
  32.     public  final AtomicReference<CollisionLevelEnum> CMD = new AtomicReference<>(null);
  33.  
  34.     private final static String ALERT="alert";
  35.     private final static String CONTROL="control";
  36.  
  37.     @Override
  38.     public void afterConnectionEstablished(@NonNull WebSocketSession session) throws Exception {
  39.         SESSIONS.add(session);
  40.         log.info(session.getId()+" OpcSessionHandler当前在线人数："+SESSIONS.size());
  41.     }
  42.  
  43.     @Override
  44.     public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
  45.         String msg = message.getPayload().toString();
  46.         log.info("接收消息"+session.getId()+":"+msg);
  47.         opcIssueService.method();
  48.     }
  49.  
  50.  
  51.  
  52.     @Override
  53.     public void handleTransportError(WebSocketSession session,@NonNull Throwable exception) throws Exception {
  54.         log.error("OpcSessionHandler连接出错"+session.getId());
  55.         SESSIONS.remove(session);
  56.         if (!session.isOpen()) {
  57.             session.close();
  58.         }
  59.     }
  60.  
  61.     @Override
  62.     public void afterConnectionClosed(WebSocketSession session, @NonNull CloseStatus closeStatus) throws Exception {
  63.        log.info("OpcSessionHandler关闭连接"+session.getId());
  64.        SESSIONS.remove(session);
  65.     }
  66.  
  67.  
  68.     @Override
  69.     public boolean supportsPartialMessages() {
  70.         return false;
  71.     }
  72.  
  73.  
  74.     public  void sendMessageToUser(WebSocketSession session, String contents) {
  75.         if (session != null && session.isOpen()) {
  76.             TextMessage message = new TextMessage(contents);
  77.             try {
  78.                 session.sendMessage(message);
  79.             } catch (IOException e) {
  80.                 log.error(e.getMessage());
  81.             }
  82.         }
  83.     }
  84.  
  85.     public void sendMessageToAllUsers(String contents) {
  86.         SESSIONS.forEach(session->sendMessageToUser(session, contents));
  87.     }
  88.  
  89. }

配置以上代码即可。

注：通过WebMVC实现的websocket暂时没办法通过异步推送数据。 

相关知识

WebSocket是一种在单个TCP连接上进行全双工通信的网络协议，它可以在客户端和服务器之间建立持久性的连接，并提供实时的双向通信能力。相较于传统的HTTP协议，WebSocket具有更低的延迟和更高的效率，适用于需要实时数据传输和实时互动的应用场景。

下面将对WebSocket技术的几个关键概念进行详细解释。

  1. 握手过程（Handshake） WebSocket连接的建立需要经过一次握手过程。在客户端发起连接请求时，会发送一个特殊的HTTP请求头（Upgrade请求头字段），并告知服务器要升级协议为WebSocket。服务器在接收到这个请求后，如果同意升级，则会返回一个特殊的HTTP响应头（Upgrade和Connection响应头字段），表示已经切换到WebSocket协议。握手完成后，连接从HTTP协议切换到WebSocket协议。

  2. 数据帧（Data Frames） WebSocket使用数据帧来传输数据，数据帧有固定的格式。数据帧包含一个帧头和一个帧载荷。帧头包含了数据帧的控制信息，例如数据帧是否是最后一个帧、数据帧的类型和长度等。帧载荷是实际传输的数据。

  3. 实时双向通信 WebSocket协议支持实时的双向通信，客户端和服务器可以同时进行数据的发送和接收。客户端可以通过JavaScript的WebSocket API与服务器建立连接，并通过发送数据帧向服务器发送消息。服务器可以向客户端发送消息，客户端通过监听WebSocket的事件来接收服务器发送的消息。

  4. 心跳机制 WebSocket支持心跳机制来保持连接的稳定性。客户端和服务器可以定时发送心跳消息以保持连接的活跃状态。如果一段时间内没有收到心跳消息，可以判定连接已断开，并进行相应的处理。

  5. 安全性（Security） WebSocket可以使用TLS/SSL加密来保护数据的传输。通过使用加密的WebSocket连接，可以确保数据的机密性和完整性，以防止数据被窃取和篡改。

  6. 使用场景 WebSocket被广泛应用于需要实时通信的应用场景，例如：

  • 即时聊天应用：通过WebSocket实现实时的文字、语音、视频聊天功能。
  • 在线协作和共享：通过WebSocket实现实时的协作和共享功能，如实时编辑、实时白板等。
  • 多人游戏：通过WebSocket实现多人游戏中的实时通信和互动。
  • 实时监控和远程控制：通过WebSocket实现实时数据的监控和远程设备的控制。
  • 股票行情和实时数据更新：通过WebSocket实时传输股票行情和实时数据，以提供实时更新的服务。

总结： WebSocket是一种实现实时双向通信的网络协议，它可以在客户端和服务器之间建立持久性的连接。WebSocket具有低延迟、高效率的特点，适用于需要实时数据传输和实时互动的应用场景。通过使用WebSocket，可以实现即时聊天、在线协作、多人游戏、实时监控等一系列的实时应用。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/从前慢现在也慢/article/detail/717194
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号