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

SpringBoot集成WebSocket实现客户端与服务端通信_websocket客户端和服务端

作者：码创造者 | 2024-07-28 03:18:21

赞

踩

websocket客户端和服务端

话不多说，直接上代码看效果！

一、服务端：

1、引用依赖

  1. <dependency>
  2.     <groupId>org.springframework.boot</groupId>
  3.     <artifactId>spring-boot-starter-websocket</artifactId>
  4. </dependency>

2、添加配置文件 WebSocketConfig

  1. @Configuration
  2. public class WebSocketConfig{
  3.     @Bean
  4.     public ServerEndpointExporter serverEndpointExporter() {
  5.         return new ServerEndpointExporter();
  6.     }
  7. }

3、编写WebSocket服务端接收、发送功能

  声明接口代码：

  1. public interface IWebSocketService {
  2.  
  3.     void sendMessage(String message, List<String> toSids);
  4.  
  5.     String getMessage(String msg);
  6. }

  实现类代码：

  1.  
  2. @Slf4j
  3. @Service
  4. @ServerEndpoint("/api/websocket/{sid}")
  5. public class IWebSocketServiceImpl implements IWebSocketService {
  6.     @Autowired
  7.     RabbitTemplate rabbitTemplate;
  8.  
  9.     //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
  10.     private static AtomicInteger onlineCount = new AtomicInteger(0);
  11.     //concurrent包的线程安全Set，用来存放每个客户端对应的WebSocket对象。
  12.     private static CopyOnWriteArraySet<IWebSocketServiceImpl> webSocketSet = new CopyOnWriteArraySet<>();
  13.  
  14.     //与某个客户端的连接会话，需要通过它来给客户端发送数据
  15.     private Session session;
  16.     //接收sid
  17.     private String sid = "";
  18.  
  19.     @Override
  20.     public void sendMessage(String message, List<String> toSids) {
  21.         log.info("推送消息到客户端 " + toSids + "，推送内容:" + message);
  22.  
  23.         for (IWebSocketServiceImpl item : webSocketSet) {
  24.             try {
  25.                 if (CollectionUtils.isEmpty(toSids)) {
  26.                     item.sendMessage(message);
  27.                 } else if (toSids.contains(item.sid)) {
  28.                     item.sendMessage(message);
  29.                 }
  30.             } catch (IOException e) {
  31.                 continue;
  32.             }
  33.         }
  34.     }
  35.  
  36.     @Override
  37.     public String getMessage(String msg) {
  38.         return null;
  39.     }
  40.  
  41.  
  42.     /**
  43.      * 连接建立成功调用的方法
  44.      */
  45.     @OnOpen
  46.     public void onOpen(Session session, @PathParam("sid") String sid) {
  47.         this.session = session;
  48.         webSocketSet.add(this);     // 加入set中
  49.         this.sid = sid;
  50.         addOnlineCount();           // 在线数加1
  51.         try {
  52. //            sendMessage("conn_success");
  53.             log.info("有新客户端开始监听,sid=" + sid + ",当前在线人数为:" + getOnlineCount());
  54.         } catch (Exception e) {
  55.             log.error("websocket IO Exception");
  56.         }
  57.     }
  58.  
  59.     /**
  60.      * 连接关闭调用的方法
  61.      */
  62.     @OnClose
  63.     public void onClose() {
  64.         webSocketSet.remove(this);  // 从set中删除
  65.         subOnlineCount();              // 在线数减1
  66.         // 断开连接情况下，更新主板占用情况为释放
  67.         log.info("释放的sid=" + sid + "的客户端");
  68.         releaseResource();
  69.     }
  70.  
  71.     @OnMessage
  72.     public void onMessage(String message, Session session) {
  73.         log.info("收到来自客户端 sid=" + sid + " 的信息:" + message);
  74.         //此处为MQ接收功能，如果不使用则可以屏蔽,如果有需要可使用
  75. //        if (rabbitTemplate == null) {
  76. //            ApplicationContext context = springUtils.getApplicationContext();
  77. //            rabbitTemplate = context.getBean(RabbitTemplate.class);
  78. //        }
  79. //        rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_TOPICS_INFORM, //RabbitmqConfig.ROUTINGKEY_RETURN, message);
  80. //        rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_TOPICS_INFORM, //RabbitmqConfig.ROUTINGKEY_RETURN, message);
  81.         // 群发消息
  82. //        List<String> sids = new ArrayList<>();
  83. //        for (WebSocketServer item : webSocketSet) {
  84. //            sids.add(item.sid);
  85. //        }
  86. //        try {
  87. //            sendMessage("客户端 " + this.sid + "发布消息：" + message, sids);
  88. //        } catch (IOException e) {
  89. //            e.printStackTrace();
  90. //        }
  91.     }
  92.  
  93.     private void releaseResource() {
  94.         // 这里写释放资源和要处理的业务
  95.         log.info("有一连接关闭！当前在线人数为" + getOnlineCount());
  96.     }
  97.  
  98.     /**
  99.      * 发生错误回调
  100.      */
  101.     @OnError
  102.     public void onError(Session session, Throwable error) {
  103.         log.error(session.getAsyncRemote() + "客户端发生错误");
  104.         error.printStackTrace();
  105.     }
  106.  
  107.     /**
  108.      * 实现服务器主动推送消息到 指定客户端
  109.      */
  110.     public void sendMessage(String message) throws IOException {
  111.         this.session.getAsyncRemote().sendText(message);
  112.     }
  113.  
  114.     /**
  115.      * 获取当前在线人数
  116.      *
  117.      * @return
  118.      */
  119.     public int getOnlineCount() {
  120.         return onlineCount.get();
  121.     }
  122.  
  123.     /**
  124.      * 当前在线人数 +1
  125.      *
  126.      * @return
  127.      */
  128.     public void addOnlineCount() {
  129.         onlineCount.getAndIncrement();
  130.     }
  131.  
  132.     /**
  133.      * 当前在线人数 -1
  134.      *
  135.      * @return
  136.      */
  137.     public void subOnlineCount() {
  138.         onlineCount.getAndDecrement();
  139.     }
  140.  
  141.     /**
  142.      * 获取当前在线客户端对应的WebSocket对象
  143.      *
  144.      * @return
  145.      */
  146.     public CopyOnWriteArraySet<IWebSocketServiceImpl> getWebSocketSet() {
  147.         return webSocketSet;
  148.     }
  149.  
  150. }

