SpringBoot 集成 WebSocket 实现消息群发推送

一. 什么是 WebSocket

WebSocket 是一种全新的协议。它将 TCP 的 Socket（套接字）应用在了web page上，从而使通信双方建立起一个保持在活动状态的连接通道，并且属于全双工通信（双方同时进行双向通信）。

二. WebSocket 的特点

WebSocket 的最大特点是，服务器可以主动向客户端推送信息，客户端也可以主动向服务器发送信息，是真正的双向平等对话，属于服务器推送技术的一种。

其他特点包括：

建立在 TCP 协议之上，服务器端的实现比较容易。
与 HTTP 协议有着良好的兼容性。默认端口是 80 和 443 ，并且握手阶段采用 HTTP 协议，因此握手时不容易屏蔽，能通过各种 HTTP 代理服务器。
数据格式比较轻量，性能开销小，通信高效。
可以发送文本，也可以发送二进制数据。
没有同源限制，客户端可以与任意服务器通信。
协议标识符是ws（如果加密，则为wss），服务器网址就是 URL。

三. WebSocket 的优势

目前，很多网站都使用 Ajax 轮询方式来实现消息推送。

轮询是指在特定的的时间间隔（如每秒），由浏览器对服务器发出 HTTP 请求，然后由服务器返回最新的数据给客户端的浏览器。

这种传统的模式带来了很明显的缺点，即浏览器需要不断地向服务器发出请求，然而 HTTP 请求可能包含较长的头部，其中真正有效的数据可能只是很小的一部分，会浪费很多的带宽资源。

而 WebSocket 允许服务端主动向客户端推送数据，这就使得客户端和服务器之间的数据交换变得更加简单。在 WebSocket API 中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输。

四. WebSocket 的实现

1. Java 后端实现

项目整体结构如图：

实现步骤

添加需要的依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- webSocket -->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<!-- hutool -->
<dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.2.5</version>
</dependency>
<!-- thymeleaf-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-thymeleaf</artifactId>
</dependency>

前端使用了 thymeleaf 模板引擎，需要在配置文件 application.yml 中对 thymeleaf 进行配置：

# thymeleaf
spring:thymeleaf:check-template-location: truesuffix: .htmlencoding: UTF-8servlet:content-type: text/htmlmode: HTML5cache: false
server:port: 9999

在 WebsocketApplication 中继承 SpringBootServletInitializer，重写 configure 方法：

package com.test.websocket;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.servlet.support.SpringBootServletInitializer;@SpringBootApplication
public class WebsocketApplication extends SpringBootServletInitializer {@Overrideprotected SpringApplicationBuilder configure(SpringApplicationBuilder application) {return application.sources(WebsocketApplication.class);}public static void main(String[] args) {SpringApplication.run(WebsocketApplication.class, args);}
}

新建配置类 WebSocketConfig，开启 WebSocket 支持

package com.test.websocket.config;/*** @author: jichunyang**/
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;@Configuration
public class WebSocketConfig {@Beanpublic ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter();}}

实现核心服务类 WebSocketServer

