How to use Spring Boot+Vue to implement Socket notification push

Spring Boot side

The first step is to introduce dependencies

First we need to introduce the dependencies required by WebSocket and the dependencies for processing the output format

<!--格式转换-->
<!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
<dependency>
   <groupId>com.alibaba</groupId>
   <artifactId>fastjson</artifactId>
   <version>1.2.73</version>
</dependency>
<!--WebSocket依赖-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

The second step , Create a WebSocket configuration class

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
 
/**
 * @author: tjp
 * @create: 2023-04-03 09:58
 * @Description: WebSocket配置
 */
@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
 
}

The third step is to create a WebSocket service

In this step, we use userId as an identifier to distinguish the corresponding users in the system. We can also perform other operations based on this in the future. Steps.

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.excel.util.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
 
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
 
/**
 * @author: tjp
 * @create: 2023-04-03 13:55
 * @Description: WebSocket服务
 */
 
@ServerEndpoint("/websocket/{userId}")
@Slf4j
@Component
public class WebSocketServer {
    /**
     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
     */
    private static int onlineCount = 0;
    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
     */
    private static ConcurrentHashMap<String, WebSocketServer> 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);
            //加入set中
        } else {
            webSocketMap.put(userId, this);
            //加入set中
            addOnlineCount();
            //在线数加1
        }
 
        log.info("用户连接:" + userId + ",当前在线人数为:" + getOnlineCount());
 
        try {
            HashMap<Object, Object> map = new HashMap<>();
            map.put("key", "连接成功");
            sendMessage(JSON.toJSONString(map));
        } 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.isNotBlank(message)) {
            try {
                //解析发送的报文
                JSONObject jsonObject = JSONObject.parseObject(message);
                //追加发送人(防止串改)
                jsonObject.put("fromUserId", this.userId);
                String fromUserId = jsonObject.getString("fromUserId");
                //传送给对应toUserId用户的websocket
                if (StringUtils.isNotBlank(fromUserId) && webSocketMap.containsKey(fromUserId)) {
                    webSocketMap.get(fromUserId).sendMessage(jsonObject.toJSONString());
                    //自定义-业务处理
 
//                    DeviceLocalThread.paramData.put(jsonObject.getString("group"),jsonObject.toJSONString());
                } else {
                    log.error("请求的userId:" + fromUserId + "不在该服务器上");
                    //否则不在这个服务器上，发送到mysql或者redis
                }
            } 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 {
        //加入线程锁
        synchronized (session) {
            try {
                //同步发送信息
                this.session.getBasicRemote().sendText(message);
            } catch (IOException e) {
                log.error("服务器推送失败:" + e.getMessage());
            }
        }
    }
 
 
    /**
     * 发送自定义消息
     * */
    /**
     * 发送自定义消息
     *
     * @param message  发送的信息
     * @param toUserId 如果为null默认发送所有
     * @throws IOException
     */
    public static void sendInfo(String message, String toUserId) throws IOException {
        //如果userId为空，向所有群体发送
        if (StringUtils.isEmpty(toUserId)) {
            //向所有用户发送信息
            Iterator<String> itera = webSocketMap.keySet().iterator();
            while (itera.hasNext()) {
                String keys = itera.next();
                WebSocketServer item = webSocketMap.get(keys);
                item.sendMessage(message);
            }
        }
        //如果不为空，则发送指定用户信息
        else if (webSocketMap.containsKey(toUserId)) {
            WebSocketServer item = webSocketMap.get(toUserId);
            item.sendMessage(message);
        } else {
            log.error("请求的userId:" + toUserId + "不在该服务器上");
        }
    }
 
    public static synchronized int getOnlineCount() {
        return onlineCount;
    }
 
    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }
 
    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }
 
    public static synchronized ConcurrentHashMap<String, WebSocketServer> getWebSocketMap() {
        return WebSocketServer.webSocketMap;
    }
 
}

The fourth step is to create a Controller for sending test

Get the current number of people online

import com.......WebSocketServer;
 
@ApiOperation(value = "获取当前在线人数")
@GetMapping("/getOnlineCount")
public Integer getOnlineCount() {
    return WebSocketServer.getOnlineCount();
}

Push messages to front-end users through the interface

import com.......WebSocketServer;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
 
import java.io.IOException;
 
/**
 * @author: tjp
 * @create: 2023-04-03 13:57
 * @Description: 测试
 */
@RestController
@RequestMapping("/news")
public class NewsController {
    @GetMapping("/send")
    public String send() {
        try {
            WebSocketServer.sendInfo("这是websocket发送过来的消息！", "需要推送的用户的编号");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return "发送消息成功";
    }
 
}

Vue side

The first step is to create the connection tool class

Create the tool class websocket.js. The userId here is the userId used as the identifier

/**
 * @author: tjp
 * @create: 2023-04-03 11:22
 * @Description: Socket客户端
 */
export class WebSocketClient {
  constructor(userId) {
    this.userId = userId;
    this.websocket = null;
    this.timeout = 10000; // 心跳超时时间，单位ms
    this.timeoutObj = null; // 心跳定时器
    this.serverTimeoutObj = null; // 服务器超时定时器
    this.lockReconnect = false; // 避免重复连接
    this.timeoutnum = null; // 重连延迟定时器
  }
 
