简介

本文主要是针对分布式场景下的使用websocket的一个解决方案。很遗憾的是，websocketsession是不支持序列化操作，所以也就不可能存在redis中。

我们知道在单节点中我们只需要把websocketsession存储在Map中就OK，每次发送通知都从map中根据clientID获取对应的websocket的session进行消息通知。但是在分布式多节点的系统中，每个节点的websocketsession是存在当前节点的内存中的，当A服务向A客户端推送消息时，B服务并不知道，此时B客户端就会无动于衷。所以存在websocketsession共享的问题，本文通过redis订阅广播的消息实现多节点服务同时向客户端推送消息。

Nginx配置

Nginx配置：

Windows Nginx安装：
官网下载链接: link.

选择Windows版本下载并解压
双击nginx.exe即可启动nginx
解压之后把前端文件放在nginx服务器上(nginx文件夹下)
在启动Nginx之前要配置nginx.conf，配置文件在conf文件夹下

#user  nobody;
worker_processes  1;

error_log  logs/error.log;
error_log  logs/error.log  notice;
error_log  logs/error.log  info;

#pid        logs/nginx.pid;


events {
    worker_connections  1024;
}
http {
	# 开启nginx对websocket的支持
    map $http_upgrade $connection_upgrade {
		default upgrade;
		'' close;
	}
	
    include       mime.types;
    default_type  application/octet-stream;

    #log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
    #                  '$status $body_bytes_sent "$http_referer" '
    #                  '"$http_user_agent" "$http_x_forwarded_for"';

    #access_log  logs/access.log  main;

    sendfile        on;
    #tcp_nopush     on;

    #keepalive_timeout  0;
    keepalive_timeout  65;
    #gzip  on;
	#配置上游服务器网关端口集群
	#upstream  backServer{
	    # 127.0.0.1:81 为网关地址  weight 为权重，值越大，访问到该台网关的几率越大
	    #server 127.0.0.1:8999 weight=1;  
	    #server 127.0.0.1:82 weight=1;
	#}
	#配置上游服务器 集群，默认轮询机制
    upstream backServer{
		#每个请求按访问ip的hash结果分配，这样每个访客固定访问一个后端服务器，可以解决session的问题.可查看参考Nginx的Upstream5种分配的方式
		ip_hash; 
        server 127.0.0.1:8999; # 网关地址
        server 127.0.0.1:8777; # 网关地址
        # 补充: backup表示从服务器或者叫备用服务器  只有当主服务器（81、82端口）都不能访问时才会访问此（83端口）备用服务器 当主服务器恢复正常后 则访问主服务器
        #server 127.0.0.1:83 backup;
    }
    server {
    	# 监听的请求地址及端口号
        listen       8200;
        server_name  localhost; 
     
        #charset koi8-r;

        #access_log  logs/host.access.log  main;

        location / {
            root   C:/nginx/nginx-1.21.1/front/;
            index  login.html;
        }
		# /aaa 代表请求地址中包含/aaa的会被分发到网关地址 
		location /aaa {
			  proxy_pass http://backServer;
			  proxy_redirect default;
			  # 开启跨域支持
			  add_header Access-Control-Allow-Origin *;
			  add_header Access-Control-Allow-Methods *;
			  add_header Access-Control-Allow-Headers 'DNT,X-Mx-ReqToken,Keep-Alive,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Authorization';
			  proxy_set_header Host $host:$server_port;
			  proxy_set_header X-Real-IP $remote_addr;
		}
		# /websocket-env-data 拦截并分发到websocket的地址
		location /websocket-env-data {
			  proxy_pass http://backServer;
			  proxy_redirect default;
			  add_header Access-Control-Allow-Origin *;
			  add_header Access-Control-Allow-Methods *;
			  add_header Access-Control-Allow-Headers 'DNT,X-Mx-ReqToken,Keep-Alive,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Authorization';
			  proxy_set_header Host $host:$server_port;
			  proxy_set_header X-Real-IP $remote_addr;
			  proxy_http_version 1.1;
			  # 开启nginx对websocket的支持，会将http请求转为websocket请求
			  proxy_set_header Upgrade $http_upgrade;
		    proxy_set_header Connection $connection_upgrade;
		}
		
        #error_page  404              /404.html;

        # redirect server error pages to the static page /50x.html
        #
        error_page   500 502 503 504  /50x.html;
        location = /login.html {
            root   front;
        }

        # proxy the PHP scripts to Apache listening on 127.0.0.1:80
        #
        #location ~ \.php$ {
        #    proxy_pass   http://127.0.0.1;
        #}

        # pass the PHP scripts to FastCGI server listening on 127.0.0.1:9000
        #
        #location ~ \.php$ {
        #    root           html;
        #    fastcgi_pass   127.0.0.1:9000;
        #    fastcgi_index  index.php;
        #    fastcgi_param  SCRIPT_FILENAME  /scripts$fastcgi_script_name;
        #    include        fastcgi_params;
        #}

        # deny access to .htaccess files, if Apache's document root
        # concurs with nginx's one
        #
        #location ~ /\.ht {
        #    deny  all;
        #}
    }


    # another virtual host using mix of IP-, name-, and port-based configuration
    #
    #server {
    #    listen       8000;
    #    listen       somename:8080;
    #    server_name  somename  alias  another.alias;

    #    location / {
    #        root   html;
    #        index  index.html index.htm;
    #    }
    #}


    # HTTPS server
    #
    #server {
    #    listen       443 ssl;
    #    server_name  localhost;

    #    ssl_certificate      cert.pem;
    #    ssl_certificate_key  cert.key;

    #    ssl_session_cache    shared:SSL:1m;
    #    ssl_session_timeout  5m;

    #    ssl_ciphers  HIGH:!aNULL:!MD5;
    #    ssl_prefer_server_ciphers  on;

    #    location / {
    #        root   html;
    #        index  index.html index.htm;
    #    }
    #}

}

