SpringBoot集成websocket（2）|（websocket服务端实现以及websocket中转实现）

客官°小女子只卖身不卖艺 2023-10-10 18:06 62阅读 0赞

## SpringBoot集成websocket（2）|（websocket服务端实现以及websocket转发实现） ##

--------------------

### 文章目录  SpringBoot集成websocket（2）|（websocket服务端实现以及websocket转发实现）  @\[TOC\]  前言  一、websocket服务端依赖引入  二、websocket服务代码实现  1.WebSocketConfig配置  2.WebSocketServer服务实现  3.WebSocketClient连接第三方客户端实现  总结 ###

章节  
第一章链接: [SpringBoot集成websocket（1）|（websocket客户端实现）][SpringBoot_websocket_1_websocket]  
第二章链接: [SpringBoot集成websocket（2）|（websocket服务端实现以及websocket中转实现）][SpringBoot_websocket_1_websocket]

## 前言 ##

本节主要介绍的是springboot实现websocket的客户端服务端，以及作为中转服务实现客户端长连接服务端，服务端长连接第三方websocket服务的数据传输。

## 一、websocket服务端依赖引入 ##

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

## 二、websocket服务代码实现 ##

### 1.WebSocketConfig配置 ###

> springboot接入websocket需要启用对应的配置

@Configuration
    @EnableWebSocket
    public class WebSocketConfig {
        
        @Bean
        public ServerEndpointExporter serverEndpointExporter(){
        
            return new ServerEndpointExporter();
        }
    }

### 2.WebSocketServer服务实现 ###

> springboot提供对外的websocket接口实现

/**
     * @author Oak
     * Created on 2023/5/15.
     * Description:
     */
    @Component
    @Data
    @Slf4j
    @ServerEndpoint(value = "/ws/{name}")
    public class WebSocketServer {
        
    	// 调用第三方接口插件
        @Autowired
        private GptRestApi gptRestApi;
    
        @Autowired
        public MilvusLargeService largeService;
    
        @Autowired
        public DocumentParagraphRepository paragraphRepository;
    
    
        /**
         * 与某个客户端的连接对话，需要通过它来给客户端发送消息
         */
        private Session session;
    
        /**
         * 标识当前连接客户端的用户名
         */
        private String name;
    
        /**
         * 用于存所有的连接服务的客户端，这个对象存储是安全的
         */
        private static ConcurrentHashMap<String, WebSocketServer> webSocketSet = new ConcurrentHashMap<>();
    
        /**
         * 用于存所有的连接第三方ws服务的客户端
         */
        private static ConcurrentHashMap<String, WebSocketClient> wsClientSet = new ConcurrentHashMap<>();
    
    
        @OnOpen
        public void OnOpen(Session session, @PathParam(value = "name") String name) {
        
            this.session = session;
            this.name = name;
            webSocketSet.put(name, this);
            WebSocketClient webSocketClient = getWebSocketClient(session);
            wsClientSet.put(name, webSocketClient);
            log.info("[web -> server] 连接成功，当前连接人数为：={}", webSocketSet.size());
        }
    
    
        @OnClose
        public void OnClose() {
        
            webSocketSet.remove(this.name);
            wsClientSet.remove(this.name);
            log.info("[web -> server] 退出成功，当前连接人数为：={}", webSocketSet.size());
        }
    
        @OnMessage
        public void OnMessage(String message) {
        
            log.info("[web -> server] 收到消息：{}", message);
            SparkWSClient sparkWSClient = (SparkWSClient) wsClientSet.get(name);
            // ping 消息响应 pong
            if ("ping".equals(message)) {
        
                try {
        
                    session.getBasicRemote().sendText("pong");
                    return;
                } catch (IOException e) {
        
                    log.error("消息发送失败");
                }
            } else {
        
            // 业务消息
                if (sparkWSClient != null && sparkWSClient.isOpen()) {
        
                        sparkWSClient.send(message);
                    } catch (Exception e) {
        
                        log.warn("异常：{}", e.getMessage());
                    }
                    return;
                } else {
         // 重新连接spark websocket
                    WebSocketClient webSocketClient = getWebSocketClient(session);
                    wsClientSet.put(name, webSocketClient);
                }
            }
        }
    
        /**
         * 群发
         *
         * @param message
         */
        public void GroupSending(String message) {
        
            for (String name : webSocketSet.keySet()) {
        
                try {
        
                    webSocketSet.get(name).session.getBasicRemote().sendText(message);
                } catch (Exception e) {
        
                    e.printStackTrace();
                }
            }
        }
    
    // 第三方websocket接口，自己可以模拟写一个
        public WebSocketClient getWebSocketClient(Session session) {
        
            String chaturl = "ws://xxx/chat";
            UserSession build1 = UserSession.builder()
                    .session(session)
                    .name(name)
                    .build();
            WebSocketClient client = new SparkWSClient(URI.create(chaturl ), build1);
            client.connect();
            return client;
        }
    
    }

