devbox
黑客灵魂
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

WebSocket零基础极速上手开发指南_websocket开发

作者：黑客灵魂 | 2024-07-04 14:43:37

赞

踩

websocket开发

一、WebSocket后台配置

1、添加依赖

  1. <dependency>
  2.     <groupId>org.springframework.boot</groupId>
  3.     <artifactId>spring-boot-starter-websocket</artifactId>
  4. </dependency>

2、WebSocket配置类 WebSocketConfig.java

  1. package com.service.websocket.config;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.context.annotation.Bean;
  5. import org.springframework.context.annotation.Configuration;
  6. import org.springframework.web.servlet.config.annotation.CorsRegistry;
  7. import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
  8. import org.springframework.web.socket.server.standard.ServerEndpointExporter;
  9.  
  10. /**
  11.  * Websocket 配置
  12.  */
  13. @Configuration
  14. @Slf4j
  15. public class WebsocketConfig implements WebMvcConfigurer {
  16.  
  17.   //服务器支持跨域
  18.   @Override
  19.   public void addCorsMappings(CorsRegistry registry) {
  20.     registry.addMapping("/**").allowedOrigins("*")
  21.         .allowedMethods("GET", "POST","OPTIONS").allowedHeaders("*")
  22.         .exposedHeaders("Access-Control-Allow-Headers",
  23.             "Access-Control-Allow-Methods",
  24.             "Access-Control-Allow-Origin",
  25.             "Access-Control-Max-Age",
  26.             "X-Frame-Options")
  27.         .allowCredentials(false).maxAge(3600);
  28.   }
  29.  
  30.   /** 注入ServerEndpointExporter，这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint 。
  31.    * 要注意，如果使用独立的servlet容器，而不是直接使用springboot的内置容器，就不要注入ServerEndpointExporter，
  32.    * 因为它将由容器自己提供和管理。在Spring中可以直接使用Java WebSocket API来提供服务，如果使用内置的web容器，需要做的仅仅是需要在下面添加
  33.    * */
  34.   @Bean
  35.   public ServerEndpointExporter serverEndpointExporter() {
  36.     return new ServerEndpointExporter();
  37.   }
  38.  
  39. }

