• Websocket升级版

    之前写过一个关于websocket的博客，是看书时候做的一个demo。但是纸上得来终觉浅，这次实战后实打实的踩了不少坑，写个博客记录总结。

    1.安装postman

    websocket接口调试，需要8.5以及以上版本的postman

    先把以前的卸载，直接exe安装就好

    点击这个连接可以下载，或者去官网下载

    https://download.csdn.net/download/qq_35653822/88419296

    2.postman调试方法

    2.1参考这个博客的postman调试方法

    Postman进行Websocket接口测试_postman websocket-CSDN博客

    2.2请求路径

    不是http开头  而是ws开头

    如 ws://127.0.0.1:9999/test/3

    2.3测试方式

    websocket接口和普通接口不同，需要先建立连接，再发送消息

    建立连接时候这个url可以是pathvariable那种

    如 /test/{id}

    3.上代码

    3.1controller层

    1.@ServerEndpoint
    这个注解算是websocket中极其重要的一个注解了
    这里边要配置上前后端建立连接的url，后端返回前端内容的解析器。我这写了两个分别对应字符串和指定实体的解析器。以及websocket对应的config文件

    2. @OnOpen  @OnMessage

    @OnOpen 前后端建立连接时走的方法，也就是postman点击connect按钮时候会走的方法

    @OnMessage 前端给后端发送消息时走的方法，也就是postman点击send的时候会走的方法

    3.session.getBasicRemote().sendObject();

    后端给前端发送消息，此处根据类型不同，走不同的解析器。也就是@ServerEndpoint中的

    encoders

    4.注入bean

    根据idea提示，需要用set方法注入 

    1. package websocket;
    2.  
    3. import com.fasterxml.jackson.databind.ObjectMapper;
    4. import com.valley.common.result.Result;
    5. import com.valley.system.enums.LinkType;
    6. import com.valley.system.pojo.bo.ConnectResultBO;
    7. import com.valley.system.pojo.form.SourceForm;
    8. import com.valley.system.service.SysDataSourceService;
    9. import lombok.extern.slf4j.Slf4j;
    10. import org.springframework.beans.factory.annotation.Autowired;
    11. import org.springframework.messaging.handler.annotation.MessageMapping;
    12. import org.springframework.messaging.handler.annotation.SendTo;
    13. import org.springframework.messaging.simp.annotation.SendToUser;
    14. import org.springframework.stereotype.Component;
    15. import org.springframework.web.bind.annotation.*;
    16.  
    17. import javax.websocket.EncodeException;
    18. import javax.websocket.OnMessage;
    19. import javax.websocket.OnOpen;
    20. import javax.websocket.Session;
    21. import javax.websocket.server.PathParam;
    22. import javax.websocket.server.ServerEndpoint;
    23. import java.io.IOException;
    24. import java.security.Principal;
    25.  
    26. /**
    27.  * @Description: websocket  url中得sourceId是为了指定websocket得连接
    28.  */
    29.  
    30. @ServerEndpoint(value = "/linkTest/{sourceId}", encoders = {ServerEncoder.class,StringEncoder.class},configurator = WebSocketConfig.class)
    31. @Component
    32. @Slf4j
    33. public class WebSocketController {
    34.  
    35.     private static SysDataSourceService sysDataSourceService;
    36.  
    37.     @Autowired
    38.     public void setDeviceListenerService(SysDataSourceService sysDataSourceService) {
    39.         WebSocketController.sysDataSourceService = sysDataSourceService;
    40.     }
    41.  
    42.     /**
    43.      * 实现服务器主动推送
    44.      */
    45.     public void sendMessage(String message, Session session) throws IOException {
    46.  
    47.         session.getBasicRemote().sendText(message);
    48.     }
    49.  
    50.     /**
    51.      * 连接建立成功调用的方法
    52.      */
    53.     @OnOpen
    54.     public void onOpen(Session session, @PathParam("sourceId") String sourceId) throws EncodeException, IOException {
    55.         log.info("连接成功，sourceId:{}", sourceId);
    56.         session.getBasicRemote().sendObject(sourceId);
    57.     }
    58.  
    59.     /**
    60.      * 收到客户端消息后调用的方法
    61.      *
    62.      * @param message 客户端发送过来的消息
    63.      */
    64.     @OnMessage
    65.     public void onMessage(String message, Session session) throws Exception {
    66.         log.info("【websocket消息】收到客户端发来的消息:{}", message);
    67.         // 创建ObjectMapper对象
    68.         ObjectMapper objectMapper = new ObjectMapper();
    69.  
    70.         // 将JSON字符串解析为JSON对象
    71.         SourceForm sourceForm = objectMapper.readValue(message, SourceForm.class);
    72.         //根据type判断是否是心跳连接
    73.         if(LinkType.HEARTBEAT.equals(sourceForm.getLinkType())){
    74.             //前端约定返回“pong”
    75.             session.getBasicRemote().sendObject("pong");
    76.             return;
    77.         }
    78.         ConnectResultBO bo = sysDataSourceService.linkTest(sourceForm);
    79.         try {
    80.             Result result = Result.success(bo);
    81.             session.getBasicRemote().sendObject(result);
    82.             //session.getBasicRemote().sendText(message.toString());
    83.         } catch (IOException e) {
    84.             e.printStackTrace();
    85.         }
    86.     }
    87.  
    88. }

    3.2config层

    1.需要配置在controller层的@ServerEndpoint中

    @ServerEndpoint(value = "/api/v1/source/linkTest/{sourceId}", encoders = {ServerEncoder.class,StringEncoder.class},configurator = WebSocketConfig.class)

    2. @Configuration修饰

    因为要返回其他的bean

    3.websocket的token上传返回

    需要取header中的 Sec-WebSocket-Protocol 属性，并且返回给前端的时候也要在header中携带

    1. package websocket;
    2.  
    3. import cn.hutool.core.lang.Assert;
    4. import com.valley.system.config.Constant;
    5. import jodd.util.StringUtil;
    6. import org.springframework.context.annotation.Bean;
    7. import org.springframework.context.annotation.Configuration;
    8. import org.springframework.messaging.simp.config.MessageBrokerRegistry;
    9. import org.springframework.security.oauth2.core.OAuth2AccessToken;
    10. import org.springframework.web.socket.config.annotation.AbstractWebSocketMessageBrokerConfigurer;
    11. import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
    12. import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
    13. import org.springframework.web.socket.server.standard.ServerEndpointExporter;
    14.  
    15. import javax.websocket.HandshakeResponse;
    16. import javax.websocket.server.HandshakeRequest;
    17. import javax.websocket.server.ServerEndpointConfig;
    18. import java.util.Collections;
    19. import java.util.List;
    20. import java.util.Map;
    21.  
    22. /**
    23.  * @Description: websocket配置
    24.  */
    25. @Configuration
    26. public class WebSocketConfig extends ServerEndpointConfig.Configurator{
    27.  
    28.     @Bean
    29.     public ServerEndpointExporter serverEndpointExporter(){
    30.         return new ServerEndpointExporter();
    31.     }
    32.  
    33.     /**
    34.      * 建立握手时，连接前的操作
    35.      */
    36.     @Override
    37.     public void modifyHandshake(ServerEndpointConfig sec, HandshakeRequest request, HandshakeResponse response) {
    38.         // 这个userProperties 可以通过 session.getUserProperties()获取
    39.         //获取请求头
    40.         String token = request.getHeaders().get("Sec-WebSocket-Protocol").get(0);
    41.         Assert.notNull(token,"token不可为空");
    42.         //todo 之后需要对token进行解析校验
    43.         //OAuth2AccessToken accessToken = tokenStore.readAccessToken(token);
    44.         //前端会把空格传成 
    45.         //token = token.replace(Constant.NBSP," ");
    46.         //当Sec-WebSocket-Protocol请求头不为空时,需要返回给前端相同的响应
    47.         response.getHeaders().put("Sec-WebSocket-Protocol", Collections.singletonList(token));
    48.         super.modifyHandshake(sec, request, response);
    49.     }
    50.  
    51.     /**
    52.      * 初始化端点对象,也就是被@ServerEndpoint所标注的对象
    53.      */
    54.     @Override
    55.     public  T getEndpointInstance(Class clazz) throws InstantiationException {
    56.         return super.getEndpointInstance(clazz);
    57.     }
    58. }
    59.

    3.3解析器

    1.如果不配置，在调用

    session.getBasicRemote().sendObject(sourceId);

    返回方法的时候会报此类型解析器不存在的错误

    2.解析器写完之后要加到controller层的@ServerEndpoint中

    @ServerEndpoint(value = "/api/v1/source/linkTest/{sourceId}", encoders = {ServerEncoder.class,StringEncoder.class},configurator = WebSocketConfig.class)

    3.3.1实体解析器

    1. package websocket;
    2.  
    3. import com.fasterxml.jackson.core.JsonProcessingException;
    4. import com.fasterxml.jackson.databind.json.JsonMapper;
    5. import com.valley.common.result.Result;
    6. import com.valley.system.pojo.bo.ConnectResultBO;
    7.  
    8. import javax.websocket.Encoder;
    9. import javax.websocket.EndpointConfig;
    10.  
    11. /**
    12.  * @Description: websocket发送给前端 泛型中Result是要发送的类型
    13.  */
    14. public class ServerEncoder implements Encoder.Text> {
    15.  
    16.     @Override
    17.     public void destroy() {
    18.         // TODO Auto-generated method stub
    19.         // 这里不重要
    20.     }
    21.  
    22.     @Override
    23.     public void init(EndpointConfig arg0) {
    24.         // TODO Auto-generated method stub
    25.         // 这里也不重要
    26.  
    27.     }
    28.  
    29.     @Override
    30.     public String encode(Result responseMessage) {
    31.         try {
    32.             JsonMapper jsonMapper = new JsonMapper();
    33.             return jsonMapper.writeValueAsString(responseMessage);
    34.  
    35.         } catch (JsonProcessingException e) {
    36.             e.printStackTrace();
    37.             return null;
    38.         }
    39.     }
    40. }

     3.3.2字符串解析器

    现在复盘来看，我这个解析器写的完全没必要，如果要发送字符串  把

    session.getBasicRemote().sendObject

    改为

    session.getBasicRemote().sendText

    就行了

    1. public class StringEncoder implements Encoder.Text{
    2.     @Override
    3.     public String encode(String s){
    4.         return s;
    5.     }
    6.  
    7.     @Override
    8.     public void init(EndpointConfig endpointConfig) {
    9.  
    10.     }
    11.  
    12.     @Override
    13.     public void destroy() {
    14.  
    15.     }
    16. }

    4.security权限过滤

    这块踩了很久的坑，我们项目是开源项目改的，然后自己拆成了好几个微服务。我开始的时候只专注去改auth项目中的，不管是忽略路径permitall，还是加切面，还是加filter。都不生效。

    最后，在公共项目common中找到一个关于security的配置，改了permitall，才生效

    但是还要注意哈，既然在这里把权限放开了，那在config中，就得把权限校验加上

    此处放开是因为，上边说过前端只能把token放到header的Sec-WebSocket-Protocol属性中，而不能像平时一样放到Authorization属性中。

     5.插曲

    在进行实体转换的时候，前端多给我传了参数，引起了报错

    objectMapper.readValue(message, SourceForm.class);

    为了让代码健壮一点，给实体加了个注解

    @JsonIgnoreProperties(ignoreUnknown = true)

    Unrecognized field , not marked as ignorable解决办法-CSDN博客

    6.F12怎么看websocket

    切换环境后，websocket接口报错了

    我以前一直以为是环境问题或者是KONG这个组件问题，后来f12看了下，是前端url写的不对，还是原来环境的地址

    7.关于心跳 

    下边的ping pong就是前后端约定好的上传返回 前端用这个来保持心跳

    至于在收到数据之后是否还要保持心跳链接，就看后续处理还要不要用到，不用就可以关，用到就继续连着

     再就是但是每个用户心跳检测时间最好不要一致 会产生高并发

    比如这个用户隔3秒 那个用户隔五秒 弄一个随机数

    8.广播

    这个没有亲测，但是看着比较靠谱。

    之前也写过websocket广播相关博客，但实现方式不同

    思路

    1.用一个线程安全得set集合，放session对象

    2.每次onOpen建立连接时，都把当前得session对象，放到集合中

    3.每次需要发消息时，都遍历这个集合

    4.提供给别人得方法，sendInfo()，不需要入参（看实际业务情况），遍历set，获取session发送消息

    5.onclose方法：接收前端close通知，将指定session从set中移除

    1. @ServerEndpoint(value = "/webSocket")//主要是将目前的类定义成一个websocket服务器端, 注解的值将被用于监听用户连接的终端访问URL地址,客户端可以通过这个URL来连接到WebSocket服务器端
    2. @Component
    3. @EnableScheduling// cron定时任务
    4. @Data
    5. public class WebSocket {
    6.  
    7.     private static final Logger logger = LoggerFactory.getLogger(WebSocket.class);
    8.  
    9.     /**
    10.      * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    11.      */
    12.     private static int onlineCount = 0;
    13.  
    14.     /**
    15.      * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    16.      */
    17.     private static CopyOnWriteArraySet webSocketSet = new CopyOnWriteArraySet<>();
    18.  
    19.     /**
    20.      * 与某个客户端的连接会话，需要通过它来给客户端发送数据
    21.      */
    22.     private Session session;
    23.  
    24.     public static CopyOnWriteArraySet getWebSocketSet() {
    25.         return webSocketSet;
    26.     }
    27.  
    28.     public static void setWebSocketSet(CopyOnWriteArraySet webSocketSet) {
    29.         WebSocket.webSocketSet = webSocketSet;
    30.     }
    31.  
    32.     /**
    33.      * 从数据库查询相关数据信息,可以根据实际业务场景进行修改
    34.      */
    35.     @Resource
    36.     private IndexService indexService;
    37.     private static IndexService indexServiceMapper;
    38.  
    39.     @PostConstruct
    40.     public void init() {
    41.         WebSocket.indexServiceMapper = this.indexService;
    42.     }
    43.  
    44.     /**
    45.      * 连接建立成功调用的方法
    46.      *
    47.      * @param session 会话
    48.      */
    49.     @OnOpen
    50.     public void onOpen(Session session) throws Exception {
    51.         this.session = session;
    52.         webSocketSet.add(this);
    53.         //查询当前在线人数
    54.         int nowOnline = indexServiceMapper.nowOnline();
    55.         this.sendMessage(JSON.toJSONString(nowOnline));
    56.     }
    57.  
    58.     /**
    59.      * 收到客户端消息后调用的方法
    60.      *
    61.      * @param message 客户端发送过来的消息
    62.      */
    63.     @OnMessage
    64.     public void onMessage(String message, Session session) throws IOException {
    65.         logger.info("参数信息：{}", message);
    66.         //群发消息
    67.         for (WebSocket item : webSocketSet) {
    68.             try {
    69.                 item.sendMessage(JSON.toJSONString(message));
    70.             } catch (IOException e) {
    71.                 e.printStackTrace();
    72.             }
    73.         }
    74.     }
    75.  
    76.     /**
    77.      * 连接关闭调用的方法
    78.      */
    79.     @OnClose
    80.     public void onClose(Session session, CloseReason reason) {
    81.         webSocketSet.remove(this);
    82.         if (session != null) {
    83.             try {
    84.                 session.close();
    85.             } catch (IOException e) {
    86.                 e.printStackTrace();
    87.             }
    88.         }
    89.     }
    90.  
    91.     /**
    92.      * 发生错误时调用
    93.      *
    94.      * @param session 会话
    95.      * @param error   错误信息
    96.      */
    97.     @OnError
    98.     public void onError(Session session, Throwable error) {
    99.         logger.error("连接异常！");
    100.         error.printStackTrace();
    101.     }
    102.  
    103.     /**
    104.      * 发送信息
    105.      *
    106.      * @param message 消息
    107.      */
    108.     public void sendMessage(String message) throws IOException {
    109.         this.session.getBasicRemote().sendText(message);
    110.     }
    111.  
    112.     /**
    113.      * 自定义消息推送、可群发、单发
    114.      *
    115.      * @param message 消息
    116.      */
    117.     public static void sendInfo() throws IOException {
    118.         for (WebSocket item : webSocketSet) {
    119.             item.sendMessage(message);
    120.         }
    121.     }
    122. }
    1. @Slf4j
    2. @Component
    3. public class IndexScheduled {
    4.  
    5.     @Autowired
    6.     private IndexMapper indexMapper;
    7.  
    8.     /**
    9.      * 每3秒执行一次
    10.      */
    11.     //@Scheduled(cron = "0/3 * * * * ? ") //我这里暂时不需要运行这条定时任务，所以将注解注释了，朋友们运行时记得放开注释啊
    12.     public void nowOnline() {
    13.         System.err.println("*********   首页定时任务执行   **************");
    14.  
    15.         CopyOnWriteArraySet webSocketSet = WebSocket.getWebSocketSet();
    16.         int nowOnline = indexMapper.nowOnline();
    17.         webSocketSet.forEach(c -> {
    18.             try {
    19.                 c.sendMessage(JSON.toJSONString(nowOnline));
    20.             } catch (IOException e) {
    21.                 e.printStackTrace();
    22.             }
    23.         });
    24.  
    25.         System.err.println("/n 首页定时任务完成.......");
    26.     }
    27.  
    28. }

    9.设计方式

    需求如下：

    跟前后端同事咨询过，一致觉得：

    websocket只返回未读数量，然后点击闹钟之后，另外走一个接口，返回消息列表。

  • 相关阅读:
    基于微信小程序的小说阅读系统（小程序+Nodejs）
    洛谷P5309 Ynoi 2011 初始化 题解
    网络代理技术的威力：保障安全、保护隐私
    Word2Vec的安装与使用
    [C++]AVL树怎么转
    CUDA initialization: The NVIDIA driver on your system is too old解决方案
    C语言数据结构-----单链表(无头单向不循环)
    四数之和 - 力扣中等
    typescript手记
    深入理解Spring注解机制（一）：注解的搜索与处理机制
  • 原文地址：https://blog.csdn.net/qq_35653822/article/details/133794112