Zuul websocket配置

zuul集群此处不再详细配置，具体参考网上教程
MyRequestInterceptor.java

import feign.RequestInterceptor;
import feign.RequestTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import org.springframework.web.context.request.RequestContextHolder;

import javax.servlet.http.HttpServletRequest;

@Configuration
@Component
public class MyRequestInterceptor implements RequestInterceptor {

    @Autowired
    HttpServletRequest request;

    @Override
    public void apply(RequestTemplate requestTemplate) {
//        System.out.println("MyRequestInterceptor apply begin.");
        try {
            String sessionId = RequestContextHolder.currentRequestAttributes().getSessionId();
            if (null != sessionId) {
                requestTemplate.header("Cookie", "SESSION=" + sessionId);
            }
        } catch (Exception e) {
            e.printStackTrace();
//            System.out.println("MyRequestInterceptor exception: "+ e);
        }
    }
}

SessionConfig.java


import jdk.nashorn.internal.runtime.GlobalConstants;
import org.springframework.context.annotation.Configuration;
import org.springframework.session.data.redis.RedisFlushMode;
import org.springframework.session.data.redis.config.annotation.web.http.EnableRedisHttpSession;
@Configuration
// 开启Spring对HttpSession和Redis的支持
@EnableRedisHttpSession(maxInactiveIntervalInSeconds = 86400*30, redisFlushMode = RedisFlushMode.IMMEDIATE)
public class SessionConfig {
    //  maxInactiveIntervalInSeconds: 设置 Session 失效时间，使用 Redis Session 之后，原 Boot 的 server.session.timeout 属性不再生效
}

WebSocketFilter.java


import com.netflix.zuul.ZuulFilter;
import com.netflix.zuul.context.RequestContext;
import org.springframework.stereotype.Component;

import javax.servlet.http.HttpServletRequest;

/**
 * 功能简述:
 * ZUUL开启websocket支持
 *
 * @date 2021/8/20
 * @since 1.0.0
 */
@Component
public class WebSocketFilter extends ZuulFilter {

    @Override
    public String filterType() {
        return "pre";
    }