3、WebSocket服务 Server

  1. package com.service.websocket.server;
  2.  
  3. import cn.gooday.jsh.service.common.dto.RestControllerResult;
  4. import java.io.IOException;
  5. import java.nio.ByteBuffer;
  6. import java.text.DateFormat;
  7. import java.text.SimpleDateFormat;
  8. import java.util.Date;
  9. import java.util.List;
  10. import java.util.Map;
  11. import java.util.concurrent.ConcurrentHashMap;
  12. import javax.websocket.EndpointConfig;
  13. import javax.websocket.OnClose;
  14. import javax.websocket.OnError;
  15. import javax.websocket.OnMessage;
  16. import javax.websocket.OnOpen;
  17. import javax.websocket.Session;
  18. import javax.websocket.server.PathParam;
  19. import javax.websocket.server.ServerEndpoint;
  20. import jsh.mg.msg.service.websocket.dto.WebsocketParamsDto;
  21. import jsh.mg.msg.service.websocket.util.HttpUtils;
  22. import jsh.mg.msg.service.websocket.util.RedisUtil;
  23. import lombok.extern.slf4j.Slf4j;
  24. import org.apache.commons.lang3.StringUtils;
  25. import org.slf4j.Logger;
  26. import org.slf4j.LoggerFactory;
  27. import org.springframework.beans.factory.annotation.Autowired;
  28. import org.springframework.stereotype.Component;
  29.  
  30. /**
  31.  * websocket服务器
  32.  */
  33. @Component
  34. @ServerEndpoint("/socket/{serialNumber}")
  35. @Slf4j
  36. public class WebsocketServer {
  37.  
  38.   private static final Logger logger = LoggerFactory.getLogger(WebsocketServer.class);
  39.  
  40.   //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
  41.   private static int onlineCount = 0;
  42.   //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。若要实现服务端与单一客户端通信的话，可以使用Map来存放，其中Key可以为用户标识
  43.   public static ConcurrentHashMap<String, WebsocketServer> webSocketMap =
  44.       new ConcurrentHashMap<String, WebsocketServer>();
  45.   //与某个客户端的连接会话，需要通过它来给客户端发送数据
  46.   private Session session;
  47.   private String callBackUrl;
  48.  
  49.   private static String PREX_STRING = "STRING_";
  50.   private static String PREX_BINARY = "BINARY_";
  51.   private static long EXPIRE_TIME = 604800; //过期时间7天
  52.  
  53.   @Autowired
  54.   private RedisUtil redisUtil;
  55.  
  56.   /**
  57.    * 连接成功后调用的方法
  58.    * @param session  可选的参数。session为与某个客户端的连接会话，需要通过它来给客户端发送数据
  59.    */
  60.   @OnOpen
  61.   public void onOpen(@PathParam(value = "serialNumber") String serialNumber, Session session, EndpointConfig config) {
  62.     this.session = session;
  63.  
  64.     String queryString = session.getQueryString();
  65.     log.info("session.getQueryString()" + queryString);
  66.     if (StringUtils.isNotBlank(queryString) && queryString.contains("url=")) {
  67.       this.callBackUrl = queryString.replace("url=", "");
  68.     }
  69.     webSocketMap.put(serialNumber, this) ;
  70.  
  71.     //在线数加1
  72.     addOnlineCount();
  73.     log.info("当前有连接" + serialNumber + "加入！当前在线人数为" + getOnlineCount());
  74.   }
  75.  
  76.   /**
  77.    * 连接关闭调用方法
  78.    */
  79.   @OnClose
  80.   public void onClose(@PathParam(value = "serialNumber") String serialNumber) {
  81.     if (StringUtils.isNotBlank(serialNumber)) {
  82.       log.info("WebsocketServer.onClose() is begin, serialNumber is " + serialNumber);
  83.       webSocketMap.remove(serialNumber);
  84.       //在线数减1
  85.       subOnlineCount();
  86.     }
  87.   }
  88.  
  89.   /**
  90.    * 连接异常
  91.    */
  92.   @OnError
  93.   public void onError(@PathParam(value = "serialNumber") String serialNumber, Throwable error) {
  94.     log.info("连接异常: serialNumber is " + serialNumber + ",error is " + error.getMessage());
  95.   }
  96.  
  97.   /**
  98.    * 接收到客户短消息
  99.    */
  100.   @OnMessage
  101.   public void onMessage(@PathParam(value = "serialNumber") String serialNumber, String message) {
  102.     try {
  103.       if (!isOnline(serialNumber)) {
  104.         return ;
  105.       }
  106.       WebsocketServer websocketServer = webSocketMap.get(serialNumber) ;
  107.       HttpUtils.sendPost(websocketServer.callBackUrl, message);
  108.     } catch (Exception e) {
  109.       log.info(e.getMessage());
  110.       return ;
  111.     }
  112.  
  113.     return ;
  114.   }
  115.  
  116.   public RestControllerResult<Boolean> checkExist(String serialNumber) {
  117.     RestControllerResult<Boolean> result =  new RestControllerResult<Boolean>();
  118.     if (!isExist(serialNumber)) {
  119.       result.setData(false);
  120.       result.setErrorMsg("请求的链接:" + serialNumber + "不在该服务器上");
  121.       return result;
  122.     }
  123.     result.setData(true);
  124.     return result;
  125.   }
  126.  
  127.   public RestControllerResult<Boolean> checkOnline(String serialNumber) {
  128.     RestControllerResult<Boolean> result =  new RestControllerResult<Boolean>();
  129.     if (!isOnline(serialNumber)) {
  130.       result.setData(false);
  131.       result.setErrorMsg("请求的链接:" + serialNumber + "不在线");
  132.       return result;
  133.     }
  134.     result.setData(true);
  135.     return result;
  136.   }
  137.  
  138.  
  139.   public RestControllerResult<Boolean> checkOfflineMessage(String serialNumber) {
  140.     RestControllerResult<Boolean> result =  new RestControllerResult<Boolean>();
  141.  
  142.     if (!isOnline(serialNumber)) {
  143.       result.setData(false);
  144.       result.setErrorMsg("请求的链接:" + serialNumber + "不在线");
  145.       return result;
  146.     }
  147.     List<String> messageList = (List<String>)redisUtil.get(PREX_STRING + serialNumber);
  148.     List<ByteBuffer> binaryList = (List<ByteBuffer>)redisUtil.get(PREX_BINARY + serialNumber);
  149.  
  150.     if ((null == messageList || messageList.size() < 1) && (null == binaryList || binaryList.size() < 1)) {
  151.       result.setData(false);
  152.     } else {
  153.       result.setData(true);
  154.     }
  155.     return result;
  156.   }
  157.  
  158.   public RestControllerResult<Boolean> pullOfflineMessage(String serialNumber, Boolean isAsync, Boolean flag) {
  159.     RestControllerResult<Boolean> result =  new RestControllerResult<Boolean>();
  160.     try {
  161.       if (!isOnline(serialNumber)) {
  162.         result.setData(false);
  163.         result.setErrorMsg("请求的链接:" + serialNumber + "不在线");
  164.         return result;
  165.       }
  166.       WebsocketServer websocketServer = webSocketMap.get(serialNumber) ;
  167.  
  168.       List<String> messageList = (List<String>)redisUtil.get(PREX_STRING + serialNumber);
  169.       if (null != messageList && messageList.size() > 0 ) {
  170.         Boolean isSuccess = false;
  171.         if (!isAsync) {
  172.           isSuccess = sendBasicMessage(websocketServer.session, messageList, flag);
  173.         } else {
  174.           isSuccess = sendAsyncMessage(websocketServer.session, messageList);
  175.         }
  176.         if (isSuccess) {
  177.           redisUtil.delete(PREX_STRING + serialNumber);
  178.         }
  179.       }
  180.  
  181.       List<ByteBuffer> binaryList = (List<ByteBuffer>)redisUtil.get(PREX_BINARY + serialNumber);
  182.       if (null != binaryList && binaryList.size() > 0 ) {
  183.         Boolean isSuccess = false;
  184.         if (!isAsync) {
  185.           isSuccess = sendBasicBinaryMessage(websocketServer.session, binaryList, flag);
  186.         } else {
  187.           isSuccess = sendAsyncBinaryMessage(websocketServer.session, binaryList);
  188.         }
  189.         if (isSuccess) {
  190.           redisUtil.delete(PREX_BINARY + serialNumber);
  191.         }
  192.       }
  193.     } catch (IOException e) {
  194.       log.info(e.getMessage());
  195.       result.setData(false);
  196.       result.setErrorMsg(e.getMessage());
  197.       return result;
  198.     }
  199.     return result;
  200.   }
  201.  
  202.   /**
  203.    * 给所有人发消息
  204.    */
  205.   public RestControllerResult<Boolean>  sendToAll(WebsocketParamsDto paramsDto) {
  206.     RestControllerResult<Boolean> result =  new RestControllerResult<Boolean>();
  207.     //遍历HashMap
  208.     for (String serialNumber : webSocketMap.keySet()) {
  209.       try {
  210.         if (paramsDto.getIsBinary()) {
  211.           sendBasicBinaryMessage(webSocketMap.get(serialNumber).session, paramsDto.getBinaryData(), paramsDto.getFlag());
  212.         } else {
  213.           sendBasicMessage(webSocketMap.get(serialNumber).session, paramsDto.getMessage(), paramsDto.getFlag());
  214.         }
  215.       } catch (IOException e) {
  216.         log.info(e.getMessage());
  217.  
  218.         if (paramsDto.getIsBinary()) {
  219.           redisUtil.addToListRightExpire(PREX_STRING + serialNumber, EXPIRE_TIME, paramsDto.getMessage());
  220.         } else {
  221.           redisUtil.addToListRightExpire(PREX_BINARY + serialNumber, EXPIRE_TIME, paramsDto.getBinaryData());
  222.         }
  223.       }
  224.     }
  225.  
  226.     result.setData(true);
  227.     return result;
  228.   }
  229.  
  230.  
  231.  
  232.   /**
  233.    * 给指定的终端发送消息
  234.    */
  235.   public RestControllerResult<Boolean> sendToTerminal(WebsocketParamsDto paramsDto) {
  236.     log.info("WebsocketServer.sendToSingle() is begin, serialNumber is " + paramsDto.getSerialNumber());
  237.     RestControllerResult<Boolean> result =  new RestControllerResult<Boolean>();
  238.     String toSerialNumber = paramsDto.getSerialNumber();
  239.     try {
  240.       if(!isOnline(toSerialNumber)) {
  241.         result.setData(false);
  242.         result.setErrorMsg("请求的链接:" + toSerialNumber + "不在该服务器上");
  243.         log.info("请求的链接:" + toSerialNumber + "不在该服务器上");
  244.         return result;
  245.       }
  246.  
  247.       WebsocketServer websocketServer = webSocketMap.get(toSerialNumber) ;
  248.       if (paramsDto.getIsBinary()) {
  249.         sendBasicBinaryMessage(websocketServer.session, paramsDto.getBinaryData(), paramsDto.getFlag());
  250.       } else {
  251.         sendBasicMessage(websocketServer.session, paramsDto.getMessage(), paramsDto.getFlag());
  252.       }
  253.       result.setData(true);
  254.     } catch (IOException e) {
  255.       log.info(e.getMessage());
  256.  
  257.       if (paramsDto.getIsBinary()) {
  258.         redisUtil.addToListRightExpire(PREX_STRING + toSerialNumber, EXPIRE_TIME, paramsDto.getMessage());
  259.       } else {
  260.         redisUtil.addToListRightExpire(PREX_BINARY + toSerialNumber, EXPIRE_TIME, paramsDto.getBinaryData());
  261.       }
  262.       result.setData(false);
  263.       result.setErrorMsg("推送消息至链接:" + toSerialNumber + "时系统异常：" + e.getMessage());
  264.       return result;
  265.     }
  266.     return result;
  267.   }
  268.  
  269.   /**
  270.    * 给所有人发消息
  271.    */
  272.   public RestControllerResult<Boolean> sendToTerminalList(WebsocketParamsDto paramsDto) {
  273.     RestControllerResult<Boolean> result =  new RestControllerResult<Boolean>();
  274.       List<String> terminalList = paramsDto.getSerialList();
  275.       //遍历HashMap
  276.       for (String serialNumber : terminalList) {
  277.         try {
  278.           if (paramsDto.getIsBinary()) {
  279.             sendBasicBinaryMessage(webSocketMap.get(serialNumber).session, paramsDto.getBinaryData(), paramsDto.getFlag());
  280.           } else {
  281.             sendBasicMessage(webSocketMap.get(serialNumber).session, paramsDto.getMessage(), paramsDto.getFlag());
  282.           }
  283.         } catch (IOException e) {
  284.           log.info(e.getMessage());
  285.  
  286.           if (paramsDto.getIsBinary()) {
  287.             redisUtil.addToListRightExpire(PREX_STRING + serialNumber, EXPIRE_TIME, paramsDto.getMessage());
  288.           } else {
  289.             redisUtil.addToListRightExpire(PREX_BINARY + serialNumber, EXPIRE_TIME, paramsDto.getBinaryData());
  290.           }
  291.         }
  292.       }
  293.     result.setData(true);
  294.     return result;
  295.   }
  296.  
  297.   /**
  298.    * 给指定的人发送消息
  299.    */
  300.   public RestControllerResult<Boolean> sendToTerminalAsync(WebsocketParamsDto paramsDto) {
  301.     RestControllerResult<Boolean> result =  new RestControllerResult<Boolean>();
  302.     String toSerialNumber = paramsDto.getSerialNumber();
  303.     try {
  304.       if(!isOnline(toSerialNumber)) {
  305.         result.setData(false);
  306.         result.setErrorMsg("请求的链接:" + toSerialNumber + "不在该服务器上");
  307.         log.info("请求的链接:" + toSerialNumber + "不在该服务器上");
  308.         return result;
  309.       }
  310.  
  311.       WebsocketServer websocketServer = webSocketMap.get(toSerialNumber) ;
  312.       if (paramsDto.getIsBinary()) {
  313.         sendAsyncBinaryMessage(websocketServer.session, paramsDto.getBinaryData());
  314.       } else {
  315.         sendAsyncMessage(websocketServer.session, paramsDto.getMessage());
  316.       }
  317.       result.setData(true);
  318.     } catch (IOException e) {
  319.       log.info(e.getMessage());
  320.  
  321.       if (paramsDto.getIsBinary()) {
  322.         redisUtil.addToListRightExpire(PREX_STRING + toSerialNumber, EXPIRE_TIME, paramsDto.getMessage());
  323.       } else {
  324.         redisUtil.addToListRightExpire(PREX_BINARY + toSerialNumber, EXPIRE_TIME, paramsDto.getBinaryData());
  325.       }
  326.  
  327.       result.setData(false);
  328.       result.setErrorMsg("推送消息至链接:" + toSerialNumber + "时系统异常：" + e.getMessage());
  329.       return result;
  330.     }
  331.     return result;
  332.   }
  333.  
  334.   /**
  335.    * 给所有人发消息
  336.    */
  337.   public RestControllerResult<Boolean> sendToTerminalListAsync(WebsocketParamsDto paramsDto) {
  338.     RestControllerResult<Boolean> result =  new RestControllerResult<Boolean>();
  339.     List<String> terminalList = paramsDto.getSerialList();
  340.     for (String serialNumber : terminalList) {
  341.       try {
  342.         if (paramsDto.getIsBinary()) {
  343.           sendAsyncBinaryMessage(webSocketMap.get(serialNumber).session, paramsDto.getBinaryData());
  344.         } else {
  345.           sendAsyncMessage(webSocketMap.get(serialNumber).session, paramsDto.getMessage());
  346.         }
  347.       } catch (IOException e) {
  348.         log.info(e.getMessage());
  349.  
  350.         if (paramsDto.getIsBinary()) {
  351.           redisUtil.addToListRightExpire(PREX_STRING + serialNumber, EXPIRE_TIME, paramsDto.getMessage());
  352.         } else {
  353.           redisUtil.addToListRightExpire(PREX_BINARY + serialNumber, EXPIRE_TIME, paramsDto.getBinaryData());
  354.         }
  355.       }
  356.     }
  357.     result.setData(true);
  358.     return result;
  359.   }
  360.  
  361.   /**
  362.    * 给所有人发消息
  363.    */
  364.   public RestControllerResult<Boolean>  sendToAllAsync(WebsocketParamsDto paramsDto) {
  365.     RestControllerResult<Boolean> result =  new RestControllerResult<Boolean>();
  366.     for (String serialNumber : webSocketMap.keySet()) {
  367.       try {
  368.         if (paramsDto.getIsBinary()) {
  369.           sendAsyncBinaryMessage(webSocketMap.get(serialNumber).session, paramsDto.getBinaryData());
  370.         } else {
  371.           sendAsyncMessage(webSocketMap.get(serialNumber).session, paramsDto.getMessage());
  372.         }
  373.       } catch (IOException e) {
  374.         log.info(e.getMessage());
  375.         if (paramsDto.getIsBinary()) {
  376.           redisUtil.addToListRightExpire(PREX_STRING + serialNumber, EXPIRE_TIME, paramsDto.getMessage());
  377.         } else {
  378.           redisUtil.addToListRightExpire(PREX_BINARY + serialNumber, EXPIRE_TIME, paramsDto.getBinaryData());
  379.         }
  380.       }
  381.     }
  382.     result.setData(true);
  383.     logger.info("给所有人发消息结果: {}", result);
  384.     return result;
  385.   }
  386.  
  387.   public static Map<String, WebsocketServer> getWebSocketMap() {
  388.     return webSocketMap ;
  389.   }
  390.  
  391.   /**
  392.    * 同步发送消息模式。
  393.    * message: 待发送的消息
  394.    * flag: 是否支持发送部分消息（true:可部分发送消息; false:一次性发布全部消息）
  395.    */
  396.   public void sendBasicMessage(Session session, String message, Boolean flag) throws IOException {
  397.     log.info("WebsocketServer.sendBasicMessage() is begin, flag is " + flag + " ,message is " + message);
  398.     session.getBasicRemote().sendText(message, flag);
  399.     log.info("WebsocketServer.sendBasicMessage() is end. ");
  400.   }
  401.  
  402.   /**
  403.    * 同步发送消息模式。
  404.    * message: 待发送的消息
  405.    * flag: 是否支持发送部分消息（true:可部分发送消息; false:一次性发布全部消息）
  406.    */
  407.   public void sendBasicBinaryMessage(Session session, ByteBuffer message, Boolean flag) throws IOException {
  408.     log.info("WebsocketServer.sendBasicMessage() is begin, flag is " + flag + " ,message is " + message);
  409.     session.getBasicRemote().sendBinary(message, flag);
  410.     log.info("WebsocketServer.sendBasicMessage() is end. ");
  411.   }
  412.  
  413.   /**
  414.    * 异步发送消息模式。
  415.    * message: 待发送的消息
  416.    * flag: 是否支持发送部分消息
  417.    */
  418.   public void sendAsyncMessage(Session session, String message) throws IOException {
  419.     log.info("WebsocketServer.sendAsyncMessage() is begin, message is " + message);
  420.     session.getAsyncRemote().sendText(message);
  421.     log.info("WebsocketServer.sendAsyncMessage() is end. ");
  422.  
  423.   }
  424.  
  425.   /**
  426.    * 异步发送消息模式。
  427.    * message: 待发送的消息
  428.    * flag: 是否支持发送部分消息
  429.    */
  430.   public void sendAsyncBinaryMessage(Session session, ByteBuffer message) throws IOException {
  431.     log.info("WebsocketServer.sendAsyncMessage() is begin, message is " + message);
  432.     session.getAsyncRemote().sendBinary(message);
  433.     log.info("WebsocketServer.sendAsyncMessage() is end. ");
  434.  
  435.   }
  436.  
  437.   /**
  438.    * 同步发送消息模式。
  439.    * message: 待发送的消息
  440.    * flag: 是否支持发送部分消息（true:可部分发送消息; false:一次性发布全部消息）
  441.    */
  442.   public Boolean sendBasicMessage(Session session, List<String> messageList, Boolean flag) throws IOException {
  443.     if (null == messageList || messageList.size() < 1) {
  444.       return true;
  445.     }
  446.     for (int i = 0; i < messageList.size(); i++) {
  447.       session.getBasicRemote().sendText(messageList.get(i), flag);
  448.     }
  449.     return true;
  450.   }
  451.  
  452.   /**
  453.    * 异步发送消息模式。
  454.    * message: 待发送的消息
  455.    * flag: 是否支持发送部分消息
  456.    */
  457.   public Boolean sendAsyncMessage(Session session, List<String> messageList) throws IOException {
  458.     if (null == messageList || messageList.size() < 1) {
  459.       return true;
  460.     }
  461.     for (int i = 0; i < messageList.size(); i++) {
  462.       session.getAsyncRemote().sendText(messageList.get(i));
  463.     }
  464.     return true;
  465.   }
  466.  
  467.   /**
  468.    * 同步发送消息模式。
  469.    * message: 待发送的消息
  470.    * flag: 是否支持发送部分消息（true:可部分发送消息; false:一次性发布全部消息）
  471.    */
  472.   public Boolean sendBasicBinaryMessage(Session session, List<ByteBuffer> messageList, Boolean flag) throws IOException {
  473.     if (null == messageList || messageList.size() < 1) {
  474.       return true;
  475.     }
  476.     for (int i = 0; i < messageList.size(); i++) {
  477.       session.getBasicRemote().sendBinary(messageList.get(i), flag);
  478.     }
  479.     return true;
  480.   }
  481.  
  482.   /**
  483.    * 异步发送消息模式。
  484.    * message: 待发送的消息
  485.    * flag: 是否支持发送部分消息
  486.    */
  487.   public Boolean sendAsyncBinaryMessage(Session session, List<ByteBuffer> messageList) throws IOException {
  488.     if (null == messageList || messageList.size() < 1) {
  489.       return true;
  490.     }
  491.     for (int i = 0; i < messageList.size(); i++) {
  492.       session.getAsyncRemote().sendBinary(messageList.get(i));
  493.     }
  494.     return true;
  495.   }
  496.  
  497.   /**
  498.    * 获取当前时间
  499.    *
  500.    * @return
  501.    */
  502.   private String getNowTime() {
  503.     Date date = new Date();
  504.     DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  505.     String time = format.format(date);
  506.     return time;
  507.   }
  508.  
  509.   public static synchronized int getOnlineCount() {
  510.     return onlineCount;
  511.   }
  512.  
  513.   public static synchronized void addOnlineCount() {
  514.     WebsocketServer.onlineCount++;
  515.   }
  516.  
  517.   public static synchronized void subOnlineCount() {
  518.     WebsocketServer.onlineCount--;
  519.   }
  520.  
  521.   public Boolean isExist(String serialNumber) {
  522.     if(StringUtils.isBlank(serialNumber) || !webSocketMap.containsKey(serialNumber)){
  523.       return false;
  524.     }
  525.     return true;
  526.   }
  527.  
  528.   public Boolean isOnline(String serialNumber) {
  529.     try {
  530.       if(StringUtils.isBlank(serialNumber) || !webSocketMap.containsKey(serialNumber)){
  531.         return false;
  532.       }
  533.       WebsocketServer websocketServer = webSocketMap.get(serialNumber) ;
  534.       if (websocketServer == null ) {
  535.         return false;
  536.       }
  537.     } catch (Exception e){
  538.       log.error(e.getMessage());
  539.       return false;
  540.     }
  541.     return true;
  542.   }
  543. }

