springBoot与Vue共同搭建webSocket环境

欢迎使用Markdown编辑器

你好！这片文章将教会你从后端springCloud到前端VueEleementAdmin如何搭建Websocket

前端

1. 创建websocket的配置文件在utils文件夹下websocket.js

// my-js-file.js
import { Notification } from 'element-ui'
// 暴露自定义websocket对象
export const socket = {
  // 后台请求路径
  url: '',
  websocketCount: -1,
  // websocket对象
  websocket: null,
  // websocket状态
  websocketState: false,
  // 重新连接次数
  reconnectNum: 0,
  // 重连锁状态，保证重连按顺序执行
  lockReconnect: false,
  // 定时器信息
  timeout: null,
  clientTimeout: null,
  serverTimeout: null,
  // 初始化方法，根据url创建websocket对象封装基本连接方法，并重置心跳检测
  initWebSocket(newUrl) {
    socket.url = newUrl
    socket.websocket = new WebSocket(socket.url)
    socket.websocket.onopen = socket.websocketOnOpen
    socket.websocket.onerror = socket.websocketOnError
    socket.websocket.onmessage = socket.webonmessage
    socket.websocket.onclose = socket.websocketOnClose
    this.resetHeartbeat()
  },
  reconnect() {
    // 判断连接状态
    console.log('判断连接状态')
    if (socket.lockReconnect) return
    socket.reconnectNum += 1
    // 重新连接三次还未成功调用连接关闭方法
    if (socket.reconnectNum === 3) {
      socket.reconnectNum = 0
      socket.websocket.onclose()
      return
    }
    // 等待本次重连完成后再进行下一次
    socket.lockReconnect = true
    // 5s后进行重新连接
    socket.timeout = setTimeout(() => {
      socket.initWebSocket(socket.url)
      socket.lockReconnect = false
    }, 5000)
  },
  // 重置心跳检测
  resetHeartbeat() {
    socket.heartbeat()
  },
  // 心跳检测
  heartbeat() {
    socket.clientTimeout = setTimeout(() => {
      if (socket.websocket) {
        // 向后台发送消息进行心跳检测
        socket.websocket.send(JSON.stringify({ type: 'heartbeat' }))
        socket.websocketState = false
        // 一分钟内服务器不响应则关闭连接
        socket.serverTimeout = setTimeout(() => {
          if (!socket.websocketState) {
            socket.websocket.onclose()
            console.log('一分钟内服务器不响应则关闭连接')
          } else {
            this.resetHeartbeat()
          }
        }, 60 * 1000)
      }
    }, 3 * 1000)
  },
  // 发送消息
  sendMsg(message) {
    socket.websocket.send(message)
  },
  websocketOnOpen(event) {
    // 连接开启后向后台发送消息进行一次心跳检测
    socket.sendMsg(JSON.stringify({ type: 'heartbeat' }))
  },
  // 初始化websocket对象
  // window.location.host获取ip和端口，
  // process.env.VUE_APP_WEBSOCKET_BASE_API获取请求前缀
  // 绑定接收消息方法
  webonmessage(event) {
    // 初始化界面时，主动向后台发送一次消息，获取数据
    this.websocketCount += 1
    if (this.websocketCount === 0) {
      const queryCondition = {
        type: 'message'
      }
      socket.sendMsg(JSON.stringify(queryCondition))
      console.log('初始化界面时，主动向后台发送一次消息，获取数据')
    }
    const info = JSON.parse(event.data)
    switch (info.type) {
      case 'heartbeat':
        socket.websocketState = true
        console.log(JSON.stringify(info))
        break
      case 'message':
        if (info.message === '物资管理模块-导入成功！') {
          Notification({
            title: '消息',
            message: '物资管理模块-导入成功,请手动刷新页面查看数据！',
            type: 'success',
            duration: 0,
            position: 'top-righ'
          })
        } else {
          Notification({
            title: '消息',
            message: '错了：' + info.message,
            type: 'error',
            duration: 0,
            position: 'top-righ'
          })
        }

        break
      case 'error':
        console.log('websocket：error')
        break
    }
  },
  websocketOnError(error) {
    console.log(error)
    console.log('websocket报错了' + error)
    socket.reconnect()
  },
  websocketOnClose() {
    console.log('websocke他关闭了')
    socket.websocket.close()
  }
}

2. 在main.js中引入配置文件，使websocket全局都能使用

import { socket } from './utils/websocket'
Vue.prototype.$socket = socket

3. 设置登陆时开启websocket连接，