    @Override
    public int filterOrder() {
        return 0;

    }

    @Override
    public boolean shouldFilter() {
        return true;

    }

    @Override
    public Object run() {
        RequestContext context = RequestContext.getCurrentContext();
        HttpServletRequest request = context.getRequest();
        System.out.println("request1:"+request.getHeader("host").toString());
        String upgradeHeader = request.getHeader("Upgrade");
        if (null == upgradeHeader) {
            upgradeHeader = request.getHeader("upgrade");
        }
        if (null != upgradeHeader && "websocket".equalsIgnoreCase(upgradeHeader)) {
            context.addZuulRequestHeader("connection", "Upgrade");
        }
//        System.out.println("request2:"+request.toString());
        return null;
    }
}

在ZuulApplication开启Websocket过滤

	/**
     * WebSocket过滤器
     *
     * @return 自定义访问过滤器
     */
    @Bean
    public WebSocketFilter webSocketFilter() {
        return new WebSocketFilter();
    }

在这里插入图片描述
在zuul的配置文件中加入

## @FeignClient(value = "服务名r") 设置可以有多个类存在相同的FeignClient 中的value值
spring.main.allow-bean-definition-overriding=true

在pom.xml中加入相关jar包

<!-- https://mvnrepository.com/artifact/org.springframework.session/spring-session-data-redis -->
    <dependency>
      <groupId>org.springframework.session</groupId>
      <artifactId>spring-session-data-redis</artifactId>
      <version>2.3.0.RELEASE</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/org.springframework.data/spring-data-redis -->
    <dependency>
      <groupId>org.springframework.data</groupId>
      <artifactId>spring-data-redis</artifactId>
      <version>2.3.0.RELEASE</version>
    </dependency>

Redis配置及websocket配置

在这里插入图片描述

RedisConfig

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration
public class RedisConfig {
    // 注入 RedisConnectionFactory
    @Autowired
    private RedisConnectionFactory redisConnectionFactory;

    @Bean
    public RedisTemplate<String, Object> functionDomainRedisTemplate() {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        initDomainRedisTemplate(redisTemplate, redisConnectionFactory);
        return redisTemplate;
    }

    /**
     * 设置数据存入 redis 的序列化方式
     * @param redisTemplate
     * @param factory
     */
    private void initDomainRedisTemplate(RedisTemplate<String, Object> redisTemplate, RedisConnectionFactory factory) {
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new JdkSerializationRedisSerializer());
        redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
        redisTemplate.setConnectionFactory(factory);
    }

    /**
     * 实例化 HashOperations 对象,可以使用 Hash 类型操作
     * @param redisTemplate
     * @return
     */
    @Bean
    public HashOperations<String, String, Object> hashOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForHash();
    }

    /**
     * 实例化 ValueOperations 对象,可以使用 String 操作
     * @param redisTemplate
     * @return
     */
    @Bean
    public ValueOperations<String, Object> valueOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForValue();
    }

    /**
     * 实例化 ListOperations 对象,可以使用 List 操作
     * @param redisTemplate
     * @return
     */
    @Bean
    public ListOperations<String, Object> listOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForList();
    }

    /**
     * 实例化 SetOperations 对象,可以使用 Set 操作
     * @param redisTemplate
     * @return
     */
    @Bean
    public SetOperations<String, Object> setOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForSet();
    }

    /**
     * 实例化 ZSetOperations 对象,可以使用 ZSet 操作
     * @param redisTemplate
     * @return
     */
    @Bean
    public ZSetOperations<String, Object> zSetOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForZSet();
    }
}

RedisMsg

import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;

@Component
public interface RedisMsg {
	//新增socket
	void afterConnectionEstablished(WebSocketSession session) throws Exception;

	//接收socket信息
	void handleMessage(WebSocketSession webSocketSession, WebSocketMessage<?> webSocketMessage) throws Exception;

	void handleTransportError(WebSocketSession session, Throwable exception) throws Exception;

	void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception;

	boolean supportsPartialMessages();