package com.test.websocket.config;import java.io.IOException;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import org.springframework.stereotype.Component;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;/*** @author: jichunyang* 因为 WebSocket 是类似客户端服务端的形式(采用 ws 协议)，* 这里的 WebSocketServer 相当于一个 ws 协议的 Controller**/
@ServerEndpoint("/imserver/{userId}")
@Component
public class WebSocketServer {static Log log = LogFactory.get(WebSocketServer.class);/*** 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的*/private static int onlineCount = 0;/*** concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象*/private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();/*** 与某个客户端的连接会话，需要通过它来给客户端发送数据*/private Session session;/*** 接收userId*/private String userId = "";/*** 连接建立成功调用的方法*/@OnOpenpublic void onOpen(Session session, @PathParam("userId") String userId) {this.session = session;this.userId = userId;if (webSocketMap.containsKey(userId)) {webSocketMap.remove(userId);//加入set中webSocketMap.put(userId, this);} else {//加入set中webSocketMap.put(userId, this);//在线数加1addOnlineCount();}log.info("用户连接:" + userId + ",当前在线人数为:" + getOnlineCount());try {sendMessage("连接成功");} catch (IOException e) {log.error("用户:" + userId + ",网络异常!");}}/*** 连接关闭调用的方法*/@OnClosepublic void onClose() {if (webSocketMap.containsKey(userId)) {//从set中删除webSocketMap.remove(userId);subOnlineCount();}log.info("用户退出:" + userId + ",当前在线人数为:" + getOnlineCount());}/*** 收到客户端消息后调用的方法** @param message 客户端发送过来的消息*/@OnMessagepublic void onMessage(String message, Session session) {log.info("用户消息:" + userId + ",报文:" + message);//可以群发消息if (StrUtil.isNotBlank(message)) {try {//解析发送的报文JSONObject jsonObject = new JSONObject(message);//追加发送人(防止串改)jsonObject.put("fromUserId", this.userId);String toUserId = jsonObject.getStr("toUserId");//传送给对应toUserId用户的websocketif (StrUtil.isNotBlank(toUserId) && webSocketMap.containsKey(toUserId)) {webSocketMap.get(toUserId).sendMessage(jsonObject.toJSONString(4));} else {//否则不在这个服务器上log.error("请求的userId:" + toUserId + "不在该服务器上");}} catch (Exception e) {e.printStackTrace();}}}/*** @param session* @param error*/@OnErrorpublic void onError(Session session, Throwable error) {log.error("用户错误:" + this.userId + ",原因:" + error.getMessage());error.printStackTrace();}/*** 实现服务器主动推送*/public void sendMessage(String message) throws IOException {this.session.getBasicRemote().sendText(message);}public static void sendToUser(List<String> persons, String message) {persons.forEach(userId -> {try {sendInfo(message, userId);} catch (IOException e) {e.printStackTrace();}});}/*** 发送自定义消息*/public static void sendInfo(String message, @PathParam("userId") String userId) throws IOException {log.info("发送消息到:" + userId + "，报文:" + message);if (StrUtil.isNotBlank(userId) && webSocketMap.containsKey(userId)) {webSocketMap.get(userId).sendMessage(message);} else {log.error("用户" + userId + ",不在线！");}}public static synchronized int getOnlineCount() {return onlineCount;}public static synchronized void addOnlineCount() {WebSocketServer.onlineCount++;}public static synchronized void subOnlineCount() {WebSocketServer.onlineCount--;}
}

新建推送信息的 DemoController

package com.test.websocket.controller;import com.test.websocket.config.WebSocketServer;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.ModelAndView;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;/*** @author: jichunyang* **/
@RestController
public class DemoController {@GetMapping("/index")public ResponseEntity<String> index(){return ResponseEntity.ok("请求成功");}// 客户端进行连接通信@GetMapping("/client")public ModelAndView client(){return new ModelAndView("websocket");}@RequestMapping("/pushMsgToUsers")public ResponseEntity<String> pushMsgToUsers(String message, String toUserIds) throws IOException {List<String> persons = Arrays.asList(toUserIds.split(","));WebSocketServer.sendToUser(persons, message);return ResponseEntity.ok("服务器信息发送成功！发送目标用户id:" + toUserIds);}
}

2. 前端页面实现

在 resources 文件夹下新增 templates 目录，用于存放模板文件，新建 websocket.html

<!DOCTYPE html>
<html>
<head><meta charset="utf-8"><title>websocket通讯</title>
</head>
<script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.js"></script>
<script>var socket;function openSocket() {if(typeof(WebSocket) == "undefined") {console.log("您的浏览器不支持WebSocket");}else{console.log("您的浏览器支持WebSocket");//实现化WebSocket对象，指定要连接的服务器地址与端口,建立连接var socketUrl="http://localhost:9999/imserver/" + $("#clientUserId").val();socketUrl=socketUrl.replace("https","ws").replace("http","ws");console.log(socketUrl);if(socket!=null){socket.close();socket=null;}socket = new WebSocket(socketUrl);//打开事件socket.onopen = function() {console.log("websocket 已打开");};//获得消息事件socket.onmessage = function(msg) {//发现消息进入,处理前端触发逻辑console.log(msg.data);document.getElementById('contentText').innerHTML += msg.data + '<br/>';};//关闭事件socket.onclose = function() {console.log("websocket 已关闭");};//发生了错误事件socket.onerror = function() {console.log("websocket 发生了错误");}}}</script>
<body>
<p>【clientUserId】：
<div><input id="clientUserId" name="clientUserId" type="text" value="1"></div>
<p>【收到的消息】：
<div id="contentText"></div>
<br/>
<p>【操作】：
<div style="color:blue;cursor:pointer;"><a onclick="openSocket()">开启socket</a></div>
</body>
</html>

3. 测试消息群发

运行 SpringBoot 程序，打开浏览器，新建3个标签页，访问地址：

http://localhost:9999/client

在3个页面里面输入不同的clientUserId，分别为1，2，3，并点击 “开启socket”，可以看到控制台均显示连接成功。

此时后台日志会输出用户连接和在线人数的信息

调用 /pushMsgToUsers 接口，给 id 为 1、2、3、4的用户群发消息：

http://localhost:9999/pushMsgToUsers?toUserIds=1,2,3,4&message=发送群发消息

可以发现，3个页面都收到了该消息，由于 id 为 4 的用户不在线，所以无法收到该条消息是正常的。