当前位置： 首页 > news >正文

【SpringBoot3】双向实时通讯 websocket

news 来源：原创 2024/9/19 21:00:42

文章目录

    • 一、Websocket使用步骤
    • 二、示例1：继承抽象类 `AbstractWebSocketHandler`
      • 后端代码
      • 前端代码
    • 三、示例2：使用注解`@ServerEndpoint`
      • 后端代码
      • 前端代码
    • 四、前端代码封装

一、Websocket使用步骤

在Spring Boot中使用WebSocket是一个常见的需求，因为WebSocket提供了一种在客户端和服务器之间进行全双工通信的方式。Spring Boot通过Spring的WebSocket支持，使得在应用程序中集成WebSocket变得非常简单。

以下是在Spring Boot中使用WebSocket的基本步骤：

1、添加依赖：
首先，你需要在你的pom.xml（如果你使用Maven）或build.gradle（如果你使用Gradle）中添加WebSocket的依赖。对于Spring Boot项目，你通常只需要添加spring-boot-starter-websocket依赖。

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

2、配置WebSocket：
接下来，你需要创建一个配置类实现接口WebSocketConfigurer来启用WebSocket，并配置相关的端点和消息代理。使用@EnableWebSocket注解来启用WebSocket功能。

3、创建控制器：
一旦WebSocket被配置，你就可以创建一个控制器来处理WebSocket消息。
创建一个class类，集成抽象类AbstractWebSocketHandler，或者使用注解@ServerEndpoint来声明Websocket Endpoint

4、客户端连接：
最后，你需要在客户端连接到WebSocket服务器。这可以通过使用WebSocket API来实现。

二、示例1：继承抽象类 AbstractWebSocketHandler

后端代码

1、创建Websocket处理类Ws1Handler，继承抽象类AbstractWebSocketHandler

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.AbstractWebSocketHandler;
import org.thymeleaf.util.StringUtils;import static com.hj.springboot3.ws.demo1.Ws1Pool.broadcast;/*** websocket事件处理* <p>* 链接：/ws/demo1?userId=xxxx*/
public class Ws1Handler extends AbstractWebSocketHandler {private static final Logger logger = LoggerFactory.getLogger(Ws1Handler.class);@Overridepublic void afterConnectionEstablished(WebSocketSession session) throws Exception {Ws1Pool.add(session);}@Overridepublic void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {Ws1Pool.remove(session);}@Overrideprotected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {if (StringUtils.equals("ping", message.getPayload())) {// 心跳消息session.sendMessage(new TextMessage("pong"));return;}logger.info("receive Msg :" + message.getPayload());TextMessage msg = new TextMessage(message.getPayload());// 回发信息 给 js前端session.sendMessage(msg);}}

消息对象VO

@Data
@NoArgsConstructor
@AllArgsConstructor
public class WsMsgVo {private String text;private Long userId;
}

将请求参数字符串转换成map 工具类

public class ParamUtil {/** 将请求参数字符串转换成map */public static Map<String, String> parser(String queryString) {Map<String, String> map = new HashMap<String, String>();if (StringUtils.isNotBlank(queryString)) {String[] params = queryString.split("&");for (String p : params) {String[] strs = p.split("=");if (strs.length == 2) {map.put(strs[0], strs[1]);}}}return map;}
}

2、创建websocket链接池

存储所有在线用户链接，并实现发送消息和广播消息的功能；使用异步线给前端发送消息

/*** websocket链接池*/
public class Ws1Pool {private static final Logger logger = LoggerFactory.getLogger(Ws1Pool.class);private static final Map<String, WebSocketSession> pool = new ConcurrentHashMap<>();private static final Map<Long, List<String>> userMap = new ConcurrentHashMap<>();private static final ExecutorService threadPool = Executors.newFixedThreadPool(10);public static void add(WebSocketSession inbound) {pool.put(inbound.getId(), inbound);Map<String, String> map = ParamUtil.parser(inbound.getUri().getQuery());Long userId = Long.valueOf(map.get("userId"));logger.info("add userId:{}", userId);List<String> lstInBound = userMap.computeIfAbsent(userId, k -> new ArrayList<>());lstInBound.add(inbound.getId());logger.info("add connetion {},total size {}", inbound.getId(), pool.size());}public static void remove(WebSocketSession socket) {String sessionId = socket.getId();List<String> lstInBound = null;Map<String, String> map = ParamUtil.parser(socket.getUri().getQuery());Long userId = Long.valueOf(map.get("userId"));logger.info("remove userId:{}", userId);if (StringUtils.isNotBlank(sessionId)) {lstInBound = userMap.get(userId);if (lstInBound != null) {lstInBound.remove(sessionId);if (lstInBound.isEmpty()) {userMap.remove(userId);}}}pool.remove(sessionId);logger.info("remove connetion {},total size {}", sessionId, pool.size());}/** 推送信息 */public static void broadcast(WsMsgVo vo) {Long userId = vo.getUserId();List<String> lstInBoundId;if (userId == null || userId == 0L) {// 发送给所有人lstInBoundId = userMap.values().stream().flatMap(List::stream).collect(Collectors.toList());} else {lstInBoundId = userMap.get(userId);}if (lstInBoundId == null || lstInBoundId.isEmpty()) {return;}threadPool.execute(() -> {try {for (String id : lstInBoundId) {// 发送给指定用户WebSocketSession connection = pool.get(id);if (connection != null) {synchronized (connection) {TextMessage msg = new TextMessage(vo.getText());connection.sendMessage(msg);}}}} catch (Exception e) {logger.error("broadcast error: userId:{}", userId, e);}});}}

3、创建Websocket配置类，并注册地址/ws/demo1，将地址和处理类绑定

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;/*** Websocket配置*/
@Configuration
@EnableWebSocket
public class Ws1Config implements WebSocketConfigurer {@Overridepublic void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {registry.addHandler(ws1Handler(), "/ws/demo1").setAllowedOrigins("*");}@Beanpublic Ws1Handler ws1Handler() {return new Ws1Handler();}}

前端代码

在这段代码中，我们首先创建了一个新的WebSocket对象，并传入了WebSocket服务的URL。然后，我们为这个WebSocket连接添加了四个事件监听器：