	/**
	 * 接收redis广播的订阅信息
	 * @param message
	 */
	public void receiveMessage(String message);
}

RedisPublishConfig

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;

@Configuration
@Component
public class RedisPublishConfig {
	/*@Autowired
	private StaticProperties staticProperties;*/
	/**
	 * redis消息监听器容器 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器
	 * 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理
	 * 
	 * @param connectionFactory
	 * @param listenerAdapter
	 * @return
	 */
	@Bean
	// 相当于xml中的bean
	RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
											MessageListenerAdapter listenerAdapter) {
		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
		container.setConnectionFactory(connectionFactory);
		// 订阅了一个叫chat 的通道
		container.addMessageListener(listenerAdapter, new PatternTopic("chat"));
		// 这个container 可以添加多个 messageListener
		return container;
	}
 
	/**
	 * 消息监听器适配器，绑定消息处理器，利用反射技术调用消息处理器的业务方法
	 * 
	 * @param receiver
	 * @return
	 */
	@Bean
	MessageListenerAdapter listenerAdapter(RedisMsg receiver) {
		// 这个地方 是给messageListenerAdapter 传入一个消息接受的处理器，利用反射的方法调用“receiveMessage”
		// 也有好几个重载方法，这边默认调用处理器的方法 叫handleMessage 可以自己到源码里面看
		return new MessageListenerAdapter(receiver, "receiveMessage");
	}
 
}

SpringUtilsCopy WebsocketHandler不支持自动注入

import org.springframework.beans.BeansException;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.stereotype.Component;
 
@Component
public class SpringUtilsCopy implements BeanFactoryPostProcessor {
 
    private static ConfigurableListableBeanFactory beanFactory; // Spring应用上下文环境
 
    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
        SpringUtilsCopy.beanFactory = beanFactory;
    }
 
    public static ConfigurableListableBeanFactory getBeanFactory() {
        return beanFactory;
    }
 
    /**
     * 获取对象
     *
     * @param name
     * @return Object 一个以所给名字注册的bean的实例
     * @throws org.springframework.beans.BeansException
     *
     */
    @SuppressWarnings("unchecked")
    public static <T> T getBean(String name) throws BeansException {
        return (T) getBeanFactory().getBean(name);
    }
 
    /**
     * 获取类型为requiredType的对象
     *
     * @param clz
     * @return
     * @throws org.springframework.beans.BeansException
     *
     */
    public static <T> T getBean(Class<T> clz) throws BeansException {
        T result = (T) getBeanFactory().getBean(clz);
        return result;
    }
 
    /**
     * 如果BeanFactory包含一个与所给名称匹配的bean定义，则返回true
     *
     * @param name
     * @return boolean
     */
    public static boolean containsBean(String name) {
        return getBeanFactory().containsBean(name);
    }
 
    /**
     * 判断以给定名字注册的bean定义是一个singleton还是一个prototype。 如果与给定名字相应的bean定义没有被找到，将会抛出一个异常（NoSuchBeanDefinitionException）
     *
     * @param name
     * @return boolean
     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
     *
     */
    public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException {
        return getBeanFactory().isSingleton(name);
    }
 
    /**
     * @param name
     * @return Class 注册对象的类型
     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
     *
     */
    public static Class<?> getType(String name) throws NoSuchBeanDefinitionException {
        return getBeanFactory().getType(name);
    }
 
    /**
     * 如果给定的bean名字在bean定义中有别名，则返回这些别名
     *
     * @param name
     * @return
     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
     *
     */
    public static String[] getAliases(String name) throws NoSuchBeanDefinitionException {
        return getBeanFactory().getAliases(name);
    }
}

WebSocketClient

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 功能简述:
 * websocket
 *
 * @date 2021/8/24
 * @since 1.0.0
 */
