后端

service层新建类WebSocketServer

package com.my.blogback.service;

import java.io.IOException;
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 com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Component;


@ServerEndpoint("/socket/{userId}")//userId：地址的2就是这个userId"ws://localhost:8000/socket/2"
@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);
            webSocketMap.put(userId,this);
            //加入set中
        }else{
            webSocketMap.put(userId,this);
            //加入set中
            addOnlineCount();
            //在线数加1
        }

        System.out.println("用户连接============================:"+userId+",当前在线人数为:" + getOnlineCount());
        try {
            sendMessage("连接成功");
        } catch (IOException e) {
            System.out.println("用户:"+userId+",网络异常!!!!!!");
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        if(webSocketMap.containsKey(userId)){
            webSocketMap.remove(userId);
            //从set中删除
            subOnlineCount();
        }
        System.out.println("用户退出:"+userId+",当前在线人数为:" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息*/
    @OnMessage
    public void onMessage(String message, Session session) {
        System.out.println("用户消息:"+userId+",报文:"+message);
        //可以群发消息
        //消息保存到数据库、redis
        if(StringUtils.isNotBlank(message)){
            try {
                //解析发送的报文
                JSONObject jsonObject = JSON.parseObject(message);
                //追加发送人(防止串改)
                jsonObject.put("fromUserId",this.userId);
                String toUserId=jsonObject.getString("toUserId");
                //传送给对应toUserId用户的websocket
                if(StringUtils.isNotBlank(toUserId)&&webSocketMap.containsKey(toUserId)){
                    webSocketMap.get(toUserId).sendMessage(jsonObject.toJSONString());
                }else{
                    System.out.println("请求的userId:"+toUserId+"不在该服务器上");
                    //否则不在这个服务器上，发送到mysql或者redis
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

    /**
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        System.out.println("用户错误:"+this.userId+",原因:"+error.getMessage());
        error.printStackTrace();
    }
    /**
     * 实现服务器主动推送
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }


    /**
     * 发送自定义消息
     * */
    public static void sendInfo(String message,@PathParam("userId") String userId) throws IOException {
        System.out.println("发送消息到:"+userId+"，报文:"+message);
        if(StringUtils.isNotBlank(userId)&&webSocketMap.containsKey(userId)){
            webSocketMap.get(userId).sendMessage(message);
        }else{
            System.out.println("用户"+userId+",不在线！");
        }
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }
}

controller层新建类WebsocketController

package com.my.blogback.controller;

import com.my.blogback.entity.dto.Response;
import com.my.blogback.service.WebSocketServer;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.io.IOException;
@RestController
@RequestMapping("/socket")
public class WebsocketController {
    @Resource
    WebSocketServer webSocketServer;

	// 写一个主动推送接口，进行测试
    @GetMapping("/sendMes")
    public Response getAll() {
        try {
            // 这里是测试接口，就直接内容和userId写死了，现实中肯定是动态的
            webSocketServer.sendInfo("hello", "2");
           // 消息发送成功之后，可以自己再写一个接口，并调用这个接口，将websocket消息存到数据库.....这里就不过多写了
        } catch (IOException e) {
            e.printStackTrace();
        }
        return Response.success();
    }
}

前端

src/api下新建socket.js

import {getCurrentLoginUser} from "@/utils/auth";
import Vue from "vue";


var websocket = null

export function initWebSocket() {

    if ('WebSocket' in window) {
        console.log(window.location.host);
        websocket = new WebSocket(process.env.VUE_APP_SOCKET_PROTOCOL + '://' + window.location.host + '/api/socket/' + getCurrentLoginUser().id);
        // 连接错误
        websocket.onerror = setErrorMessage
        // 连接成功
        websocket.onopen = setOnopenMessage
        // 收到消息的回调
        websocket.onmessage = setOnmessageMessage
        // 连接关闭的回调
        websocket.onclose = setOncloseMessage
        // 监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
        window.onbeforeunload = onbeforeunload
    } else {
        alert('当前浏览器不支持WebSocket!!!')
    }
}

function setErrorMessage() {
    console.log('WebSocket连接发生错误，状态码：' + websocket.readyState)
}

function setOnopenMessage() {
    console.log('WebSocket连接成功，状态码：' + websocket.readyState)
}

function setOnmessageMessage(event) {
    // 根据服务器推送的消息做自己的业务处理
    Vue.prototype.$notify({
        title: '通知',
        message: event.data
    })
}

function setOncloseMessage() {
    console.log('WebSocket连接关闭，状态码：' + websocket.readyState)
}

function onbeforeunload() {
    closeWebSocket()
}

export function closeWebSocket() {

    if (websocket === null || websocket === undefined) {
        return
    }
    websocket.close()
}

export function getWebSocket() {
    return websocket
}

测试

  <el-button type="primary" @click="sendMes">
    测试<i class="el-icon-plus"></i>
  </el-button>

sendMes() {
  // 发起一个post请求
  let token = sessionStorage.getItem("tokenKey") ? sessionStorage.getItem("tokenKey") : null
  axios({
    method: 'get',
    url: 'http://localhost:9000/api/socket/sendMes',
    headers: {'token': token},
  });
},

当点击按钮触发sendMes函数时，就会调用socket.js中的setOnmessageMessage方法，这个setOnmessageMessage可以自定义设置为弹出窗之类的

结果