> userSession代码如下

@Data
    @Builder
    public class UserSession {
        
        public  Session session;
        public String name;
    
        public UserSession(Session session,String name) {
        
            this.session = session;
            this.name = name;
        }
    
        public void sendUserMsg(String msg) {
        
            sendUserMsg(msg, false);
        }
    
        public void sendUserMsg(String msg, boolean close) {
        
            synchronized (session) {
        
                if (!session.isOpen()) {
        
                    return;
                }
                try {
        
                    session.getBasicRemote().sendText(msg);
                    if (close) {
        
                        session.close();
                    }
                } catch (IOException ex) {
        
                    ex.printStackTrace();
                }
            }
        }
    }

### 3.WebSocketClient连接第三方客户端实现 ###

> springboot提供对第三方websocket连接的客户端  
> 实现代码如下

> 思路，两个websocket各自监听自己的数据，想要串起来可以通过消息中间件来实现，但这样无疑会增加系统的复杂性，还可以通过注入Session的方式，与客户端建立连接后，可以通过session进行数据推送，这个时候只要在 连接第三方的websocket的时候注入这个session兑现。那调用第三方的websocketClient就能通过session 将监听到的数据直接返回给客户端。

@Slf4j
    public class SparkWSClient extends WebSocketClient {
        
    
        UserSession userSession;
    
        public SparkWSClient(URI serverUri, UserSession session) {
        
            super(serverUri);
            this.userSession = session;
        }
    
        @Override
        public void onOpen(ServerHandshake serverHandshake) {
        
            log.info("[server-> spark] 连接成功");
        }
    
        @Override
        public void onMessage(String message) {
        
            log.debug("[server-> spark] 收到消息={}...", message);
            //收到远程的第三方返回数据，这里通过注入的session直接返回给客户端
            userSession.sendUserMsg(message);
        }
    
        @Override
        public void onClose(int i, String s, boolean b) {
        
            log.debug("[server-> spark] 退出连接");
            String errMsg = String.format(MVS_ERR_MSG_FORMAT, userSession.getName(), "spark connection closed!");
            userSession.sendUserMsg(errMsg, true);
        }
    
        @Override
        public void onError(Exception e) {
        
            //转发给用户端
            String errMsg = String.format(MVS_ERR_MSG_FORMAT, userSession.getName(), "spark connection refused!");
            userSession.sendUserMsg(errMsg, true);
            log.error("[server-> spark] 连接错误={}", e.getMessage());
        }
    }

## 总结 ##

本文主要介绍websocket客户端、服务端的实现，以及客户端通过websocket连接服务端，服务端连接第三方的websocket接口，并且将第三个方数据直接返回给客户端，起一个中间websockey代理服务作用，实现数据的中转。

第一章链接: [SpringBoot集成websocket（1）|（websocket客户端实现）][SpringBoot_websocket_1_websocket]  
第二章链接: [SpringBoot集成websocket（2）|（websocket服务端实现以及websocket中转实现）][SpringBoot_websocket_1_websocket]

[SpringBoot_websocket_1_websocket]: https://blog.csdn.net/Oaklkm/article/details/130705555?spm=1001.2014.3001.5501