SpringBoot 集成 WebSocket

前言

最近在做一个 WebSocket 通信服务的软件，所以必须跟着学一学。

1、WebSocket 概述

一般情况下，我们的服务器和服务器之间可以发送请求，但是服务器是不能向浏览器去发送请求的。因为设计之初并没有想到以后会出现服务端频繁向客户端发送请求的情况。

全双工的通信协议（WebSocket 最大的特点是浏览器也可以往服务器发请求，服务器也可以向浏览器发请求）。

1.1、浏览器和服务器使用WebSocket通信流程

1. 浏览器发起http请求，请求建立 WebSocket 连接

这里的协议升级就是说，我想通过这个 http 连接去升级为 WebSocket 连接

2. 服务器响应统一协议更改

3. 相互发送数据

升级了协议之后浏览器就可以和服务器相互通信了：

1.2、总结

WebSocket 协议是建立在 tcp 协议基础上的，所以不同语言也都支持
tcp 协议是全双工协议，http 协议基于它是单向的
WebSocket 没有同源限制，所以前后端端口不一致也不影响信息的发送

2、Java 实现 WebSocket 的两种方式

2.1、基于注解实现WebSocket服务器端

服务终端类：

@ServerEndpoint：监听连接（需要传递一个地址参数）
@OnOpen：连接成功
@OnClose：连接关闭
@OnMessage：收到消息

配置类

把 Spring 中的 ServerEndpointExporter 对象注入进来

2.2.1、编写服务终端类

// 监听哪些客户端来连接了WebSocket服务端
// 监听websocket地址 /myWs
@ServerEndpoint("/myWs")
@Component
@Slf4j
public class WebServerEndpoint {

    // 因为可能有多个客户端所以这里需要保证线程安全
    static Map<String,Session> sessionMap = new ConcurrentHashMap<>();

    // 建立连接时执行的操作
    @OnOpen
    public void onOpen(Session session){   // 每个websocket连接对于服务端来说都是一个Session
        sessionMap.put(session.getId(),session);
        log.info("websocket is open");
    }

    /**
     * 收到客户端消息时执行的操作
     * @param text 接受到客户端的消息
     * @return 返回给客户端的消息
     */
    @OnMessage
    public String onMessage(String text){
        log.info("收到一条新消息: " + text);
        return "收到 !";
    }

    // 连接关闭时执行的操作
    @OnClose
    public void onClose(Session session){
        sessionMap.remove(session.getId());
        log.info("websocket is close");
    }

    @Scheduled(fixedRate = 2000) // 每隔2s执行一次
    public static void sendMessage() throws IOException {
        for(String key: sessionMap.keySet()){ // 给所有客户端发送消息
            sessionMap.get(key).getBasicRemote().sendText("beat");
        }
    };
}

注意：这里监听的地址不可以是 "ws" 不然会报错，可能这是关键字吧，毕竟我们的协议就叫 ws 。

2.2.2、编写配置类

// 需要注入Bean的话必须声明为配置类
@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter(){
        return new ServerEndpointExporter();
    }
}

2.2、HTML + JS 实现客户端

在 resources 目录下创建 static/ws-client.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>WebSocket Client</title>
</head>
<script>
    // 客户端和服务器连接的地址(我们服务端监听的地址)
    let ws = new WebSocket("ws://localhost:8080/myWs1")
    ws.onopen=function (){ // 连接打开的时候向服务器发送一条消息
        ws.send("hey man")
    }
    ws.onmessage=function (message) {
        console.log(message.data)
    }
</script>
<body>
<h1>WebSocket</h1>
</body>
</html>

测试：启动 SpringBoot 并访问 localhost:8080/ws-client.html

执行结果：

2.3、基于 Spring 框架实现 WebSocket 服务器端

Spring 提供的类和框架

HttpSessionHandshakeInterceptor（抽象类）：握手拦截器，在握手前后添加操作
AbstractWebSocketHandler（抽象类）：WebSocket 处理程序，监听连接前、中、后
WebSocketConfigurer（接口）：配置程序，比如配置监听哪个端口，配置自定义的握手拦截器，配置我们自定义的处理程序