4、如果不需要实现客户端功能，此处可选择前端调用，奉上代码

  1. <!DOCTYPE html>
  2. <html>
  3.  
  4. <head>
  5.     <meta charset="utf-8">
  6.     <title>Java后端WebSocket的Tomcat实现</title>
  7.     <script src="https://cdn.bootcdn.net/ajax/libs/jquery/3.6.0/jquery.min.js"></script>
  8. </head>
  9.  
  10. <body>
  11.     <div id="message"></div>
  12.     <hr />
  13.     <div id="main"></div>
  14.     <div id="client"></div>
  15.     <input id="text" type="text" />
  16.     <button onclick="send()">发送消息</button>
  17.     <hr />
  18.     <button onclick="closeWebSocket()">关闭WebSocket连接</button>
  19. </body>
  20. <script type="text/javascript">
  21.  
  22.     var cid = Math.floor(Math.random() * 100); // 随机生成客户端id
  23.     document.getElementById('client').innerHTML += "客户端 id = " + cid + '<br/>';
  24.  
  25.     var websocket = null;
  26.  
  27.     //判断当前浏览器是否支持WebSocket
  28.     if ('WebSocket' in window) {
  29.         // 改成你的地址
  30.         websocket = new WebSocket("ws://127.0.0.1:9204/api/websocket/" + cid);
  31.     } else {
  32.         alert('当前浏览器 Not support websocket')
  33.     }
  34.  
  35.     //连接发生错误的回调方法
  36.     websocket.onerror = function () {
  37.         setMessageInnerHTML("websocket.onerror: WebSocket连接发生错误");
  38.     };
  39.  
  40.     //连接成功建立的回调方法
  41.     websocket.onopen = function () {
  42.         setMessageInnerHTML("websocket.onopen: WebSocket连接成功");
  43.     }
  44.  
  45.     //接收到消息的回调方法
  46.     websocket.onmessage = function (event) {
  47.         setMessageInnerHTML("websocket.onmessage: " + event.data);
  48.     }
  49.  
  50.     //连接关闭的回调方法
  51.     websocket.onclose = function () {
  52.         setMessageInnerHTML("websocket.onclose: WebSocket连接关闭");
  53.     }
  54.  
  55.     //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
  56.     window.onbeforeunload = function () {
  57.         closeWebSocket();
  58.     }
  59.  
  60.     //将消息显示在网页上
  61.     function setMessageInnerHTML(innerHTML) {
  62.         document.getElementById('message').innerHTML += innerHTML + '<br/>';
  63.     }
  64.  
  65.     //关闭WebSocket连接
  66.     function closeWebSocket() {
  67.         websocket.close();
  68.         alert('websocket.close: 关闭websocket连接')
  69.     }
  70.  
  71.     //发送消息
  72.     function send() {
  73.         var message = document.getElementById('text').value;
  74.         try {
  75.             websocket.send('{"msg":"' + message + '"}');
  76.             setMessageInnerHTML("websocket.send: " + message);
  77.         } catch (err) {
  78.             console.error("websocket.send: " + message + " 失败");
  79.         }
  80.     }
  81. </script>
  82.  
  83. </html>

二、客户端：

1、引用依赖

  1. <dependency>
  2.     <groupId>org.java-websocket</groupId>
  3.     <artifactId>Java-WebSocket</artifactId>
  4.     <version>1.3.8</version>
  5. </dependency>