  1. onopen：当WebSocket连接打开时触发。
  2. onmessage：当从服务器接收到消息时触发。
  3. onerror：当WebSocket发生错误时触发。
  4. onclose：当WebSocket连接关闭时触发。

你可以根据自己的需求，在这些事件监听器中添加相应的逻辑。例如，在onopen事件监听器中发送一个消息给服务器，或者在onmessage事件监听器中处理从服务器接收到的消息。

<!DOCTYPE html>
<html xmlns:th="http://www.thymeleaf.org"xmlns:layout="http://www.ultraq.net.nz/thymeleaf/layout">
<head>
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>websocket demo1</title>
</head>
<body>
<div><div id="content"><div>信息面板</div></div><input type="text" id="input" placeholder="请输入内容" /><button id="send" onclick="doSend()">发送</button></div>
<script type="text/javascript">// 创建一个新的WebSocket并连接到指定的URLvar socket = new WebSocket('ws://localhost:8080/ws/demo1?userId=001');// 当WebSocket连接打开时触发socket.onopen = function(event) {console.log('Connection opened');// 可以选择在这里发送一些数据到服务器socket.send('Hello, server!');};// 当从服务器接收到数据时触发socket.onmessage = function(event) {if (event.data == null || event.data == '' || "pong" == event.data) {//心跳消息console.log("Info: 心跳消息");} else {console.log('Message from server ', event.data);var div_msg = document.createElement('div');div_msg.textContent = event.data;document.getElementById('content').appendChild(div_msg)}};// 当WebSocket连接关闭时触发socket.onclose = function(event) {console.log('Connection closed');};// 当WebSocket连接发生错误时触发socket.onerror = function(error) {console.error('WebSocket Error:', error);};function doSend(){var input_dom = document.getElementById('input')var value = input_dom.value;input_dom.value=''input_dom.focus()socket.send(value);}</script></body>
</html>

三、示例2：使用注解@ServerEndpoint

后端代码

1、创建Websocket处理类Ws3Handler，并使用注解@ServerEndpoint声明

import jakarta.websocket.*;
import jakarta.websocket.server.ServerEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.thymeleaf.util.StringUtils;import java.io.IOException;/*** websocket事件处理* <p>* 链接：/ws/demo3?userId=xxxx*/
@Component
@ServerEndpoint("/ws/demo3")
public class Ws3Handler {private static final Logger logger = LoggerFactory.getLogger(Ws3Handler.class);@OnOpenpublic void onOpen(Session session) {Ws3Pool.add(session);}@OnClosepublic void OnClose(Session session) {Ws3Pool.remove(session);}@OnMessagepublic void onMessage(Session session, String message) throws IOException {if (StringUtils.equals("ping", message)) {// 心跳消息session.getBasicRemote().sendText("pong");return;}logger.info("receive Msg :" + message);
//        session.getBasicRemote().sendText(message);Ws3Pool.broadcast(new WsMsgVo(message, 0L));}/*** 错误时调用*/@OnErrorpublic void onError(Session session, Throwable throwable) {throwable.printStackTrace();}}

2、创建websocket链接池

存储所有在线用户链接，并实现发送消息和广播消息的功能；使用异步线给前端发送消息

/*** websocket链接池*/
public class Ws3Pool {private static final Logger logger = LoggerFactory.getLogger(Ws3Pool.class);private static Map<String, Session> pool = new ConcurrentHashMap<>();private static final Map<Long, List<String>> userMap = new ConcurrentHashMap<>();private static final ExecutorService threadPool = Executors.newFixedThreadPool(10);public static void add(Session session) {pool.put(session.getId(), session);Map<String, String> map = ParamUtil.parser(session.getQueryString());Long userId = Long.valueOf(map.get("userId"));logger.info("add userId:{}", userId);List<String> lstInBound = userMap.computeIfAbsent(userId, k -> new ArrayList<>());lstInBound.add(session.getId());logger.info("add connetion {},total size {}", session.getId(), pool.size());}public static void remove(Session session) {String sessionId = session.getId();List<String> lstInBound = null;Map<String, String> map = ParamUtil.parser(session.getQueryString());Long userId = Long.valueOf(map.get("userId"));logger.info("remove userId:{}", userId);if (StringUtils.isNotBlank(sessionId)) {lstInBound = userMap.get(userId);if (lstInBound != null) {lstInBound.remove(sessionId);if (lstInBound.isEmpty()) {userMap.remove(userId);}}}pool.remove(sessionId);logger.info("remove connetion {},total size {}", sessionId, pool.size());}/** 推送信息 */public static void broadcast(WsMsgVo vo) {Long userId = vo.getUserId();List<String> lstInBoundId;if (userId == null || userId == 0L) {// 发送给所有人lstInBoundId = userMap.values().stream().flatMap(List::stream).collect(Collectors.toList());} else {lstInBoundId = userMap.get(userId);}if (lstInBoundId == null || lstInBoundId.isEmpty()) {return;}threadPool.execute(() -> {try {for (String id : lstInBoundId) {// 发送给指定用户Session session = pool.get(id);if (session != null) {synchronized (session) {session.getBasicRemote().sendText(vo.getText());}}}} catch (Exception e) {logger.error("broadcast error: userId:{}", userId, e);}});}}

3、创建Websocket配置类，并注册 ServerEndpointExporter

ServerEndpointExporter 是 Spring Boot 中的一个重要组件，用于导出 WebSocket 服务器端点配置。在 Spring 应用程序中，特别是当你使用 Spring Boot 时，ServerEndpointExporter 能够自动注册使用 @ServerEndpoint 注解声明的 WebSocket 端点。

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
@Configuration
@EnableWebSocket
public class Ws3Config {/*** ServerEndpointExporter 作用* 这个Bean会自动注册使用@ServerEndpoint注解声明的websocket endpoint*/@Beanpublic ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter();}
}

在这个配置类中，我们定义了一个 serverEndpointExporter 方法，它返回一个 ServerEndpointExporter 的实例。这样，Spring 容器就会管理这个 bean，并自动注册所有使用 @ServerEndpoint 注解声明的 WebSocket 端点。

前端代码

（可以和上面的例子一致）

四、前端代码封装

为了更方便的使用websocket，以及确保在后端服务器重启后，前端websocket能够自动重连，我们可以增加心跳机制

1、创建 ws.handler.js

var WsHander = {};
WsHander.socket = null;
WsHander.connect = (function (host) {WsHander.host = host;if ("WebSocket" in window) {WsHander.socket = new WebSocket(host);} else if ("MozWebSocket" in window) {WsHander.socket = new MozWebSocket(host);} else {console.log("Error: WebSocket is not supported by this browser.");return;}WsHander.socket.onopen = function () {console.log("Info: websocket已启动.");// 心跳检测重置heartCheck.reset().start(WsHander.socket);};WsHander.socket.onclose = function () {console.log("Info: websocket已关闭.");WsHander.reconnect();};WsHander.socket.onmessage = function (message) {heartCheck.reset().start(WsHander.socket);if (message.data == null || message.data === '' || "pong" === message.data) {//心跳消息console.log("Info: 心跳消息");} else {// 收到 Websocket消息，执行业务操作if (doOnWsMessage){doOnWsMessage(message.data);}}};
});
WsHander.reconnect = function (){WsHander.connect(WsHander.host);
}
WsHander.initialize = function (userId, uri) {WsHander.currUserId = userId;if (WsHander.currUserId) {if (window.location.protocol === "http:") {WsHander.connect("ws://" + window.location.host + uri+"?userId=" + WsHander.currUserId);} else {WsHander.connect("wss://" + window.location.host + uri+"?userId=" + WsHander.currUserId);}}
};
WsHander.sendMessage = (function (message) {if (message !== "") {WsHander.socket.send(message);}
});//心跳检测
var heartCheck = {timeout: 5000,// 5秒timeoutObj: null,reset: function () {clearTimeout(this.timeoutObj);return this;},start: function (ws) {var self = this;this.timeoutObj = setTimeout(function () {// 这里发送一个心跳，后端收到后，返回一个心跳消息，// onmessage拿到返回的心跳就说明连接正常ws.send("ping");}, this.timeout);}
}

2、在html页面中使用

<!DOCTYPE html>
<html xmlns:th="http://www.thymeleaf.org"xmlns:layout="http://www.ultraq.net.nz/thymeleaf/layout">
<head>
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>websocket demo1</title>
</head>
<body>
<div><div id="content"><div>信息面板</div></div><input type="text" id="input" placeholder="请输入内容" /><button id="send" onclick="doSend()">发送</button></div>
<script src="../static/ws.handler.js" ></script>
<script type="text/javascript">WsHander.initialize("001", "/ws/demo3")// 当从服务器接收到数据时触发function doOnWsMessage(messageData){console.log('Message from server ', messageData);var div_msg = document.createElement('div');div_msg.textContent = messageData;document.getElementById('content').appendChild(div_msg)}function doSend(){var input_dom = document.getElementById('input')var value = input_dom.value;input_dom.value=''input_dom.focus()WsHander.sendMessage(value)}</script></body>
</html>

相关文章：

  • 北京网站建设多少钱？
  • 辽宁网页制作哪家好_网站建设
  • 高端品牌网站建设_汉中网站制作
  • 第二章 方法与方法重载
  • 在HFSS中对曲线等结构进行分割（Split）
  • ubuntu 安装两个nginx实例时的坑，非默认nginx实例配置修改总也不生效的问题
  • HTML及CSS面试题4
  • 66_1JSON【浏览器中处理JSON、Java中处理JSON(FastJSON、Jackson)】、Java中的bean
  • XML外部实体注入
  • 使用docxtemplater-image-module-free时支持动态获取图片大小
  • SpringBoot：将单体项目拆分成微服务项目
  • 【PGCCC】pg_bestmatch.rs：使用 BM25 提升您的 PostgreSQL 文本查询#PCA
  • Windows下添加开机启动项
  • Vue.js 中的指令（Vue自定义指令）
  • 在小程序添加公司官网访问
  • 使用 Vue 2 搭建后台管理系统
  • 学习计算机网络（五）——ICMP协议
  • 仕考网：考外省公务员可以调回本地吗?
  • 「前端」从UglifyJSPlugin强制开启css压缩探究webpack插件运行机制
  • 【comparator, comparable】小总结
  • Bytom交易说明（账户管理模式）
  • CSS 提示工具(Tooltip)
  • HTTP传输编码增加了传输量，只为解决这一个问题 | 实用 HTTP
  • JS创建对象模式及其对象原型链探究（一）：Object模式
  • JS基础篇--通过JS生成由字母与数字组合的随机字符串
  • js面向对象
  • k个最大的数及变种小结
  • LeetCode541. Reverse String II -- 按步长反转字符串
  • PHP CLI应用的调试原理
  • php的插入排序，通过双层for循环
  • Python socket服务器端、客户端传送信息
  • 初识 beanstalkd
  • 分享自己折腾多时的一套 vue 组件 --we-vue
  • 构建二叉树进行数值数组的去重及优化
  • 开源地图数据可视化库——mapnik
  • 理清楚Vue的结构
  • 携程小程序初体验
  • 云大使推广中的常见热门问题
  • ［地铁译］使用SSD缓存应用数据——Moneta项目: 低成本优化的下一代EVCache ...
  • postgresql行列转换函数
  • ​​​​​​​GitLab 之 GitLab-Runner 安装，配置与问题汇总
  • # Spring Cloud Alibaba Nacos_配置中心与服务发现（四）
  • # Swust 12th acm 邀请赛# [ K ] 三角形判定 [题解]
  • # 学号 2017-2018-20172309 《程序设计与数据结构》实验三报告
  • (22)C#传智:复习,多态虚方法抽象类接口,静态类,String与StringBuilder,集合泛型List与Dictionary,文件类,结构与类的区别
  • （附源码）python旅游推荐系统 毕业设计 250623
  • （附源码）springboot太原学院贫困生申请管理系统 毕业设计 101517
  • （附源码）ssm基于jsp高校选课系统 毕业设计 291627
  • （附源码）基于SpringBoot和Vue的厨到家服务平台的设计与实现 毕业设计 063133
  • （原创）boost.property_tree解析xml的帮助类以及中文解析问题的解决
  • (转)h264中avc和flv数据的解析
  • （最简单，详细，直接上手）uniapp/vue中英文多语言切换
  • *上位机的定义
  • .bat批处理（八）：各种形式的变量%0、%i、%%i、var、%var%、!var!的含义和区别
  • .NET C# 使用 iText 生成PDF
  • .NET Standard、.NET Framework 、.NET Core三者的关系与区别？
  • .net web项目 调用webService
  • .net 后台导出excel ,word