2.3.1、编写握手拦截器类

/**
 * WebSocket 自定义握手拦截器
 */
@Component
@Slf4j
public class MyInterceptor extends HttpSessionHandshakeInterceptor {
    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
        log.info("握手前");
        log.info("远程地址 => " + request.getRemoteAddress());
        // 保留父类的操作
        return super.beforeHandshake(request, response, wsHandler, attributes);
    }

    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception ex) {
        log.info("完成握手");
        // 完成握手后它就会把 HTTP 协议升级为 WebSocket 协议
        super.afterHandshake(request, response, wsHandler, ex);
    }
}

2.3.2、编写 WebSocket 处理程序

/**
 * WebSocket 自定义处理程序
 */
@Slf4j
@Component
public class MyWsHandler extends AbstractWebSocketHandler {
    // WebSocketSession 对象可以封装一下吧用户的信息封装进去
    private static Map<String, SessionBean> sessionMap = new ConcurrentHashMap<>();
    // 线程安全的int值
    private static AtomicInteger clientIdMaker = new AtomicInteger(0);

    // 连接建立
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        super.afterConnectionEstablished(session);
        // 放在父方法调用之后
        SessionBean sessionBean = new SessionBean(session, clientIdMaker.getAndIncrement());
        sessionMap.put(session.getId(),sessionBean);
        log.info(sessionMap.get(session.getId()) + " connected");
    }

    // 收到消息
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        super.handleTextMessage(session, message);
        log.info(sessionMap.get(session.getId()).getClientId() + " : " + message.getPayload());
    }

    // 传输异常
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        super.handleTransportError(session, exception);
        // 如果异常就关闭 session
        if (session.isOpen())
            session.close();
        sessionMap.remove(session.getId());
    }

    // 连接关闭
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        super.afterConnectionClosed(session, status);
        log.info(sessionMap.get(session.getId()) + " closed");
        sessionMap.remove(session.getId());
    }

    @Scheduled(fixedRate = 2000) // 每隔2s执行一次
    public static void sendMessage() throws IOException {
        for(String key: sessionMap.keySet()){ // 给所有客户端发送消息
            sessionMap.get(key).getWebSocketSession().sendMessage(new TextMessage("beat"));
        }
    }
}

2.3.3、编写配置类

@Configuration
@EnableWebSocket    // 启用 spring 提供的 websocket 功能
public class MyWsConfig implements WebSocketConfigurer {

    @Resource
    MyWsHandler myWsHandler; // 引入我们在MyWsHandler上声明的Bean(@Component)
    @Resource
    MyInterceptor myInterceptor;
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        // 监听的地址
        registry.addHandler(myWsHandler,"/myWs1")
                .addInterceptors(myInterceptor)
                .setAllowedOrigins("*"); // 允许的源
    }
}

修改 html 中的 websocket 地址为 /myWs1

测试：访问 localhost:8080/ws-client.html

3、总结

服务器会和每个客户端维护一个连接：

3、WebSocket 实现多人聊天室

3.1、需求

进入聊天室
群聊功能，任何人说话，所有人都能接受到消息
退出群聊

3.2、实现

3.2.1、编写前端界面和通信逻辑

这里我们通过 js 实现了当用户点击发送按钮的时候，把文本框的内容通过 websocket 发送给指定的地址，我们的后台 websocket 程序收到后会全部群发给所有当前连接的客户端：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>WebSocket Client</title>
</head>
<script>
    // 客户端和服务器连接的地址(我们服务端监听的地址)
    let ws = new WebSocket("ws://localhost:8080/myWs1")
    ws.onopen=function (){ // 连接打开的时候向服务器发送一条消息
        // ws.send("hey man")
    }
    ws.onmessage=function (message) {
        // console.log(message.data)
        document.getElementById("talkMsg").innerHTML = message.data
    }
    function sendMsg(){
        // 发送文本框的信息
        ws.send(document.getElementById("message").value)
        // 发送完清空输入框
        document.getElementById("message").value=""
    }