2、自定义WebSocket客户端，继承WebSocketClient类，实现发送、接收等功能

  1.  
  2. /**
  3.  * 自定义WebSocket客户端
  4.  */
  5. public class CustomizedWebSocketClient extends WebSocketClient {
  6.     private static final Logger logger = LoggerFactory.getLogger(CustomizedWebSocketClient.class);
  7.  
  8.     //用来接收数据
  9.     private String excptMessage;
  10.  
  11.     /**
  12.      * 线程安全的Boolean -是否受到消息
  13.      */
  14.     public AtomicBoolean hasMessage = new AtomicBoolean(false);
  15.  
  16.     /**
  17.      * 线程安全的Boolean -是否已经连接
  18.      */
  19.     private AtomicBoolean hasConnection = new AtomicBoolean(false);
  20.  
  21.     /**
  22.      * 构造方法
  23.      *
  24.      * @param serverUri
  25.      */
  26.     public CustomizedWebSocketClient(URI serverUri) {
  27.         super(serverUri);
  28.         logger.info("CustomizeWebSocketClient init:" + serverUri.toString());
  29.     }
  30.  
  31.     /**
  32.      * 打开连接是方法
  33.      *
  34.      * @param serverHandshake
  35.      */
  36.     @Override
  37.     public void onOpen(ServerHandshake serverHandshake) {
  38.         logger.info("CustomizeWebSocketClient onOpen");
  39.     }
  40.  
  41.     /**
  42.      * 收到消息时
  43.      *
  44.      * @param s
  45.      */
  46.     @Override
  47.     public void onMessage(String s) {
  48.         hasMessage.set(true);
  49.         logger.info("CustomizeWebSocketClient onMessage:" + s);
  50.     }
  51.  
  52.     public void sendMessage(String message){
  53.         this.send(message);
  54.         logger.info("已发送消息：" + message);
  55.     }
  56.  
  57.     /**
  58.      * 当连接关闭时
  59.      *
  60.      * @param i
  61.      * @param s
  62.      * @param b
  63.      */
  64.     @Override
  65.     public void onClose(int i, String s, boolean b) {
  66.         this.hasConnection.set(false);
  67.         this.hasMessage.set(false);
  68.         logger.info("CustomizeWebSocketClient onClose:" + s);
  69.     }
  70.  
  71.     /**
  72.      * 发生error时
  73.      *
  74.      * @param e
  75.      */
  76.     @Override
  77.     public void onError(Exception e) {
  78.         logger.error("CustomizeWebSocketClient onError:" + e);
  79.     }
  80.  
  81.     @Override
  82.     public void connect() {
  83.         if(!this.hasConnection.get()){
  84.             super.connect();
  85.             hasConnection.set(true);
  86.         }
  87.     }
  88.  
  89.     //获取接收到的信息
  90.     public String getExcptMessage() {
  91.         if(excptMessage != null){
  92.             String message = new String(excptMessage);
  93.             excptMessage = null;
  94.             return message;
  95.         }
  96.         return null;
  97.     }
  98. }
  99.  
  100.

3、创建连接封装类，uri对应socket服务的ip和端口号

  1. public class WebSocketClientSingleton {
  2.     private static CustomizedWebSocketClient client;
  3.     private WebSocketClientSingleton(){}
  4.  
  5.     public static CustomizedWebSocketClient getInstance() {
  6.         if (client == null || client.isClosed()) {
  7.             String webSocketUri = "ws://localhost:9204/api/websocket/12345678";
  8.             try {
  9.                 //实例WebSocketClient对象，并连接到WebSocket服务端
  10.                 client = new CustomizedWebSocketClient(new URI(webSocketUri));
  11.                 client.connect();
  12.                 //等待服务端响应
  13.                 while (!client.getReadyState().equals(WebSocket.READYSTATE.OPEN)) {
  14.                     System.out.println("连接中···请稍后");
  15.                     Thread.sleep(1000);
  16.                 }
  17.             } catch (URISyntaxException e) {
  18.                 e.printStackTrace();
  19.             } catch (InterruptedException e) {
  20.                 throw new RuntimeException(e);
  21.             }
  22. //            return client;
  23.         }
  24.         return client;
  25.     }
  26.  
  27.     public static void closeClient(CustomizedWebSocketClient client){
  28.         client.close();
  29.     }
  30. }

三、测试

  1. @Autowired
  2.     private IWebSocketService webSocketService;

调用代码：

  1. //懒汉模式创建客户端连接
  2. CustomizedWebSocketClient customizedWebSocketClient = WebSocketClientSingleton.getInstance();
  3. //客户端发送消息
  4. customizedWebSocketClient.sendMessage("----客户端发送消息啦-----1123");
  5. //服务端发送消息
  6. webSocketService.sendMessage("服务端发送消息啦----223333",null);
  7. //关闭连接
  8. WebSocketClientSingleton.closeClient(customizedWebSocketClient);

控制台打印效果：

总结完毕

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

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

  

闽ICP备14008679号