spring boot 网络协议 websocket 实现后端主动前端推送数据、及时通讯(vue3 + springboot)

简介

WebSocket 是一种全双工通信协议，用于在 Web 浏览器和服务器之间建立持久的连接。

WebSocket 协议由 IETF 定为标准，WebSocket API 由 W3C 定为标准。一旦 Web 客户端与服务器建立连接，之后的全部数据通信都通过这个连接进行。可以互相发送 JSON、XML、HTML 或图片等任意格式的数据。

WebSocket 与 HTTP 协议的异同:

相同点：

都是基于 TCP 的应用层协议。都使用 Request/Response 模型进行连接的建立。可以在网络中传输数据。

不同点：

WebSocket 使用 HTTP 来建立连接，但定义了一系列新的 header 域，这些域在 HTTP 中并不会使用。WebSocket 支持持久连接，而 HTTP 协议不支持持久连接。

WebSocket 优点: 高效性： 允许在一条 WebSocket 连接上同时并发多个请求，避免了传统 HTTP 请求的多个 TCP 连接。 WebSocket 的长连接特性提高了效率，避免了 TCP 慢启动和连接握手的开销。 节省带宽：HTTP 协议的头部较大，且请求中的大部分头部内容是重复的。WebSocket 复用长连接，避免了这一问题。 服务器推送：WebSocket 支持服务器主动推送消息，实现实时消息通知。

WebSocket 缺点: 长期维护成本：服务器需要维护长连接，成本较高。 浏览器兼容性：不同浏览器对 WebSocket 的支持程度不一致。 受网络限制：WebSocket 是长连接，受网络限制较大，需要处理好重连。

WebSocket 应用场景:

实时通信领域：社交聊天弹幕多玩家游戏协同编辑股票基金实时报价体育实况更新视频会议/聊天基于位置的应用在线教育智能家居等需要高实时性的场景。

一、后端代码

1、安装核心jar包: spring-boot-starter-websocket

org.springframework.boot

spring-boot-starter-websocket

cn.hutool

hutool-all

5.1.0

com.alibaba

fastjson

2.0.22

2. 来一个配置类注入

package com.codeSE.config;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.web.socket.server.standard.ServerEndpointExporter;

@Configuration

public class WebSocketConfig2 {

@Bean

public ServerEndpointExporter serverEndpointExporter(){

return new ServerEndpointExporter();

}

}

3. 写个基础webSocket服务

import cn.hutool.log.Log;

import cn.hutool.log.LogFactory;

import com.alibaba.fastjson.JSON;

import com.alibaba.fastjson.JSONObject;

//import org.apache.commons.lang3.StringUtils;

import org.springframework.stereotype.Component;

import org.springframework.util.StringUtils;

import javax.websocket.*;

import javax.websocket.server.PathParam;

import javax.websocket.server.ServerEndpoint;

import java.io.IOException;

import java.util.concurrent.ConcurrentHashMap;

/**

* @ClassName: 开启WebSocket支持

*/

@ServerEndpoint("/dev-api/websocket/{userId}")

@Component

