SpringBoot系列之搭建WebSocket应用

SpringBoot系列之@ServerEndpoint方式开发WebSocket应用。在实时的数据推送方面，经常会使用WebSocket或者MQTT来实现，WebSocket是一种不错的方案，只需要建立连接，服务端和客户端就可以进行双向的数据通信。很多网站的客户聊天，也经常使用WebSocket技术来实现。

WebSocket简介

WebSocket是一种建立在TCP协议上的一种网络协议，与Http协议类似，端口都是80或者443，协议标识符是ws、如果是加密安全的就是wss，这个和http/https有点类似。WebSocket 连接以 HTTP 请求/响应握手开始，连接成功后，客户端可以向服务端发送消息，反之亦可，WebSocket协议支持二进制数据和文本字符串的传输。因为客户端和服务端之间只有一条TCP通信连接，以后所有的请求都使用这条连接，所以Websocket也是属于长连接。下面给出WebSocket通讯示意图：

WebSocket官网给出的HTTP和WebSocket的对比图：https://websocket.org/guides/road-to-websockets

实验环境准备

• JDK 1.8
• SpringBoot 3.3.0
• Maven 3.3.9
• 开发工具
  • IntelliJ IDEA
  • smartGit

新建WebSocket项目

在idea里新建一个module，选择Spring Initializr项目，默认选择Spring官网的https://start.spring.io

选择需要的依赖，这里可以选择Springboot集成的WebSocket starter

生成项目后，检查一下对应Maven配置文件中是否有加上spring-boot-starter-websocket

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

如果不是Springboot项目，加入spring-websocket即可

 <dependency><groupId>org.springframework</groupId><artifactId>spring-websocket</artifactId><version>5.2.1.RELEASE</version><scope>compile</scope></dependency>

创建ServerEndpoint

创建一个Websocket的ServerEndpoint类，这个类是为了创建一个WebSocket服务端，这个类使用线程安全的CopyOnWriteArrayList集合来存储所有的WebSocket对象，再自定义一个socketClientCode，目的是为了客户端只和对应的服务端通信，客户端建立连接会进入onOpen方法，发送消息会调用onMessage方法

package com.example.springboot.websocket.message;import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.CopyOnWriteArrayList;@ServerEndpoint("/ws/webSocketServer")
@Component
@Slf4j
public class WebSocket {private static final String PREFIX = "socketClient=";private String socketClientCode;private static CopyOnWriteArrayList<WebSocket> webSocketSet = new CopyOnWriteArrayList<>();private Session session ;@OnOpenpublic void onOpen(Session session) {this.session = session;webSocketSet.add(this);log.info("open a webSocket {}, online num: {}",getSocketClientCode(), getOnlineNum());}@OnClosepublic void onClose() {webSocketSet.remove(this);log.error("close a webSocket {}, online num:{}", getSocketClientCode(), getOnlineNum());printOnlineClientCode();}@OnErrorpublic void onError(Session session, Throwable error) {webSocketSet.remove(this);log.error("webSocket error {}, {}, online num:{}", error, getSocketClientCode(), getOnlineNum());printOnlineClientCode();}@OnMessagepublic void onMessage(String message, Session session) {log.info("receive message from client:{}", message);// 业务实现if (message.startsWith(PREFIX)) {String socketClientCode = message.substring(PREFIX.length());this.setSocketClientCode(socketClientCode);sendMessage(message);printOnlineClientCode();} else {sendMessage(message);}}/*** 发送消息** @Date 2024/06/19 16:36* @Param [message]* @return void*/public void sendMessage(String message) {if (!this.session.isOpen()) {log.warn("webSocket is close");return;}try {this.session.getBasicRemote().sendText(message);} catch (IOException e) {log.error("sendMessage exception:{}", e);}}/*** 给客户端发送消息** @Date 2024/06/19 16:37* @Param [message, socketClientCode]* @return void*/public void sendMessageToClient(String message, String socketClientCode) {log.info("send message to client, message:{}, clientCode:{}", message, socketClientCode);printOnlineClientCode();webSocketSet.stream().forEach(ws -> {if (StrUtil.isNotBlank(socketClientCode) && StrUtil.isNotBlank(ws.getSocketClientCode()) && ws.getSocketClientCode().equals(socketClientCode)) {ws.sendMessage(message);}});}/*** 群发消息** @Date 2024/06/19 16:37* @Param [message]* @return void*/public void fanoutMessage(String message) {webSocketSet.forEach(ws -> {ws.sendMessage(message);});}private static synchronized int getOnlineNum() {return webSocketSet.size();}private void printOnlineClientCode() {webSocketSet.stream().forEach(ws -> {log.info("webSocket online:{}", ws.getSocketClientCode());});}public String getSocketClientCode() {return socketClientCode;}public void setSocketClientCode(String socketClientCode) {this.socketClientCode = socketClientCode;}
}

加上ServerEndpointExporter

在Springboot项目中为了能扫描到所有的ServerEndpoint，需要注入一个ServerEndpointExporter，这个类能扫描项目里所有的ServerEndpoint类，不加的话，客户端会一直连不上服务端

package com.example.springboot.websocket.configuration;import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;@Configuration
public class WebSocketConfiguration {@Beanpublic ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter();}}