this. $soc k e t . ini t W e b S oc k e t (‘ w s :$ {process.env.VUE_APP_WEBSOCKET_BASE_API}/websocket/` + this.loginForm.username
) 这句是开启websocket连接的。

// 登录方法
handleLogin() {
  this.$refs.loginForm.validate(valid => {
    if (valid) {
      this.loading = true
      this.$store.dispatch('user/login', this.loginForm)
        .then(() => {
          this.$router.push({ path: this.redirect || '/', query: this.otherQuery })
          this.$store.dispatch('user/addInfomation', this.infoMation)
          this.$socket.initWebSocket(
            `ws:${process.env.VUE_APP_WEBSOCKET_BASE_API}/websocket/` + this.loginForm.username
          )
          this.loading = false
        })
        .catch(() => {
          this.loading = false
        })
    } else {
      console.log('error submit!!')
      return false
    }
  })
},

4. 设置路由跳转时判断websocket连接是否断开，断开重连（在router的router.beforeEach函数中）

 if (socket.websocketState === false) {
      socket.initWebSocket(
        `ws:${process.env.VUE_APP_WEBSOCKET_BASE_API}/websocket/` + localStorage.getItem('USERNAME')
      )
    }

5. 设置退出时关闭websocket连接

this.$socket.websocketOnClose()这句是关闭websocket连接的

 logout() {
      await this.$store.dispatch('user/logout')
      // 重点
      this.$socket.websocketOnClose()
      this.$router.push(`/login?redirect=${this.$route.fullPath}`)
    }

重点和需要配置的地方都在websocket.js里比如接收消息方法webonmessage

后端

1.引依赖

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

2.写配置

package com.szc.material.analysisService.confg.WebSocket;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

@Configuration
public class WebSocketConfig   {
    /**
     * 	注入ServerEndpointExporter，
     * 	这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }

}

3.创建Websocket服务类（依据自己的业务去改@OnMessage方法）

package com.szc.material.analysisService.confg.WebSocket;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import javax.security.auth.message.MessageInfo;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @author zhj
 * @ServerEndpoint：将目前的类定义成一个websocket服务器端，注解的值将被用于监听用户连接的终端访问URL地址，客户端可以通过这个URL来连接到WebSocket服务器端
 * @OnOpen：当WebSocket建立连接成功后会触发这个注解修饰的方法。
 * @OnClose：当WebSocket建立的连接断开后会触发这个注解修饰的方法。
 * @OnMessage：当客户端发送消息到服务端时，会触发这个注解修改的方法。
 * @OnError：当WebSocket建立连接时出现异常会触发这个注解修饰的方法。
 * ————————————————
 * 版权声明：本文为CSDN博主「人人都在发奋」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
 * 原文链接：https://blog.csdn.net/qq991658923/article/details/127022522
 */
@Component
@Slf4j
@ServerEndpoint("/websocket/{userId}")
public class MyWebSocketHandler extends TextWebSocketHandler {

    /**
     * 线程安全的无序的集合
     */
    private static final CopyOnWriteArraySet<Session> SESSIONS = new CopyOnWriteArraySet<>();

    /**
     * 存储在线连接数
     */
    private static final Map<String, Session> SESSION_POOL = new HashMap<>();

    @OnOpen
    public void onOpen(Session session, @PathParam(value = "userId") String userId) {
        try {

            SESSIONS.add(session);
            SESSION_POOL.put(userId, session);
            log.info("【WebSocket消息】有新的连接，总数为：" + SESSIONS.size());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @OnClose
    public void onClose(Session session,@PathParam(value = "userId") String userId) {
        try {

            SESSIONS.remove(session);
            SESSION_POOL.remove(userId);
            log.info("【WebSocket消息】连接断开，总数为：" + SESSION_POOL.size());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @OnMessage
    public void onMessage(String message, @PathParam(value = "userId") String userId) {
        JSONObject jsonObject = JSON.parseObject(message);
        if("heartbeat".equals(jsonObject.get("type").toString())){
            Map<String,String> messageInfor=new HashMap<>();
            messageInfor.put("type","heartbeat");
            messageInfor.put("message","我收到了你的心跳");
            sendOneMessage( userId,messageInfor);
            log.info("【WebSocket消息】收到客户端消息：" + message);
        }

    }

    /**
     * 此为广播消息
     *
     * @param message 消息
     */
    public void sendAllMessage(String message) {
        log.info("【WebSocket消息】广播消息：" + message);
        for (Session session : SESSIONS) {
            try {
                if (session.isOpen()) {
                    session.getAsyncRemote().sendText(message);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 此为单点消息
     *
     * @param userId  用户编号
     * @param message 消息
     */
    public void sendOneMessage(String userId, Map<String,String> message) {
        Session session = SESSION_POOL.get(userId);
        if (session != null && session.isOpen()) {
            try {
                synchronized (session) {
                    log.info("【WebSocket消息】单点消息：" + message.get("message"));
                    session.getAsyncRemote().sendText(JSON.toJSONString(message));
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 此为单点消息(多人)
     *
     * @param userIds 用户编号列表
     * @param message 消息
     */
    public void sendMoreMessage(String[] userIds, String message) {
        for (String userId : userIds) {
            Session session = SESSION_POOL.get(userId);
            if (session != null && session.isOpen()) {
                try {
                    log.info("【WebSocket消息】单点消息：" + message);
                    session.getAsyncRemote().sendText(message);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    }

4.在需要的地方进行调用

（1）先注入websocket服务类

    @Autowired
    MyWebSocketHandler myWebSocketHandler;

（2）在方法中调用给前端发消息的方法

myWebSocketHandler.sendOneMessage(userId,messageInfor);

问题：

1.如果你在controller层调用了service层中身为异步的方法出现了HttpServeletrequst空指针你需要在controller层调用异步方法前加入下面的代码。

ServletRequestAttributes servletRequestAttributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
        RequestContextHolder.getRequestAttributes().setAttribute("request", request, RequestAttributes.SCOPE_REQUEST);
        RequestContextHolder.setRequestAttributes(servletRequestAttributes,true);//设置子线程共享

调用requst中的参数时必须使用下面的方法

 HttpServletRequest request2 = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
  if( request2.getHeader(UserContext.USER_NAME)!=null){
            return Optional.of(request2.getHeader(UserContext.USER_NAME));
        }