public class WebSocketServer {

static Log log = LogFactory.get(WebSocketServer.class);

/**

* 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。

*/

private static int onlineCount = 0;

/**

* concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。

*/

private static ConcurrentHashMap webSocketMap = new ConcurrentHashMap<>();

/**

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

*/

private Session session;

/**

* 接收userId

*/

private String userId = "";

/**

* 连接建立成功调用的方法

*/

@OnOpen

public void onOpen(Session session, @PathParam("userId") String userId) {

this.session = session;

this.userId = userId;

if (webSocketMap.containsKey(userId)) {

webSocketMap.remove(userId);

webSocketMap.put(userId, this);

//加入set中

} else {

webSocketMap.put(userId, this);

//加入set中

addOnlineCount();

//在线数加1

}

log.info("用户连接:" + userId + ",当前在线人数为:" + getOnlineCount());

try {

sendMessage("连接成功");

} catch (IOException e) {

log.error("用户:" + userId + ",网络异常!!!!!!");

}

}

/**

* 连接关闭调用的方法

*/

@OnClose

public void onClose() {

if (webSocketMap.containsKey(userId)) {

webSocketMap.remove(userId);

//从set中删除

subOnlineCount();

}

log.info("用户退出:" + userId + ",当前在线人数为:" + getOnlineCount());

}

/**

* 收到客户端消息后调用的方法

*

* @param message 客户端发送过来的消息

*/

@OnMessage

public void onMessage(String message, Session session) {

log.info("用户消息:" + userId + ",报文:" + message);

//可以群发消息

//消息保存到数据库、redis

if (! StringUtils.isEmpty(message)) {

try {

//解析发送的报文

JSONObject jsonObject = JSON.parseObject(message);

} catch (Exception e) {

e.printStackTrace();

}

}

}

/**

* @param session

* @param error

*/

@OnError

public 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 sendAllMessage(String message) throws IOException {

ConcurrentHashMap.KeySetView userIds = webSocketMap.keySet();

for (String userId : userIds) {

WebSocketServer webSocketServer = webSocketMap.get(userId);

webSocketServer.session.getBasicRemote().sendText(message);

System.out.println("webSocket实现服务器主动推送成功userIds====" + userIds);

}

}

/**

* 发送自定义消息

*/

public static void sendInfo(String message, @PathParam("userId") String userId) throws IOException {

log.info("发送消息到:" + userId + "，报文:" + message);

if (!StringUtils.isEmpty(message) && 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--;

}

}

4. 写一个测试类，定时向客户端推送数据或者可以发起请求推送

package com.codeSE.controller;

import com.alibaba.fastjson2.JSONObject;

import com.codeSE.config.WebSocketServer;

import org.springframework.scheduling.annotation.Scheduled;

import org.springframework.web.bind.annotation.PostMapping;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;

import java.time.format.DateTimeFormatter;

import java.util.HashMap;

import java.util.Map;

@RestController

@RequestMapping("/money")

public class Test {

//设置定时十秒一次

@Scheduled(cron = "0/10 * * * * ?")

@PostMapping("/send")

public String sendMessage() throws Exception {

Map map = new HashMap<>();

// 获取当前日期和时间

LocalDateTime nowDateTime = LocalDateTime.now();

DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

System.out.println(dateTimeFormatter.format(nowDateTime));

map.put("server_time",dateTimeFormatter.format(nowDateTime));

map.put("server_code","200");

map.put("server_message","这是服务器推送到客户端的消息哦！！");

JSONObject jsonObject = new JSONObject(map);

WebSocketServer.sendAllMessage(jsonObject.toString());

return jsonObject.toString();

}

}

5. 启动springboot

package com.codeSE;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.boot.web.servlet.ServletComponentScan;

import org.springframework.scheduling.annotation.EnableScheduling;

@EnableScheduling //定时任务

@ServletComponentScan //webSocket

@SpringBootApplication

public class WebSocketAppMain {

public static void main(String[] args) {

SpringApplication.run(WebSocketAppMain.class);

}

}

使用网上的测试工具试一下：http://coolaf.com/tool/chattest 或者http://www.jsons.cn/websocket/ 效果如下：

二、前端代码

使用vue3和原生websocket

1、简单写一个websocket的公共类

需求：commentUtil/WebsocketTool.js

//需求：在JavaScript中实现WebSocket连接失败后3分钟内尝试重连3次的功能，你可以设置一个重连策略，

// 包括重连的间隔时间、尝试次数以及总时间限制。

/**

* @param {string} url Url to connect

* @param {number} maxReconnectAttempts Maximum number of times

* @param {number} reconnect Timeout

* @param {number} reconnectTimeout Timeout

*

*/

class WebSocketReconnect {

constructor(url, maxReconnectAttempts = 3, reconnectInterval = 20000, maxReconnectTime = 180000) {

this.url = url

this.maxReconnectAttempts = maxReconnectAttempts

this.reconnectInterval = reconnectInterval

this.maxReconnectTime = maxReconnectTime

this.reconnectCount = 0

this.reconnectTimeout = null

this.startTime = null

this.socket = null

this.connect()

}

//连接操作

connect() {

console.log('connecting...')

this.socket = new WebSocket(this.url)

//连接成功建立的回调方法

this.socket.onopen = () => {

console.log('WebSocket Connection Opened!')

this.clearReconnectTimeout()

this.reconnectCount = 0

}

//连接关闭的回调方法

this.socket.onclose = (event) => {

console.log('WebSocket Connection Closed:', event)

this.handleClose()

}

//连接发生错误的回调方法

this.socket.onerror = (error) => {

console.error('WebSocket Connection Error:', error)

this.handleClose() //重连

}

}

//断线重连操作

handleClose() {

if (this.reconnectCount < this.maxReconnectAttempts && (this.startTime === null ||

Date.now() - this.startTime < this.maxReconnectTime)) {

this.reconnectCount++

console.log(`正在尝试重连 (${this.reconnectCount}/${this.maxReconnectAttempts})次...`)

this.reconnectTimeout = setTimeout(() => {

this.connect()

}, this.reconnectInterval)

if (this.startTime === null) {

this.startTime = Date.now()

}

} else {

console.log('超过最大重连次数或重连时间超时，已放弃连接！Max reconnect attempts reached or exceeded max reconnect time. Giving up.')

this.reconnectCount = 0 // 重置连接次数0

this.startTime = null // 重置开始时间

}

}

//清除重连定时器

clearReconnectTimeout() {

if (this.reconnectTimeout) {

clearTimeout(this.reconnectTimeout)

this.reconnectTimeout = null

}

}

//关闭连接

close() {

if (this.socket && this.socket.readyState === WebSocket.OPEN) {

this.socket.close()

}

this.clearReconnectTimeout()

this.reconnectCount = 0

this.startTime = null

}

}

// WebSocketReconnect 类封装了WebSocket的连接、重连逻辑。

// maxReconnectAttempts 是最大重连尝试次数。

// reconnectInterval 是每次重连尝试之间的间隔时间。

// maxReconnectTime 是总的重连时间限制，超过这个时间后不再尝试重连。

// reconnectCount 用于记录已经尝试的重连次数。

// startTime 用于记录开始重连的时间。

// connect 方法用于建立WebSocket连接，并设置相应的事件监听器。

// handleClose 方法在WebSocket连接关闭或发生错误时被调用，根据条件决定是否尝试重连。

// clearReconnectTimeout 方法用于清除之前设置的重连定时器。

// close 方法用于关闭WebSocket连接，并清除重连相关的状态。

// 使用示例

// const webSocketReconnect = new WebSocketReconnect('ws://your-websocket-url')

// 当不再需要WebSocket连接时，可以调用close方法

// webSocketReconnect.close();

export default WebSocketReconnect

2、在任意Vue页面

效果：

好文推荐