启动SpringBoot项目，一个WebSocket服务端就建立好了。网上找一个websocket测试网站，https://www.wetools.com/websocket，测试一下服务是否正常，如图：

前端WebSocket客户端

写一个WebSocket调用的客户端，启动服务器，换一下WebSocket的地址，ws://127.0.0.1:8080/ws/webSocketServer，如果是https的，就换成wss://127.0.0.1:8080/ws/webSocketServer

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>webSocket</title><style type="text/css"></style>
</head>
<body>
<h1>WebSocket Demo</h1>
<input type="button" onclick="websocket.send('666666')" value="点我发消息"/>
</body>
<script type="application/javascript">var websocket = {send: function (str) {}};window.onload = function () {if (!'WebSocket' in window) return;webSocketInit();};function webSocketInit() {websocket = new WebSocket("ws://127.0.0.1:8080/ws/webSocketServer");//建立连接websocket.onopen = function () {websocket.send("socketClient=666");console.log("成功连接到服务器");};//接收到消息websocket.onmessage = function (event) {console.log(event.data);};//连接发生错误websocket.onerror = function () {alert("WebSocket连接发生错误");};//连接关闭websocket.onclose = function () {alert("WebSocket连接关闭");};//监听窗口关闭window.onbeforeunload = function () {websocket.close()};}
</script>
</html>

走一个，在浏览器按F12，看看日志

后端的日志打印：

服务端给客户端发送消息

给客户端发送消息，为了只给对应的客户端发送消息，这里加上一个校验，只给注册的客户端发送

/*** 给客户端发送消息** @Date 2024/06/19 16:37* @Param [message, socketClientCode]* @return void*/
public void sendMessageToClient(String message, String socketClientCode) {log.info("send message to client, message:{}, clientCode:{}", message, socketClientCode);printOnlineClientCode();webSocketSet.stream().forEach(ws -> {if (StrUtil.isNotBlank(socketClientCode) && StrUtil.isNotBlank(ws.getSocketClientCode()) && ws.getSocketClientCode().equals(socketClientCode)) {ws.sendMessage(message);}});
}

写一个API接口：

package com.example.springboot.websocket.rest;import cn.hutool.json.JSONUtil;
import com.example.springboot.websocket.dto.WebSocketDto;
import com.example.springboot.websocket.message.WebSocket;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.web.bind.annotation.*;import javax.annotation.Resource;@RestController
@RequestMapping("/api")
@Slf4j
public class WebSocketApiController {@Resource@Qualifier("webSocket")private WebSocket webSocket;@PostMapping@RequestMapping("/sendMessage")public ResultBean<Boolean> sendMessage(@RequestBody WebSocketDto sendDto) {log.info("webSocket发送消息给客户端:{}", JSONUtil.toJsonStr(sendDto));try {webSocket.sendMessageToClient(sendDto.getMessage(), sendDto.getSocketClient());return ResultBean.ok(true);} catch (Exception e) {log.error("发送WebSocket消息异常:{}", e);return ResultBean.badRequest("发送WebSocket消息异常", false);}}}