4、入参对象配置

  1. package com.service.websocket.dto;
  2.  
  3. import io.swagger.annotations.ApiModel;
  4. import io.swagger.annotations.ApiModelProperty;
  5. import java.nio.ByteBuffer;
  6. import java.util.List;
  7. import lombok.Data;
  8. import org.w3c.dom.Text;
  9.  
  10. /**
  11.  * Websocket参数 DTO.
  12.  *
  13.  * @author ****
  14.  */
  15. @Data
  16. @ApiModel(value = "WebsocketParamsDto", description = "test")
  17. public class WebsocketParamsDto {
  18.  
  19.   @ApiModelProperty(value = "序列号", name = "serialNumber", example = "key.12345")
  20.   private String serialNumber;
  21.  
  22.   @ApiModelProperty(value = "键值", name = "serialList", example = "key.12345")
  23.   private List<String> serialList;
  24.  
  25.   @ApiModelProperty(value = "信息类型:true 异步；false 同步", name = "dataType", example = "1/空:Sring;2-binary;3-text;")
  26.   private Boolean isBinary;
  27.  
  28.   @ApiModelProperty(value = "信息", name = "message", example = "待发送信息")
  29.   private String message;
  30.  
  31.   @ApiModelProperty(value = "二进制数据", name = "binaryData", example = "01010101....")
  32.   private ByteBuffer binaryData;
  33.  
  34.   @ApiModelProperty(value = "Text数据", name = "textData", example = "key.67890")
  35.   private Text textData;
  36.  
  37.   @ApiModelProperty(value = "模式", name = "同步false/异步模式true", example = "true")
  38.   private Boolean isAsync;
  39.  
  40.   @ApiModelProperty(value = "传送标识", name = "flag", example = "true--一次性发送全部信息；false--可部分发送信息；")
  41.   private Boolean flag;
  42. }

