Spring Websocket教程

在网络通讯领域，我们最熟悉的通讯协议即 HTTP，其在 TCP 的基础上进一步扩展使之更为的简单易用。而 WebSocket 作为一种通讯方式在某些应用场景下也是一种不错的选择。

WebSocket 是一种在单个 TCP 连接上进行全双工通信的通信协议，它提供了一个持久的连接（即长链接），允许客户端和服务器之间进行实时、双向的数据传输。WebSocket 的出现解决了传统的 HTTP 协议的一些限制，特别是对于需要频繁更新的实时应用程序（如聊天应用、在线游戏等）而言，WebSocket 提供了更低的延迟和更高的效率。

Websocket 一个经典应用场景即消息的主动推送，即由后端向主动推送讯息，在通知与聊天场景中应用广泛。本文即着重介绍如何在前后端分离项目中实现后端的消息主动推送。

一、后端配置

1. 依赖导入

老规矩，在后端的 Spring 项目工程中导入对应的依赖，内容如下：

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

2. 配置管理

在项目中注入 WebSocket Bean 对象，代码如下：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

@Configuration
public class WebSocketConfig {

    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

3. 服务启动

新建 WebSocketServer 类用于管理请求的连接与关闭，当接收到新的请求时存入缓存容器 websocketMap。

其中 @ServerEndpoint 注解可以理解为 HTTP 接口中 @RequstMapping 与 @RestController 的复合体， onMessage() 方法为首次连接时返回值，用于响应以便知晓是否连接成功。

同时为了区分不同的连接，这里通过请求参数的 userId 作为唯一标识符，即请求地址格式为：ws://${host}:${port}/websocket?userId=1，用于后续能够精准的推送消息给指定会话。

@Data
@Slf4j
@Component
@ServerEndpoint(value = "/websocket")
public class WebSocketServer {

    /**
     * 会话标识
     */
    private String flag;

    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;


    /**
     * 连接成功回调
     */
    @OnOpen
    public void onOpen(Session session) {
        // 解析验证 URL
        String reqUrl = session.getRequestURI().toString();
        UriComponents uriComponents = UriComponentsBuilder.fromUriString(reqUrl).build();
        Map<String, String> paramMap = uriComponents.getQueryParams().toSingleValueMap();
        if (paramMap.isEmpty()) {
            throw new IllegalArgumentException("userId required");
        }
        this.flag = paramMap.get("userId");
        if (Objects.isNull(flag)) {
            throw new IllegalArgumentException("userId required");
        }
        // 添加会话至容器
        this.session = session;
        SessionManager.SERVER_MAP.put(flag, this);

        try {
            this.sendMessage(session, "Connected successfully");
        } catch (Exception e) {
            log.error("Session onOpen", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 连接关闭回调
     */
    @OnClose
    public void onClose() {
        // 从容器中删除会话
        SessionManager.SERVER_MAP.remove(flag);
    }

    /**
     * 收到客户端消息回调
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("Receive from client, message: {}", message);

        try {
            this.sendMessage(session, "Server receive: " + message);
        } catch (Exception e) {
            log.error("Session onMessage", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 会话异常回调
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("Session onError", error);
        try {
            this.sendMessage(session, "Program went wrong, message: " + error.getMessage());
        } catch (Exception e) {
            log.error("Session onMessage", e);
            throw new RuntimeException(e);
        }
    }


    protected void sendMessage(Session session, String message) throws IOException {
        session.getBasicRemote().sendText(message);
    }
}

4. 会话管理

接下来创建 SessionManager 用于实现会话的管理，通过 Map 缓存的形式将用户会话信息暂存。

其中 send() 的 flag 即建立连接时传入的唯一标识符，sendAll 即为全局广播将推送消息至所有当前的连接的用户。

另外需注意一点，会话最简单的传输方式即纯文本字符串，可使用 Jackson 等类库序列化为 Json 字符串后传输。

public class SessionManager {

    private final static Logger log = LoggerFactory.getLogger(SessionManager.class);

    /**
     * 存放客户端对应的 WebSocket 对象
     */
    protected static final Map<String, WebSocketServer> SERVER_MAP = new ConcurrentHashMap<>();


    /**
     * 发送消息
     */
    public static boolean send(String flag, String message) {
        return batchSend(message, SERVER_MAP.get(flag));
    }

    /**
     * 群发消息
     */
    public static boolean sendAll(String message) {
        return batchSend(message, SERVER_MAP.values().toArray(new WebSocketServer[0]));
    }

    /**
     * 批量发送
     */
    public static boolean batchSend(String message, WebSocketServer... servers) {
        if (Objects.isNull(servers) || servers.length == 0) {
            log.warn("Servers is empty.");
            return false;
        }

        boolean success = true;
        for (WebSocketServer item : servers) {
            try {
                item.sendMessage(item.getSession(), message);
            } catch (IOException e) {
                success = false;
                log.error("send error", e);
            }
        }
        return success;
    }
}

二、前端应用

1. 依赖安装

在 Vue 前端项目中同样需要安装 WebSocket 依赖，命令如下：

npm install vue-websocket

2. 应用监听

完成依赖安装后即可在项目中建立依赖，通过 new WebSocket(url) 即可相对较为简单。

连接建立之后通过 addEventListener 回调函数监听不同的连接事件，代码如下不详细介绍。

created() {
    const host = 127.0.0.1
    const port = 9090
    const socket = new WebSocket(`ws://${host}:${port}/websocket?userId=1`);

    // 添加WebSocket事件监听器
    socket.addEventListener('open', (event) => {
      console.log('WebSocket 连接已打开', event);
    })

    socket.addEventListener('message', (event) => {
      const data = event.data
      console.log('接受到后端信息: ', data)
      if (data !== null) {
        this.$message.success('接受到后端消息: ' + data)
      }
    })

    socket.addEventListener('close', (event) => {
      console.log('WebSocket 连接已关闭', event.data);
    })

    socket.addEventListener('error', (event) => {
      console.error('WebSocket 错误', event);
    })
  }
}