一、第一种方式-原生注解（tomcat内嵌）

1.1、引入依赖

   org.springframework.boot
   spring-boot-starter-websocket

1.2、配置文件

package cn.jt.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
/**
 * @author GXM
 * @version 1.0.0
 * @Description TODO
 * @createTime 2023年07月06日
 */
@Configuration
public class WebSocketConfig {
    /**
     * 初始化Bean，它会自动注册使用了 @ServerEndpoint 注解声明的 WebSocket endpoint
     *
     * @return
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

1.3、构建安全的WebSocket抽象层

1、该类可以作为一个基础的安全抽象层，后续项目中如果需要做认证的操作，都可以继承该抽象类

ClientUserInfoService 大家可以看作一个 UserService 就是一张用户表的service类

这里认证采用的是 jwt的方式，大家可以换成自己的

2、大家这里注意，我们使用的是 javax.websocket.Session; 这个是tomcat下的

package cn.jt.websocket;
import cn.jt.client.entity.ClientUserInfo;
import cn.jt.client.service.ClientUserInfoService;
import cn.jt.jwt.JwtUtils;
import cn.jt.utils.SpringContextUtils;
import lombok.extern.slf4j.Slf4j;
import javax.websocket.Session;
import java.io.IOException;
import java.util.Date;
/**
 * @author GXM
 * @version 1.0.0
 * @Description TODO
 * @createTime 2023年07月06日
 */
@Slf4j
public abstract class SecureWebSocket {
    private static final ClientUserInfoService clientUserInfoService;
    static {
        clientUserInfoService = SpringContextUtils.getBean(ClientUserInfoService.class);
    }
    protected Session session;
    protected String token;
    protected Long tokenExpiresAt;
    protected ClientUserInfo clientUserInfo;
    /**
     * 验证token是否有效（包含有效期）
     *
     * @param token  token
     * @param isInit 是否对token和userInfo进行初始化赋值
     * @return boolean
     */
    protected boolean isTokenValid(String token, boolean isInit) {
        ClientUserInfo clientUserInfo;
        try {
            clientUserInfo = JwtUtils.getClientUserInfo(token);
        } catch (Exception e) {
            log.error("ws 认证失败", e);
            return false;
        }
        if (isInit) {
            this.clientUserInfo = clientUserInfo;
            this.tokenExpiresAt = JwtUtils.getDecodedJWT(token).getExpiresAt().getTime();
            this.token = token;
        }
        return true;
    }
    /**
     * 认证失败，断开连接
     *
     * @param session session
     */
    protected void sendAuthFailed(Session session) {
        try {
            session.getBasicRemote().sendText("认证失败");
            session.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

1.4、构建基础的WebSocket

1、代码很简单，大家一看就知道逻辑了，这里就解释一下各个注解的含义

• @ServerEndpoint：将目前的类定义成一个websocket服务器端，注解的值将被用于监听用户连接的终端访问URL地址，客户端可以通过这个URL来连接到WebSocket服务器端
• @OnOpen：当WebSocket建立连接成功后会触发这个注解修饰的方法。
• @OnClose：当WebSocket建立的连接断开后会触发这个注解修饰的方法。
• @OnMessage：当客户端发送消息到服务端时，会触发这个注解修改的方法。
• @OnError：当WebSocket建立连接时出现异常会触发这个注解修饰的方法。

  2、大家这里注意，我们使用的是 javax.websocket.Session; 这个是tomcat下的

  

  package cn.jt.websocket;
  import com.alibaba.fastjson.JSON;
  import com.google.common.util.concurrent.ThreadFactoryBuilder;
  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.Map;
  import java.util.concurrent.*;
  /**
   * @author GXM
   * @version 1.0.0
   * @Description 
   * @createTime 2023年07月06日
   */
  @Slf4j
  @ServerEndpoint("/globalWs/{token}")
  @Component
  public class GlobalWebsocket extends SecureWebSocket {
   
      /**
       * key: userKye
       * value: GlobalWebsocket  这里你直接存储 session 也是可以的
       */
      private static final Map CLIENTS = new ConcurrentHashMap<>();
      /**
       * // 如果允许 一个账号 多人登录的话  就 加上  "-" + tokenTime，因为每次登录的token过期时间都是不一样的
       * clientUserInfo.getId() + "-" + clientUserInfo.getAccount() ;
       */
      private String userKye;
      @OnOpen
      public void onOpen(Session session, @PathParam("token") String token) {
          if (!isTokenValid(token, true)) {
              sendAuthFailed(session);
              return;
          }
          this.session = session;
          this.userKye = clientUserInfo.getId() + "-" + clientUserInfo.getAccount() + "-" + super.tokenExpiresAt;
          CLIENTS.put(userKye, this);
          log.info("当前在线用户:{}", CLIENTS.keySet());
          try {
              session.getBasicRemote().sendText("连接成功！");
          } catch (IOException e) {
              e.printStackTrace();
          }
      }
      @OnMessage
      public String onMessage(Session session, String message) {
          // 先判断当前token 是否已经到期了
          if (!isTokenValid(token, false)) {
              sendAuthFailed(session);
              return null;
          }
          try {
              session.getBasicRemote().sendText("received");
          } catch (IOException e) {
              e.printStackTrace();
          }
          return null;
      }
      @OnError
      public void onError(Session session, Throwable throwable) {
  //        log.error("ws session 发生错误,session key is {}",throwable);
          log.error("ws session 发生错误:{}", throwable.getMessage());
      }
      @OnClose
      public void onClose(Session session) {
          CLIENTS.remove(userKye);
          log.info("ws 用户 userKey {} 已下线,当前在线用户:{}", userKye, CLIENTS.keySet());
      }
      /**
       * 发送消息
       *
       * @param messageVo
       */
      public void sendMessage(MessageVo messageVo) {
          try {
              this.session.getBasicRemote().sendText(JSON.toJSONString(messageVo));
          } catch (IOException e) {
              log.error("发送消息异常", e);
          }
      }
      /**
       * 向user精确用户发送消息
       *
       * @param userKey   由 account + "-" + refreshToken的签发时间组成，例："admin-1635830649000"
       * @param messageVo 消息内容
       */
      public static void sendToUser(String userKey, MessageVo messageVo) {
          GlobalWebsocket globalWebsocket = CLIENTS.get(userKey);
          if (null != globalWebsocket) {
              globalWebsocket.sendMessage(messageVo);
              return;
          }
          log.error("发送消息到指定用户,但是用户不存在,userKey is {},message is {}", userKey, JSON.toJSONString(messageVo));
      }
      /**
       * 全体组播消息
       *
       * @param
       */
      public static void broadcast(MessageVo messageVo) {
          CLIENTS.values().forEach(c -> {
                      Session curSession = c.session;
                      if (curSession.isOpen()) {
                              try {
                                  curSession.getBasicRemote().sendText(JSON.toJSONString(messageVo));
                              } catch (IOException e) {
                                  log.error("发送ws数据错误:{}", e.getMessage());
                              }
                      }
                  }
          );
      }
  }
  

  1.5、SpringBoot 开启 WebSocket

  @EnableWebSocket
  

  

  1.6、高并发时候的问题

  1、这里要说明一下在高并发下的问题，如果你同时向在线的 3 个webSocket 在线客户端发送消息，即广播所有在线用户（目前是3个），每个用户每秒10条，那就是说，你每秒要发送 30 条数据，我们调用上述的广播函数 broadcast()，有时候会出现

  java.lang.IllegalStateException: 远程 endpoint 处于 [xxxxxx] 状态，如：

  The remote endpoint was in state [TEXT_FULL_WRITING] which is an invalid state for calle

  这是因为在高并发的情况下，出现了session抢占的问题，导致session，的状态不一致，所以，这里可以去尝试加锁操作，如下

   public static final ExecutorService WEBSOCKET_POOL_EXECUTOR = new ThreadPoolExecutor(
              20, 20,
              Integer.MAX_VALUE, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(),
              new ThreadFactoryBuilder()
                      .setNameFormat("GlobalWebsocket-executor-" + "%d")
                      .setUncaughtExceptionHandler((thread, throwable) -> log.error("ThreadPool {} got exception", thread, throwable)).build(),
              new ThreadPoolExecutor.AbortPolicy());
              
      /**
       * 全体组播消息
       *
       * @param
       */
      public static void broadcast(MessageVo messageVo) {
          CLIENTS.values().forEach(c -> {    
                      Session curSession = c.session;
                      if (curSession.isOpen()) {
                          // 建议单个session 一个线程，避免  一个session会话网络不好，会出现超时异常，当前线程会因此中断。
                          // 导致后面的session没有进行发送操作。使用单个线程，单个session情况下避免session之间的相互影响。
                          WEBSOCKET_POOL_EXECUTOR.execute(() -> {
                              synchronized (curSession) {
                                  // 双重锁检查，外边的 isOpen 第一遍过滤，里面枷加锁之后，第二遍过滤
                                  if (curSession.isOpen()) {
                                      try {
                                          curSession.getBasicRemote().sendText(JSON.toJSONString(messageVo));
                                      } catch (IOException e) {
                                          log.error("发送ws数据错误:{}", e.getMessage());
                                      }
                                  }
                              }
                          });
                      }
                  }
          );
      }
  

  其中增加了，双重锁检查，以及线程池操作，当然加上锁之后，性能是肯定会有所下降的

  建议单个session 一个线程，避免 一个session会话网络不好，会出现超时异常，当前线程会因此中断

  2、按照上述的代码，我这边测试12个webSocket 链接，每秒每个客户端都发送10条数据，相当于每秒发送120条数据，目前看来，速度还是不错的，但是当客户端重连后，偶尔会出现错误信息 远程主机已经关闭了一个链接,类似于这种错误，这条错误日志是在广播代码的如下位置打印的，这是因为当准备发送消息的时候，远程客户端还是关闭了。

   try {
                                          curSession.getBasicRemote().sendText(JSON.toJSONString(messageVo));
                                      } catch (IOException e) {
                                          log.error("发送ws数据错误:{}", e.getMessage());
                                      }
  

  二、第二种方式-Spring封装

  2.1、引入依赖

  
              org.springframework.boot
              spring-boot-starter-websocket
          
  

  2.2、自己的webSocket处理service

  1、WebSocketService 处理器类如下

  类似于 UserService 等等，主要是抽出一部分的业务逻辑

  package cn.jt.websocket.spring;
  import org.springframework.web.socket.TextMessage;
  import org.springframework.web.socket.WebSocketSession;
  import java.io.IOException;
  /**
   * @author GXM
   * @version 1.0.0
   * @Description TODO
   * @createTime 2023年07月19日
   */
  public interface WebSocketService {
      /**
       * 会话开始回调
       *
       * @param session 会话
       */
      void handleOpen(WebSocketSession session);
      /**
       * 会话结束回调
       *
       * @param session 会话
       */
      void handleClose(WebSocketSession session);
      /**
       * 处理消息
       *
       * @param session 会话
       * @param message 接收的消息
       */
      void handleMessage(WebSocketSession session, String message);
      /**
       * 发送消息
       *
       * @param session 当前会话
       * @param message 要发送的消息
       * @throws IOException 发送io异常
       */
      void sendMessage(WebSocketSession session, String message) throws IOException;
      /**
       * 发送消息
       *
       * @param userId  用户id
       * @param message 要发送的消息
       * @throws IOException 发送io异常
       */
      void sendMessage(Integer userId, TextMessage message) throws IOException;
      /**
       * 发送消息
       *
       * @param userId  用户id
       * @param message 要发送的消息
       * @throws IOException 发送io异常
       */
      void sendMessage(Integer userId, String message) throws IOException;
      /**
       * 发送消息
       *
       * @param session 当前会话
       * @param message 要发送的消息
       * @throws IOException 发送io异常
       */
      void sendMessage(WebSocketSession session, TextMessage message) throws IOException;
      /**
       * 广播
       *
       * @param message 字符串消息
       * @throws IOException 异常
       */
      void broadCast(String message) throws IOException;
      /**
       * 广播
       *
       * @param message 文本消息
       * @throws IOException 异常
       */
      void broadCast(TextMessage message) throws IOException;
      /**
       * 处理会话异常
       *
       * @param session 会话
       * @param error   异常
       */
      void handleError(WebSocketSession session, Throwable error);
  }
  

  2、WebSocketServiceImpl 实现类如下

  类似于 UserServiceImpl 等等，主要是抽出一部分的业务逻辑

  package cn.jt.websocket.spring;
  import cn.jt.client.entity.ClientUserInfo;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.web.socket.TextMessage;
  import org.springframework.web.socket.WebSocketSession;
  import java.io.IOException;
  import java.util.Map;
  import java.util.concurrent.ConcurrentHashMap;
  /**
   * @author GXM
   * @version 1.0.0
   * @Description TODO
   * @createTime 2023年07月19日
   */
  @Slf4j
  public class WebSocketServiceImpl implements WebSocketService {
      private final Map clients = new ConcurrentHashMap<>();
      @Override
      public void handleOpen(WebSocketSession session) {
          // 这个时候就需要在建立 webSocket 时存储的 用户信息了
          Map attributes = session.getAttributes();
          ClientUserInfo clientUserInfo = (ClientUserInfo) attributes.get("clientUserInfo");
          clients.put(clientUserInfo.getId(), session);
          log.info("a new connection opened，current online count：{}", clients.size());
      }
      @Override
      public void handleClose(WebSocketSession session) {
          // 这个时候就需要在建立 webSocket 时存储的 用户信息了
          Map attributes = session.getAttributes();
          ClientUserInfo clientUserInfo = (ClientUserInfo) attributes.get("clientUserInfo");
          clients.remove(clientUserInfo.getId());
          log.info("a new connection closed，current online count：{}", clients.size());
      }
      @Override
      public void handleMessage(WebSocketSession session, String message) {
          // 只处理前端传来的文本消息，并且直接丢弃了客户端传来的消息
          log.info("received a message：{}", message);
      }
      @Override
      public void sendMessage(WebSocketSession session, String message) throws IOException {
          this.sendMessage(session, new TextMessage(message));
      }
      @Override
      public void sendMessage(Integer userId, TextMessage message) throws IOException {
          WebSocketSession webSocketSession = clients.get(userId);
          if (webSocketSession.isOpen()) {
              webSocketSession.sendMessage(message);
          }
      }
      @Override
      public void sendMessage(Integer userId, String message) throws IOException {
          this.sendMessage(userId, new TextMessage(message));
      }
      @Override
      public void sendMessage(WebSocketSession session, TextMessage message) throws IOException {
          session.sendMessage(message);
      }
      @Override
      public void broadCast(String message) throws IOException {
          clients.values().forEach(session -> {
              if (session.isOpen()) {
                  try {
                      session.sendMessage(new TextMessage(message));
                  } catch (IOException e) {
                      throw new RuntimeException(e);
                  }
              }
          });
      }
      @Override
      public void broadCast(TextMessage message) throws IOException {
          clients.values().forEach(session -> {
              if (session.isOpen()) {
                  try {
                      session.sendMessage(message);
                  } catch (IOException e) {
                      throw new RuntimeException(e);
                  }
              }
          });
      }
      @Override
      public void handleError(WebSocketSession session, Throwable error) {
          log.error("websocket error：{}，session id：{}", error.getMessage(), session.getId());
          log.error("", error);
      }
  }
  

  2.3、实现spring框架的WebSocket处理器

  1、注意这里的 webSocketSession 就是 spring 包下的了，不再是 tomcat包下的了

  

  这里其实就和我们之前使用原生注解(tomcat)的那个一样了，都是几个特定的函数

  我们在特定的方法下，调用我们自己的 service去单独处理，解耦合

  package cn.jt.websocket.spring;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.web.socket.*;
  /**
   * @author GXM
   * @version 1.0.0
   * @Description TODO
   * @createTime 2023年07月19日
   */
  public class DefaultWebSocketHandler implements WebSocketHandler {
      @Autowired
      private WebSocketService webSocketService;
      /**
       * 建立连接
       *
       * @param session Session
       */
      @Override
      public void afterConnectionEstablished(WebSocketSession session) {
          webSocketService.handleOpen(session);
      }
      /**
       * 接收消息
       *
       * @param session Session
       * @param message 消息
       */
      @Override
      public void handleMessage(WebSocketSession session, WebSocketMessage message) {
          if (message instanceof TextMessage) {
              TextMessage textMessage = (TextMessage) message;
              webSocketService.handleMessage(session, textMessage.getPayload());
          }
      }
      /**
       * 发生错误
       *
       * @param session   Session
       * @param exception 异常
       */
      @Override
      public void handleTransportError(WebSocketSession session, Throwable exception) {
          webSocketService.handleError(session, exception);
      }
      /**
       * 关闭连接
       *
       * @param session     Session
       * @param closeStatus 关闭状态
       */
      @Override
      public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) {
          webSocketService.handleClose(session);
      }
      /**
       * 是否支持发送部分消息
       *
       * @return false
       */
      @Override
      public boolean supportsPartialMessages() {
          return false;
      }
  }
  

  2.4、自定义拦截器

  这里，我们可以设置拦截器，在做请求参数，或者权限认证的时候，不用在建立链接的函数afterConnectionEstablished里面去处理

  可以理解为 springMvc 每次请求前的拦截器

  package cn.jt.websocket.spring;
  import cn.jt.client.entity.ClientUserInfo;
  import cn.jt.jwt.JwtUtils;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.http.server.ServerHttpRequest;
  import org.springframework.http.server.ServerHttpResponse;
  import org.springframework.http.server.ServletServerHttpRequest;
  import org.springframework.web.socket.WebSocketHandler;
  import org.springframework.web.socket.server.HandshakeInterceptor;
  import java.util.Map;
  /**
   * @author GXM
   * @version 1.0.0
   * @Description TODO
   * @createTime 2023年07月19日
   */
  @Slf4j
  public class WebSocketInterceptor implements HandshakeInterceptor {
      /**
       * 建立请求之前，可以用来做权限判断
       *
       * @param request    the current request
       * @param response   the current response
       * @param wsHandler  the target WebSocket handler
       * @param attributes the attributes from the HTTP handshake to associate with the WebSocket
       *                   session; the provided attributes are copied, the original map is not used.
       * @return
       * @throws Exception
       */
      @Override
      public boolean beforeHandshake(ServerHttpRequest request,
                                     ServerHttpResponse response, WebSocketHandler wsHandler,
                                     Map attributes) throws Exception {
          if (request instanceof ServletServerHttpRequest) {
              ServletServerHttpRequest servletServerHttpRequest = (ServletServerHttpRequest) request;
              // 模拟用户（通常利用JWT令牌解析用户信息）
              String token = servletServerHttpRequest.getServletRequest().getParameter("token");
              try {
                  ClientUserInfo clientUserInfo = JwtUtils.getClientUserInfo(token);
                  // 设置当前这个session的属性，后续我们在发送消息时，可以通过 session.getAttributes().get("clientUserInfo")可以取出 clientUserInfo参数
                  attributes.put("clientUserInfo", clientUserInfo);
              } catch (Exception e) {
                  log.error("webSocket 认证失败 ", e);
                  return false;
              }
              return true;
          }
          return false;
      }
      /**
       * 建立请求之后
       *
       * @param request   the current request
       * @param response  the current response
       * @param wsHandler the target WebSocket handler
       * @param exception an exception raised during the handshake, or {@code null} if none
       */
      @Override
      public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,
                                 WebSocketHandler wsHandler, Exception exception) {
      }
  }
  