</script>
<body>
<h1>WebSocket 多人聊天室</h1>
<p style="border: 1px solid black;width: 600px;height: 500px" id="talkMsg"></p>
<input id="message" />
<button id="sendBtn" onclick="sendMsg()">发 送</button>
</body>
</html>

3.2.2、WebSocket 后台处理程序

这里我们编写了一个 sendMessage 的方法来给所有已连接的客户端进行消息的群发，而我们上面的前端代码中设置当 websocket 收到消息后，会把文本域（<p>标签）中的内容替换掉（innerHTML），因为我们的消息是不断累加到 StringBuffer 当中的，而文本域内容的替换开销并不会造成视觉上的卡顿：

/**
 * WebSocket 自定义处理程序
 */
@Slf4j
@Component
public class MyWsHandler extends AbstractWebSocketHandler {
    // WebSocketSession 对象可以封装一下吧用户的信息封装进去
    private static Map<String, SessionBean> sessionMap = new ConcurrentHashMap<>();
    // 线程安全的int值
    private static AtomicInteger clientIdMaker = new AtomicInteger(0);
    //
    private static StringBuffer sb = new StringBuffer();

    // 连接建立
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        super.afterConnectionEstablished(session);
        // 放在父方法调用之后
        SessionBean sessionBean = new SessionBean(session, clientIdMaker.getAndIncrement());
        sessionMap.put(session.getId(),sessionBean);
        log.info(sessionMap.get(session.getId()) + " connected");
        sb.append(sessionBean.getClientId()).append(" 进入了群聊<br/>");
        sendMessage(sessionMap);
    }

    // 收到消息
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        super.handleTextMessage(session, message);
        log.info(sessionMap.get(session.getId()).getClientId() + " : " + message.getPayload());
        sb.append(sessionMap.get(session.getId()).getClientId()).append(" : ").append(message.getPayload()).append("<br/>");
        sendMessage(sessionMap);
    }

    // 传输异常
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        super.handleTransportError(session, exception);
        // 如果异常就关闭 session
        if (session.isOpen())
            session.close();
        sessionMap.remove(session.getId());
    }

    // 连接关闭
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        super.afterConnectionClosed(session, status);
        Integer clientId = sessionMap.get(session.getId()).getClientId();
        log.info(sessionMap.get(session.getId()) + " closed");
        sessionMap.remove(session.getId());
        sb.append(clientId).append("退出了群聊<br/>");
        sendMessage(sessionMap);
    }

    public void sendMessage(Map<String,SessionBean> sessionMap){
        for(String sessionId: sessionMap.keySet()){
            try {
                sessionMap.get(sessionId).getWebSocketSession().sendMessage(new TextMessage(sb.toString()));
            } catch (IOException e) {
                e.printStackTrace();
                log.error(e.getMessage());
            }
        }
    }
}

3.2.3、测试

启动项目，并访问 http://localhost:8080/ws-client.html

可以看到，当我们启动多个客户端（开启多个网页），就会显示多人加入群聊，当任意客户端发送消息后，所有客户端都可以看到，当有客户端退出后，同样所有人都可以看到。

4、WebSocket 使用场景

总的来讲，websocket 主要的特点就是全双工通信吗，它最大的优势就是实时性特别好，因为传统的单工通信我们客户端需要不断的向服务器发送请求，要追求实时性就必须不断频繁发送请求，效率肯定是低下的。而基于 WebSocket 协议的全双工通信的话就不需要了，因为服务端下你在可以主动给客户端发送消息了，就不需要客户端不断的去发送请求了。

比如股价分析，股票什么时候发生变动我们谁也说不清楚，但是难道让客户端每几百毫秒去访问一次服务器吗？那必不可能，最好的办法就是每当股价发生变动，让服务器主动给客户端发送消息，客户端收到消息以后再做更新。这样通信效率一下子刷就上来了。