@Component
@Slf4j
public class WebSocketClient {

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 向登录指定用户的所有客户端推送消息
     * @param messageContent 消息内容
     * @param userName 推送用户
     */
    public void pushInfo(String messageContent,String userName){
        // TODO Auto-generated method stub
        JSONObject result = new JSONObject();
        if(StringUtil.isBlank(messageContent)) {
            result.put("result", "error");
        }else {
            try {
                //发送失败广播出去，让其他节点发送
                //广播消息到各个订阅者
                JSONObject message = new JSONObject();
                message.put("userName", userName);
                message.put("message", messageContent);
                // 通过redis的订阅发布消息，所有订阅的用户均可以收到消息
                redisTemplate.convertAndSend("chat",message.toString());

            } catch (Exception e) {
                e.printStackTrace();
                log.error("推送给客户端失败");
            }
            result.put("result", "success");
        }
        return ;
    }

    /**
     * 向所有客户端推送消息
     * @param messageContent
     */
    public void pushInfoToAll(String messageContent){
        // TODO Auto-generated method stub
        JSONObject result = new JSONObject();
        if(StringUtil.isBlank(messageContent)) {
            result.put("result", "error");
        }else {
            try {
                //发送失败广播出去，让其他节点发送
                //广播消息到各个订阅者
                JSONObject message = new JSONObject();
                message.put("userName", "");
                message.put("message", messageContent);
                // 通过redis的订阅发布消息，所有订阅的用户均可以收到消息
                redisTemplate.convertAndSend("chat",message.toString());

            } catch (Exception e) {
                e.printStackTrace();
                log.error("推送给客户端失败");
            }
            result.put("result", "success");
        }
        return ;
    }

    /**
     * 向多个指定用户推送消息
     * @param messageContent
     */
    public void pushInfoToUsers(String messageContent, List<SysUser> userList){
        // Auto-generated method stub
        JSONObject result = new JSONObject();
        if(StringUtil.isBlank(messageContent)) {
            result.put("result", "error");
        }else {
            try {
                userList.stream().forEach(p ->{
                    //广播消息到各个订阅者
                    JSONObject message = new JSONObject();
                    message.put("userName", p.getLoginName());
                    message.put("message", messageContent);
                    // 通过redis的订阅发布消息，所有订阅的用户均可以收到消息
                    redisTemplate.convertAndSend("chat",message.toString());
                });
            } catch (Exception e) {
                e.printStackTrace();
                log.error("推送给客户端失败");
            }
            result.put("result", "success");
        }
        return ;
    }
}

WebSocketConfig

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;

/***
 * WebSocketConfig
 */
@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {


    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        // webSocket通道
        // 指定处理器和路径
        registry.addHandler(new CTIHandler(), "/websocket-env-data/{userId}")
                // 指定自定义拦截器
                .addInterceptors(new WebSocketInterceptor())
                // 允许跨域
                .setAllowedOrigins("*");
        // sockJs通道
        registry.addHandler(new CTIHandler(), "/sock-js")
                .addInterceptors(new WebSocketInterceptor())
                .setAllowedOrigins("*")
                // 开启sockJs支持
                .withSockJS();

        registry.addHandler(new CTIHandler(), "/websocket-env-data1")
                // 指定自定义拦截器
                .addInterceptors(new WebSocketInterceptor())
                // 允许跨域
                .setAllowedOrigins("*");
    }

}

WebSocketInterceptor

import lombok.extern.slf4j.Slf4j;
import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.support.HttpSessionHandshakeInterceptor;
import java.util.Map;

@Slf4j
public class WebSocketInterceptor extends HttpSessionHandshakeInterceptor {


    @Override
    public boolean beforeHandshake(ServerHttpRequest serverHttpRequest, ServerHttpResponse seHttpResponse,
                                   WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
//		HttpServletRequest request = ((ServletServerHttpRequest) serverHttpRequest).getServletRequest();
        String[] aa = serverHttpRequest.getURI().toString().split("/");
        String userName = aa[aa.length-1];
        attributes.put("userName", userName);
        log.info("握手之前");
        //从request里面获取对象，存放attributes
        return super.beforeHandshake(serverHttpRequest, seHttpResponse, wsHandler, attributes);
    }

    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,
                               Exception ex) {
        log.info("握手之后");
        super.afterHandshake(request, response, wsHandler, ex);
    }

}