  // 初始化WebSocket连接
  initWebSocket() {
    let wsUrl = `ws://127.0.0.1:8080/websocket/${this.userId}`;
    this.websocket = new WebSocket(wsUrl);
    this.websocket.onopen = this.websocketonopen.bind(this);
    this.websocket.onerror = this.websocketonerror.bind(this);
    this.websocket.onmessage = this.setOnmessageMessage.bind(this);
    this.websocket.onclose = this.websocketclose.bind(this);
    // 监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
    window.onbeforeunload = this.websocketclose.bind(this);
  }
 
  // 启动心跳
  start() {
    console.log('start');
    // 清除延时器
    this.timeoutObj && clearTimeout(this.timeoutObj);
    this.serverTimeoutObj && clearTimeout(this.serverTimeoutObj);
    /*// 向服务器发送心跳消息
    let actions = { "test": "12345" };
    this.websocket && this.websocket.readyState == 1 && this.websocket.send(JSON.stringify(actions));
    // 启动心跳定时器
    this.timeoutObj = setTimeout(() => {
      this.start();
      // 定义一个延时器等待服务器响应，若超时，则关闭连接，重新请求server建立socket连接
      this.serverTimeoutObj = setTimeout(() => {
        this.websocket.close();
      }, this.timeout)
    }, this.timeout)*/
  }
 
  // 重置心跳
  reset() {
    // 清除时间
    clearTimeout(this.timeoutObj);
    clearTimeout(this.serverTimeoutObj);
    // 重启心跳
    this.start();
  }
 
  // 重新连接
  reconnect() {
    if (this.lockReconnect) return;
    this.lockReconnect = true;
    // 没连接上会一直重连，设置延迟避免请求过多
    this.timeoutnum && clearTimeout(this.timeoutnum);
    this.timeoutnum = setTimeout(() => {
      this.initWebSocket();
      this.lockReconnect = false;
    }, 5000)
  }
 
  // 处理收到的消息
  async setOnmessageMessage(event) {
    console.log(event.data, '获得消息');
    // 重置心跳
    // this.reset();
    // 自定义全局监听事件
    window.dispatchEvent(new CustomEvent('onmessageWS', {
      detail: {
        data: event.data
      }
    }))
    // //发现消息进入    开始处理前端触发逻辑
    // if (event.data === 'success' || event.data === 'heartBath') return
  }
 
  // WebSocket连接成功回调
  websocketonopen() {
    // 开启心跳
    this.start();
    console.log("WebSocket连接成功!!!" + new Date() + "----" + this.websocket.readyState);
    clearInterval(this.otimer);//停止
  }
 
  // WebSocket连接错误回调
  websocketonerror(e) {
    console.log("WebSocket连接发生错误" + e);
  }
 
  // WebSocket连接关闭回调
  websocketclose(e) {
    this.websocket.close();
    clearTimeout(this.timeoutObj);
    clearTimeout(this.serverTimeoutObj);
    console.log("websocketcloe关闭连接")
  }
 
  // 关闭WebSocket连接
  closeWebSocket() {
    this.websocket.close();
    console.log("closeWebSocket关闭连接")
  }
 
  // 监听窗口关闭事件
  onbeforeunload() {
    this.closeWebSocket();
  }
}

The second step is to create Connection

Establish a Socket connection in any page you want to establish a connection

For example, after the user clicks the login button

You can use the prototype here to create a connection object, and Start the connection

<script>
import Vue from "vue";
import {WebSocketClient} from "@/utils/websocket";
......
......
methods:{
 
 handleLogin() {
      this.$refs.loginForm.validate(valid => {
        if (valid) {
          this.loading = true
          this.$store.dispatch(&#39;user/login&#39;, this.loginForm).then(() => {
            this.$router.push({path: this.redirect || &#39;/&#39;})
            this.loading = false
            /*-----------在此处放入原型中------------*/
            Vue.prototype.$WebSocketClientInstance = new WebSocketClient(&#39;t&#39;);
            Vue.prototype.$WebSocketClientInstance.initWebSocket()
            /*-----------------end------------*/
          }).catch(() => {
            this.loading = false
          })
        } else {
          this.$message({message: &#39;请填写正确格式的用户名或密码&#39;, type: &#39;error&#39;})
          return false
        }
      })
    }    
}
 
.....
.....
</script>

The third step is to listen to the messages sent by the server

On the page you want to monitor, use the listener to listen

<script>
....
....
mounted() {
    // 添加socket通知监听
    window.addEventListener(&#39;onmessageWS&#39;, this.getSocketData)
},
methods: {
    // 收到消息处理
    getSocketData(res) {
      console.log(res.detail)
      console.log("llll")
    },
}
 
....
....
</script>

At this time, you can send it through the back-end interface

Do a test

Four steps, close the connection

Get a button

<template>
  <div>
    <button @click="closeConnect">关闭连接</button>
  </div>
</template>
 
<script>
import {WebSocketClient} from "@/utils/websocket";
import Vue from "vue";
 
export default {
  methods: {
    closeConnect() {
      console.dir(Vue.prototype)
      Vue.prototype.$WebSocketClientInstance.closeWebSocket();
    },
  }
}
</script>

The above is the detailed content of How to use Spring Boot+Vue to implement Socket notification push. For more information, please follow other related articles on the PHP Chinese website!