5、业务调用

  1. /**
  2.    * 给所有人发消息
  3.    */
  4.   public RestControllerResult<Boolean>  sendToAll(WebsocketParamsDto paramsDto) {
  5.     RestControllerResult<Boolean> result =  new RestControllerResult<Boolean>();
  6.     //遍历HashMap
  7.     for (String serialNumber : webSocketMap.keySet()) {
  8.       try {
  9.         if (paramsDto.getIsBinary()) {
  10.           sendBasicBinaryMessage(webSocketMap.get(serialNumber).session, paramsDto.getBinaryData(), paramsDto.getFlag());
  11.         } else {
  12.           sendBasicMessage(webSocketMap.get(serialNumber).session, paramsDto.getMessage(), paramsDto.getFlag());
  13.         }
  14.       } catch (IOException e) {
  15.         log.info(e.getMessage());
  16.  
  17.         if (paramsDto.getIsBinary()) {
  18.           redisUtil.addToListRightExpire(PREX_STRING + serialNumber, EXPIRE_TIME, paramsDto.getMessage());
  19.         } else {
  20.           redisUtil.addToListRightExpire(PREX_BINARY + serialNumber, EXPIRE_TIME, paramsDto.getBinaryData());
  21.         }
  22.       }
  23.     }
  24.  
  25.     result.setData(true);
  26.     return result;
  27.   }
  1. @Resource
  2. private WebsocketServer webSocketServer;
  3.  
  4. @Override
  5. public RestControllerResult<Boolean>
  6.   sendToAll(@RequestBody WebsocketParamsDto paramsDto) {
  7.   return webSocketServer.sendToAll(paramsDto);
  8. }

二、Websocket 测试

1、在线测试

测试URL：Websocket在线测试-Websocket接口测试-Websocket模拟请求工具 (jsons.cn)

2、业务调用 WebSocket 服务，发出消息

通过 postman 测试工具调用接口，或者其他工具

3、Websocket 监听端回应情况如下 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/黑客灵魂/article/detail/787263
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号