  2.5、WebSocket配置

  将自定义处理器、拦截器以及WebSocket操作类依次注入到IOC容器中。

  • @EnableWebSocket：开启WebSocket功能
  • addHandler：添加处理器
  • addInterceptors：添加拦截器
  • setAllowedOrigins：设置允许跨域（允许所有请求来源）

    package cn.jt.websocket.spring;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
    import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
    /**
     * @author GXM
     * @version 1.0.0
     * @Description TODO
     * @createTime 2023年07月19日
     */
    @Configuration
    public class WebSocketConfiguration implements WebSocketConfigurer {
        @Bean
        public DefaultWebSocketHandler defaultWebSocketHandler() {
            return new DefaultWebSocketHandler();
        }
        @Bean
        public WebSocketService webSocket() {
            return new WebSocketServiceImpl();
        }
        @Bean
        public WebSocketInterceptor webSocketInterceptor() {
            return new WebSocketInterceptor();
        }
        @Override
        public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
            // 链接方式如下 ws://127.0.0.1:9085/globalWs/message?token=qwncjncwqqdnjncz.adqwascsvcgrgb.cbrtbfvb
            // 如果你设置了springboot的 contentPath 那就需要在:9085端口后面 加上contentPath 的值，在拼接上  globalWs/message?token=qwncjncwqqdnjncz.adqwascsvcgrgb.cbrtbfvb
            registry.addHandler(defaultWebSocketHandler(), "/globalWs/message")
                    .addInterceptors(webSocketInterceptor())
                    .setAllowedOrigins("*");
        }
    }
    

