java服务端如何接入WebSocket？

Myth丶恋晨 2024-03-16 11:01 20阅读 0赞

日常工作中，我们都是使用http请求，来进行前后交互，那么我们也会有使用websocket来进行前后交互的时候，那么它俩有什么区别呢？

## http和websocket区别 ##

1.  **WebSocket是双向通信协议，模拟Socket协议，可以双向发送或接受信息**
2.  **HTTP是单向的**
3.  **WebSocket是需要浏览器和服务器握手进行建立连接的**
4.  **而http是浏览器发起向服务器的连接，服务器预先并不知道这个连接**

根据它的特点就可以知道，如果场景中需要服务端向客户端推送消息，那么使用http就是不行的。所以今天就用spring boot 来接入 WebSocket。

## 代码实现 ##

1、首先，我们需要在pom文件内引入它的jar包依赖，如下：

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

这个时候我们就可以直接利用该第三方的API直接代码实现了。

### 新增websocket配置类 ###

因为项目中有http请求，也有websocket请求，那么该怎么去区分请求连接的类型呢？这个时候我们需要增加一个websocket的配置类，来进行拦截websocket的请求。

@Configuration
    public class WebSocketConfig {
    
        /**
         * 注入ServerEndpointExporter，
         * 这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
         */
    
        @Bean
        public ServerEndpointExporter serverEndpointExporter() {
            return new ServerEndpointExporter();
        }
    
    }

注意：不要忘记添加 **@Configuration** 注解，这个类，会自动拦截websocket请求。

### 完整代码 ###

@Component
    @Slf4j
    @ServerEndpoint("/websocket/{userId}")
    public class WebSocketHandler {
        // 接口路径 ws://127.0.0.1:9000/websocket/userId;
        private Session session;
    
        //concurrent包的线程安全Set，用来存放每个客户端对应的WebSocket对象。
        private static CopyOnWriteArraySet<WebSocketHandler> webSocketUtils = new CopyOnWriteArraySet<>();
        // 用来存在线连接数
        private static Map<String, Session> sessionPool = new HashMap<>();
    
        /**
         * 链接成功调用的方法
         */
        @OnOpen
        public void onOpen(Session session, @PathParam(value = "userId") String userId) {
            try {
                this.session = session;
                webSocketUtils.add(this);
                sessionPool.put(userId, session);
                sendOneMessage(userId, "连接成功");
                //加入队列
                QueueUtils.addMap(userId);
    
                log.info("【websocket消息】有新的连接，总数为:" + webSocketUtils.size());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        /**
         * 链接关闭调用的方法
         */
        @OnClose
        public void onClose() {
            try {
                webSocketUtils.remove(this);
                log.info("【websocket消息】连接断开，总数为:" + webSocketUtils.size());
    
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        /**
         * 收到客户端消息后调用的方法
         *
         * @param message
         * @param
         */
        @OnMessage
        public void onMessage(@PathParam(value = "userId") String userId, String message) {
    
            log.info("【websocket消息】收到客户端消息:" + message);
            if (!StringUtils.isEmpty(userId) && !StringUtils.isEmpty(message) && message.equals("ping")) {
                sendOneMessage(userId, "pong");
            } else {
                //1、轮询用户数最少的服务器IP
                String serverIP = QueueUtils.getServer();
                log.info("serverIP为：" + serverIP);
                //2、调用大模型
                WebSocketClient client = JavaClient.getClient(userId, serverIP);
                if (client != null) {
                    JSONObject object = new JSONObject();
                    object.put("", message);
                    String[][] strings = {
       {"aaa"}};
                    object.put("history", strings);
                    object.put("knowledge_base_id", "/home/nemo/aigc/langchain-ChatGLM/vector_store/knowledge_FAISS_20230516_133609");
                    //3、发送问题
                    client.send(object.toString());
    
                }
            }
        }
    
        /**
         * 发送错误时的处理
         *
         * @param session
         * @param error
         */
        @OnError
        public void onError(Session session, Throwable error) {
    
            log.error("用户错误,原因:" + error.getMessage());
            error.printStackTrace();
        }
    
    
        /**
         * 推消息给前端
         *
         * @param userId
         * @param message
         * @return
         */
        public static Runnable sendOneMessage(String userId, String message) {
            Session session = sessionPool.get(userId);
            if (session != null && session.isOpen()) {
                try {
                    log.info("【推给前端消息】 :" + message);
    
                    //高并发下，防止session占用期间，被其他线程调用
                    synchronized (session) {
                        session.getBasicRemote().sendText(message);
                    }
    
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            return null;
        }
    }

### 测试 ###

我们可以使用第三方websocket在线测试工具，[http://www.jsons.cn/websocket/][http_www.jsons.cn_websocket]，连进行测试。

我们可以发送请求连接为：**ws://127.0.0.1:9000/websocket/1**

然后点击Websocket连接，即客户端可以收到服务端的回应：连接成功。

![image.png][]

[http_www.jsons.cn_websocket]: http://www.jsons.cn/websocket/
[image.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/8f4f7adedc384b00b972cb680276b967.png