CTIHandler

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.*;

import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;

@Service
@Slf4j
public class CTIHandler implements WebSocketHandler,RedisMsg{

    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
     */
    private static ConcurrentHashMap<String, WebSocketSession> socketMap = new ConcurrentHashMap<String, WebSocketSession>();
    //新增socket
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        log.info("websocket连接成功");
        //获取用户信息
        String userName = (String) session.getAttributes().get("userName");
//        log.info("获取当前"+socketMap.get(userName));
        if(socketMap.get(userName)==null) {
            socketMap.put(userName,session);
            sendMessageToUser(userName, new TextMessage("链接建立成功"));
            //并且通过redis发布和订阅广播给其他的的机器，或者通过消息队列
        }
//        log.info("链接成功");
    }

    //接收socket信息
    @Override
    public void handleMessage(WebSocketSession webSocketSession, WebSocketMessage<?> webSocketMessage) throws Exception {
//        log.info("收到信息"+webSocketMessage.toString());
        String userName = (String) webSocketSession.getAttributes().get("userName");
        synchronized (webSocketSession){
            webSocketSession.sendMessage(new TextMessage("aaa"));
        }
        sendMessageToUser(userName, new TextMessage("我收到你的信息了"));
    }

    /**
     * 发送信息给指定用户 （所有登录该账号的客户端）
     * @param clientId 指定用户
     * @param message
     * @return
     */
    public boolean sendMessageToUser(String clientId, TextMessage message) {
        socketMap.forEach((key, value) -> {
            boolean flag = true;
            if(StringUtil.isNotBlank(clientId)){
                // 向指定用户发送消息
                if(key.contains(clientId+"websocket")){
                    WebSocketSession session = value;
                    if(session==null) {
                        flag = false;
                    }
//                log.info("进入发送消息");
                    if (!session.isOpen()) {
                        flag = false;
                    }
                    try {
                        if(flag){
//                        log.info("正在发送消息");
                            synchronized (session){
                                session.sendMessage(message);
                            }
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }else{
                // 如果指定用户为空，则向所有客户端发送消息
                WebSocketSession session = value;
                if(session==null) {
                    flag = false;
                }
//                log.info("进入发送消息");
                if (!session.isOpen()) {
                    flag = false;
                }
                try {
                    if(flag){
//                        log.info("正在发送消息");
                        synchronized (session){
                            session.sendMessage(message);
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
        return true;
    }


    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        if (session.isOpen()) {
            session.close();
        }
        log.info("连接出错");
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        //获取用户信息
        String userName = (String) session.getAttributes().get("userName");
        if(socketMap.get(userName)!=null) {
            socketMap.remove(userName);
            //并且通过redis发布和订阅广播给其他的的机器，或者通过消息队列
        }
        log.info("连接已关闭：" + status);
    }

    @Override
    public boolean supportsPartialMessages() {
        return false;
    }
    /**
     * 接受订阅信息
     */
    @Override
    public void receiveMessage(String message) {
        // TODO Auto-generated method stub
        JSONObject sendMsg = JSONObject.parseObject(message.substring(message.indexOf("{")));
        String clientId = sendMsg.getString("userName");
        TextMessage receiveMessage = new TextMessage(sendMsg.getString("message"));
        // 获取当前内存中clientId 的session,根据 clientId (即userName）推送消息
        boolean flag = sendMessageToUser(clientId, receiveMessage);
        if(flag) {
            log.info("推送消息("+sendMsg.getString("message")+")成功！！！");
        }
    }
}

调用WebSocketClient的pushInfo方法即可实现推送

前端代码

let callback = null;
let ws = null;
let close = null;
let lockReconnect = false; //避免重复连接
let stop = false;
let closeCallback = null;
let errorCallback = null

function getIp() {
  var ip
  if (httpUrl) {
    return ip = httpUrl.split(":")[1]
  }
}


function initWebSocket() {
  if (typeof WebSocket == "undefined") {
    console.log("当前浏览器 Not support websocket");
  } else {
    // 用户名  token
    var loginName = JSON.parse(localStorage.getItem('userInfo')).loginName;
    var userToken = localStorage.getItem('token');
    var userFlag = loginName +  "websocket" + userToken;
    var wsUrl = "ws:" + getIp() + ":8080/websocket-env-data/"+userFlag;
    console.log("websocket URL:"+wsUrl);
    if (window.soketFlag == null) {
      ws = new WebSocket(wsUrl);
      window.soketFlag = ws;
    } else {
      ws = window.soketFlag;
    }

    ws.onopen = function () {
      // $message.success("WebSocket连接成功")
      heartCheck.reset().start(); //传递信息
      console.log("WebSocket连接成功");
    };
    ws.onerror = function () {
      console.log("WebSocket连接失败1");
      // setTimeout(function () {
      //   window.location.reload()
      // }, 2000);
      reconnect(wsUrl);
      // $message.success("WebSocket连接失败");
      console.log("WebSocket连接失败2");
      if (typeof errorCallback === "function") {
        errorCallback("WebSocket连接失败");
      }
    };
    ws.onclose = function () {
      reconnect(wsUrl);
      // $message.success("WebSocket关闭");
      console.log("WebSocket关闭");
      // setTimeout(function () {
      //   window.location.reload()
      // }, 2000);
      if (typeof closeCallback === "function") {
        closeCallback("WebSocket关闭");
      }
    };
    close = ws.onclose;
    ws.onmessage = function (e) {
      // console.log("心跳开始");
      heartCheck.reset().start();
      if (typeof callback === "function") {
        callback(e.data);
      }
    };
  }
}

function setStop() {
  stop = true;
}

//websocket重连
function reconnect(url) {
  if (stop) {
    return;
  }
  if (lockReconnect) {
    return;
  }
  lockReconnect = true;
  setTimeout(function () {
    console.log("重连中");
    initWebSocket();
    lockReconnect = false;
  }, 2000);
}

// 心跳检测
//websocket心跳检测
var heartCheck = {
  timeout: 1000 * 25,
  timeoutObj: null,
  // serverTimeoutObj: null,
  reset: function () {
    clearTimeout(this.timeoutObj);
    // clearTimeout(this.serverTimeoutObj);
    return this;
  },
  start: function () {
    var self = this;
    this.timeoutObj = setTimeout(function () {
      //这里发送一个心跳，后端收到后，返回一个心跳消息，
      //onmessage拿到返回的心跳就说明连接正常
      ws.send("HeartBeat");
      // console.log("心跳开始");

      // self.serverTimeoutObj = setTimeout(function() {
      //   //如果超过一定时间还没重置，说明后端主动断开了
      //   console.log("关闭服务");
      //   wsReconnect(); //重新连接
      //   // ws.close(); //如果onclose会执行reconnect，我们执行ws.close()就行了.如果直接执行reconnect 会触发onclose导致重连两次
      // }, self.timeout);
    }, this.timeout);
  }
};

// send
function websock(sendData) {
  if (ws.readyState === ws.OPEN) {
    // 若是开启状态
    // ws.send(sendData);
  } else if (ws.readyState === ws.CONNECTING) {
    // 若是正在开启状态 则等待1s后重新调用
    setTimeout(function () {
      websock(sendData);
    }, 1000);
  } else if (ws.readyState === ws.CLOSED) {
    setTimeout(function () {
      initWebSocket();
      websock(sendData);
    }, 1000);
  } else {
    // 若是未开启状态 则等待1s重新调用
    setTimeout(function () {
      websock(sendData);
    }, 1000);
  }
}

// bing  onmessage
function bingWebsockMsg(call) {
  callback = call;
}

function wsOnCloseMsg(call) {
  closeCallback = call;

}

function wsOnErrorMsg(call) {
  errorCallback = call
}

// close
function closeWs() {
  close();
}


// #index.html
function divShow(e) {
  // console.log(e)
  if (e == "消息发布") {
    wsFlag = true
    // 调用相关方法
    isHasMsg2()
  } 
}