    2.6、SpringBoot 开启 WebSocket

    @EnableWebSocket
    

    

    2.7、链接

    1、其中 thermal-api 是我的项目名称

    

    2、链接路径如下

    ws://127.0.0.1:9085/thermal-api/globalWs/message?token=qwncjncwqqdnjncz.adqwascsvcgrgb.cbrtbfvb
    

    2.8、高并发时候的问题

    1、如果在广播的时候，客户端很多，发送的消息也是很多，还是会出现和之前 第一种方式-原生注解（tomcat内嵌）相同的问题，出现类似如下报错

    The remote endpoint was in state [xxxx] which is an invalid state for calle

    2、错误分析可以看 踩坑笔记 Spring websocket并发发送消息异常，写的很清楚。

    2.8.1、解决方案一

    1、和之前一样，加锁

    @Override
        public void broadCast(String message) throws IOException {
            clients.values().forEach(session -> {
                if (session.isOpen()) {
                    synchronized (session){
                        try {
                            session.sendMessage(new TextMessage(message));
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    }
                }
            });
        }
    

    2.8.2、解决方案二

    1、使用 spring 的，Spring 的解决方案是把原来的 WebSocketSession 封了一层，即 org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator

    3、代码稍微改一下，如下

    @Override
        public void handleOpen(WebSocketSession session) {
            // 这个时候就需要在建立 webSocket 时存储的 用户信息了
            Map attributes = session.getAttributes();
            ClientUserInfo clientUserInfo = (ClientUserInfo) attributes.get("clientUserInfo");
            
            clients.put(clientUserInfo.getId(), new ConcurrentWebSocketSessionDecorator(session, 10 * 1000, 64000));
            log.info("a new connection opened，current online count：{}", clients.size());
        }
    

    第三种方式-TIO

    1、请上网了解，用的比较少，不做过多说明

    第四种方式-STOMP

    1、请上网了解，用的比较少，不做过多说明

     
     东莞网络推广营销  营销网课  东莞网站开发  沈阳网页制作公司哪家好  百度站长之家  怎么免费做网站