devbox
我家小花儿
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Websocket客户端实现平台数据-后端-前端数据传输 wss协议_websocketclient.setsocketfactory

作者：我家小花儿 | 2024-02-29 08:41:23

赞

踩

websocketclient.setsocketfactory
WebSocketClientAbs
  1.  
  2. import org.apache.http.conn.ssl.TrustSelfSignedStrategy;
  3. import org.apache.http.ssl.SSLContexts;
  4. import org.java_websocket.client.WebSocketClient;
  5. import org.springframework.beans.factory.annotation.Value;
  6. import org.springframework.core.io.DefaultResourceLoader;
  7. import org.springframework.core.io.ResourceLoader;
  8.  
  9. import javax.net.ssl.SSLContext;
  10. import javax.net.ssl.SSLSocketFactory;
  11. import java.io.IOException;
  12. import java.io.InputStream;
  13. import java.security.KeyStore;
  14. import java.util.Map;
  15.  
  16. public abstract class WebSocketClientAbs {
  17.  
  18.  
  19.     @Value("${server.ssl.key-store}")
  20.     private String cerPath;
  21.  
  22.     @Value("${server.ssl.key-store-password}")
  23.     private String cerPwd;
  24.  
  25.     /**
  26.      * 创建WebSocket客户端
  27.      *
  28.      * @param wsUri
  29.      * @param httpHeaders
  30.      * @return
  31.      */
  32.     public abstract WebSocketClient createWebSocketClient(String wsUri, Map<String, String> httpHeaders);
  33.  
  34.     /**
  35.      * 客户端连接
  36.      *
  37.      * @param uri
  38.      * @param httpHeaders
  39.      * @return
  40.      */
  41.     public abstract WebSocketClient connect(String uri, Map<String, String> httpHeaders);
  42.  
  43.     /**
  44.      * wss协议证书认证
  45.      *
  46.      * @param webSocketClient
  47.      */
  48.     public void createWebSocketClient(WebSocketClient webSocketClient) {
  49.         try {
  50.             KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
  51.             keyStore.load(resourceLoader(cerPath), cerPwd.toCharArray());
  52.             SSLContext sslContext = SSLContexts.custom()
  53.                     .loadTrustMaterial(keyStore, new TrustSelfSignedStrategy()).build();
  54.             SSLSocketFactory sslFactory = sslContext.getSocketFactory();
  55.             webSocketClient.setSocketFactory(sslFactory);
  56.         } catch (Exception e) {
  57.             e.printStackTrace();
  58.         }
  59.     }
  60.  
  61.  
  62.     /**
  63.      * 读取文件信息
  64.      *
  65.      * @param fileFullPath
  66.      * @return
  67.      * @throws IOException
  68.      */
  69.     public InputStream resourceLoader(String fileFullPath) throws IOException {
  70.         ResourceLoader resourceLoader = new DefaultResourceLoader();
  71.         return resourceLoader.getResource(fileFullPath).getInputStream();
  72.     }
  73.  
  74. }
PlatformWebsocketClient：平台客户端
  1.  
  2. import lombok.Data;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.java_websocket.client.WebSocketClient;
  5. import org.java_websocket.handshake.ServerHandshake;
  6. import org.springframework.stereotype.Component;
  7.  
  8. import javax.annotation.Resource;
  9. import java.net.URI;
  10. import java.net.URISyntaxException;
  11. import java.util.Map;
  12.  
  13.  
  14. @Component
  15. @Slf4j
  16. @Data
  17. public class PlatformWebsocketClient extends WebSocketClientAbs {
  18.  
  19.  
  20.     @Resource
  21.     private MessageService messageService;
  22.     private WebSocketClient wsClient;
  23.  
  24.     // 消息类型
  25.     private String type;
  26.  
  27.     // 0：链接断开或者异常；1：代表链接中；2：代表正在连接；
  28.     public static int isConnect = 0;
  29.  
  30.     /**
  31.      * 获取客户端连接实例
  32.      *
  33.      * @param wsUri
  34.      * @param httpHeaders
  35.      * @return
  36.      */
  37.     @Override
  38.     public WebSocketClient createWebSocketClient(String wsUri, Map<String, String> httpHeaders) {
  39.  
  40.         try {
  41.             //创建客户端连接对象
  42.             return new WebSocketClient(new URI(wsUri), httpHeaders) {
  43.  
  44.                 /**
  45.                  * 建立连接调用
  46.                  * @param serverHandshake
  47.                  */
  48.                 @Override
  49.                 public void onOpen(ServerHandshake serverHandshake) {
  50.                     isConnect = 1;
  51.                 }
  52.  
  53.                 /**
  54.                  * 收到服务端消息调用
  55.                  * @param s
  56.                  */
  57.                 @Override
  58.                 public void onMessage(String s) {
  59. //                    log.info("WebsocketClient-> 收到服务端消息：{}", s);
  60.                     messageService.handleMessage(s);
  61.                 }
  62.  
  63.                 /**
  64.                  * 断开连接调用
  65.                  * @param i
  66.                  * @param s
  67.                  * @param b
  68.                  */
  69.                 @Override
  70.                 public void onClose(int i, String s, boolean b) {
  71.                     isConnect = 0;
  72.                 }
  73.  
  74.                 /**
  75.                  * 连接报错调用
  76.                  * @param e
  77.                  */
  78.                 @Override
  79.                 public void onError(Exception e) {
  80.                     if (null != wsClient) {
  81.                         wsClient.close();
  82.                     }
  83.                     isConnect = 0;
  84.                 }
  85.             };
  86.         } catch (URISyntaxException e) {
  87.             e.printStackTrace();
  88.         }
  89.         return null;
  90.     }
  91.  
  92.     /**
  93.      * 连接websocket服务端
  94.      * 注意 synchronized 关键字，保证多个请求同时连接时，
  95.      * 只有一个连接在创建
  96.      *
  97.      * @param uri
  98.      * @param httpHeaders
  99.      * @return
  100.      */
  101.     @Override
  102.     public synchronized WebSocketClient connect(String uri, Map<String, String> httpHeaders) {
  103.         WebSocketClient oldWsClient = this.getWsClient();
  104.         if (null != oldWsClient) {
  105.             log.info("WebsocketClient -> 已存在连接，oldWsClient：{}-{}",
  106.                     oldWsClient.getReadyState(), oldWsClient.getReadyState().ordinal());
  107.             if (1 == oldWsClient.getReadyState().ordinal()) {
  108.                 log.info("WebsocketClient -> 使用存在且已打开的连接");
  109.                 return oldWsClient;
  110.             } else {
  111.                 log.info("WebsocketClient -> 注销存在且未打开的连接，并重新获取新的连接");
  112.                 oldWsClient.close();
  113.             }
  114.         }
  115.  
  116.         WebSocketClient newWsClient = createWebSocketClient(uri, httpHeaders);
  117.         // 如果是 "wss" 协议，则进行证书认证，认证方法在父类中
  118.         if (uri.startsWith("wss")) {
  119.             createWebSocketClient(newWsClient);
  120.         }
  121.         if (null == newWsClient) {
  122.             log.error("创建失败");
  123.         }
  124.         assert newWsClient != null;
  125.         newWsClient.connect();
  126.         // 设置连接状态为正在连接
  127.         isConnect = 2;
  128.         // 连接状态不再是0请求中，判断建立结果是不是1已建立
  129.         long startTime = System.currentTimeMillis();
  130.         while (1 != newWsClient.getReadyState().ordinal()) {
  131.             // 避免网络波动，设置持续等待连接时间
  132.             long endTime = System.currentTimeMillis();
  133.             long waitTime = (endTime - startTime) / 1000;
  134.             if (5L < waitTime) {
  135. //                log.info("WebsocketClient -> 建立连接异常，请稍后再试");
  136.                 break;
  137.             }
  138.         }
  139.         if (1 == newWsClient.getReadyState().ordinal()) {
  140.             this.setWsClient(newWsClient);
  141. //            newWsClient.send("WebsocketClient -> 服务端连接成功！");
  142. //            log.info("WebsocketClient -> 服务端连接成功！");
  143.             return newWsClient;
  144.         }
  145.         return null;
  146.     }
  147.  
  148. }
LocalWebsocketClient：本地客户端
  1.  
  2. import lombok.Data;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.java_websocket.client.WebSocketClient;
  5. import org.java_websocket.handshake.ServerHandshake;
  6. import org.springframework.stereotype.Component;
  7.  
  8. import java.net.URI;
  9. import java.net.URISyntaxException;
  10. import java.util.Map;
  11.  
  12.  
  13. @Component
  14. @Slf4j
  15. @Data
  16. public class LocalWebsocketClient extends WebSocketClientAbs {
  17.  
  18.  
  19.     private WebSocketClient wsClient;
  20.  
  21.  
  22.     public static int isConnect = 0;
  23.  
  24.     /**
  25.      * 获取客户端连接实例
  26.      *
  27.      * @param wsUri
  28.      * @param httpHeaders
  29.      * @return
  30.      */
  31.     @Override
  32.     public WebSocketClient createWebSocketClient(String wsUri, Map<String, String> httpHeaders) {
  33.  
  34.         try {
  35.             //创建客户端连接对象
  36.             WebSocketClient client = new WebSocketClient(new URI(wsUri), httpHeaders) {
  37.  
  38.                 /**
  39.                  * 建立连接调用
  40.                  * @param serverHandshake
  41.                  */
  42.                 @Override
  43.                 public void onOpen(ServerHandshake serverHandshake) {
  44.                     isConnect = 1;
  45.                 }
  46.  
  47.                 /**
  48.                  * 收到服务端消息调用
  49.                  * @param s
  50.                  */
  51.                 @Override
  52.                 public void onMessage(String s) {
  53.                     log.info("WebsocketClient-> 收到服务端消息：{}", s);
  54.                 }
  55.  
  56.                 /**
  57.                  * 断开连接调用
  58.                  * @param i
  59.                  * @param s
  60.                  * @param b
  61.                  */
  62.                 @Override
  63.                 public void onClose(int i, String s, boolean b) {
  64.                     isConnect = 0;
  65.                 }
  66.  
  67.                 /**
  68.                  * 连接报错调用
  69.                  * @param e
  70.                  */
  71.                 @Override
  72.                 public void onError(Exception e) {
  73.                     if (null != wsClient) {
  74.                         wsClient.close();
  75.                     }
  76.                     isConnect = 0;
  77.                 }
  78.             };
  79.             return client;
  80.         } catch (URISyntaxException e) {
  81.             e.printStackTrace();
  82.         }
  83.         return null;
  84.     }
  85.  
  86.     /**
  87.      * 连接websocket服务端
  88.      * 注意 synchronized 关键字，保证多个请求同时连接时，
  89.      * 只有一个连接在创建
  90.      *
  91.      * @param uri
  92.      * @param httpHeaders
  93.      * @return
  94.      */
  95.     @Override
  96.     public synchronized WebSocketClient connect(String uri, Map<String, String> httpHeaders) {
  97.         WebSocketClient oldWsClient = this.getWsClient();
  98.         if (null != oldWsClient) {
  99.             log.info("WebsocketClient -> 已存在连接，oldWsClient：{}-{}",
  100.                     oldWsClient.getReadyState(), oldWsClient.getReadyState().ordinal());
  101.             if (1 == oldWsClient.getReadyState().ordinal()) {
  102.                 log.info("WebsocketClient -> 使用存在且已打开的连接");
  103.                 return oldWsClient;
  104.             } else {
  105.                 log.info("WebsocketClient -> 注销存在且未打开的连接，并重新获取新的连接");
  106.                 oldWsClient.close();
  107.             }
  108.         }
  109.  
  110.         WebSocketClient newWsClient = createWebSocketClient(uri, httpHeaders);
  111.         // 如果是 "wss" 协议，则进行证书认证，认证方法在父类中
  112.         if (uri.startsWith("wss")) {
  113.             createWebSocketClient(newWsClient);
  114.         }
  115.         if (null == newWsClient) {
  116.             log.error("创建失败");
  117.         }
  118.         assert newWsClient != null;
  119.         newWsClient.connect();
  120.         // 设置连接状态为正在连接
  121.         isConnect = 2;
  122.         // 连接状态不再是0请求中，判断建立结果是不是1已建立
  123.         long startTime = System.currentTimeMillis();
  124.         while (1 != newWsClient.getReadyState().ordinal()) {
  125.             // 避免网络波动，设置持续等待连接时间
  126.             long endTime = System.currentTimeMillis();
  127.             long waitTime = (endTime - startTime) / 1000;
  128.             if (5L < waitTime) {
  129.                 break;
  130.             }
  131.         }
  132.         if (1 == newWsClient.getReadyState().ordinal()) {
  133.             this.setWsClient(newWsClient);
  134. //            newWsClient.send("WebsocketClient -> 服务端连接成功！");
  135.             return newWsClient;
  136.         }
  137.         return null;
  138.     }
  139.  
  140. }
WebSocketHeartbeatTimer：心跳检测
  1.  
  2. import javax.annotation.Resource;
  3. import java.time.LocalDateTime;
  4. import java.time.format.DateTimeFormatter;
  5.  
  6.  
  7. /**
  8.  * 心跳重连机制
  9.  */
  10. @Component
  11. @Slf4j
  12. public class WebSocketHeartbeatTimer {
  13.  
  14.     @Resource
  15.     private PlatformWebsocketClient platformWebsocketClient;
  16.  
  17.     @Resource
  18.     private LocalWebsocketClient localWebsocketClient;
  19.  
  20.  
  21.     @Value("${websocket.url}")
  22.     private String platformWebSocketUrl;
  23.  
  24.  
  25.     @Value("${localWebsocket.url}")
  26.     private String localWebSocketUrl;
  27.  
  28.     private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
  29.  
  30.     /**
  31.      * 平台WebSocket连接心跳检测，重连机制，每30秒触发一次
  32.      * 注意 @Async 注解，要使用异步线程的方式来执行心跳检测，
  33.      * 避免任务线程被其他任务占用
  34.      */
  35.     @Async
  36.     @Scheduled(cron = "0/30 * * * * ?")
  37.     public void platformWebSocket() {
  38.         try {
  39.             int isConnect = PlatformWebsocketClient.isConnect;
  40. //            log.info("心跳检测 -> platformWebSocket： {}-{}", isConnect, ((isConnect == 1) ? "连接中" : "未连接"));
  41.             if (1 != PlatformWebsocketClient.isConnect) {
  42.                 String now = DATE_TIME_FORMATTER.format(LocalDateTime.now());
  43. //                log.info("心跳检测 -> platformWebSocket服务连接异常，时间：{}，尝试重新连接---", now);
  44.                 platformWebsocketClient.connect(platformWebSocketUrl, null);
  45.             }
  46.         } catch (Exception e) {
  47.             e.printStackTrace();
  48.         }
  49.     }
  50.  
  51.  
  52.  
  53.     /**
  54.      * 本地WebSocket连接心跳检测，重连机制，每30秒触发一次
  55.      * 注意 @Async 注解，要使用异步线程的方式来执行心跳检测，
  56.      * 避免任务线程被其他任务占用
  57.      */
  58.     @Async
  59.     @Scheduled(cron = "0/30 * * * * ?")
  60.     public void LocalWebSocket() {
  61.         try {
  62.             int isConnect = LocalWebsocketClient.isConnect;
  63. //            log.info("心跳检测 -> LocalWebSocket： {}-{}", isConnect, ((isConnect == 1) ? "连接中" : "未连接"));
  64.             if (1 != LocalWebsocketClient.isConnect) {
  65.                 String now = DATE_TIME_FORMATTER.format(LocalDateTime.now());
  66. //                log.info("心跳检测 -> LocalWebSocket服务连接异常，时间：{}，尝试重新连接---", now);
  67.                 localWebsocketClient.connect(localWebSocketUrl, null);
  68.             }
  69.         } catch (Exception e) {
  70.             e.printStackTrace();
  71.         }
  72.     }
  73.  
  74. }
MessageService：消息转发，从平台客户端转发至本地客户端，本地客户端需要连接websocket服务

  1.  
  2. import com.alibaba.fastjson.JSONObject;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.java_websocket.client.WebSocketClient;
  5. import org.springframework.stereotype.Service;
  6.  
  7. import javax.annotation.Resource;
  8. import java.util.ArrayList;
  9. import java.util.List;
  10.  
  11. @Service
  12. @Slf4j
  13. public class MessageService {
  14.     /**
  15.      * 处理消息
  16.      * 这里可根据具体业务来实现，比如解析入库、再次分发发送MQ等
  17.      * @param message
  18.      */
  19.  
  20.     @Resource
  21.     private LocalWebsocketClient localWebsocketClient;
  22.  
  23.     public void handleMessage(String message) {
  24.         JSONObject jsonObject = JSONObject.parseObject(message);
  25.         send(jsonObject.toJSONString());
  26.       
  27.     }
  28.  
  29.  
  30.     public void send(String message){
  31.         WebSocketClient client = localWebsocketClient.getWsClient();
  32.         client.send(message);
  33.     }
  34. }
SubWebSocket：前端推送服务
  1.  
  2. import com.alibaba.fastjson.JSONArray;
  3. import com.alibaba.fastjson.JSONObject;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.stereotype.Component;
  6.  
  7. import javax.websocket.OnClose;
  8. import javax.websocket.OnMessage;
  9. import javax.websocket.OnOpen;
  10. import javax.websocket.Session;
  11. import javax.websocket.server.PathParam;
  12. import javax.websocket.server.ServerEndpoint;
  13. import java.util.*;
  14. import java.util.concurrent.CopyOnWriteArraySet;
  15.  
  16. @Component
  17. @Slf4j
  18. @ServerEndpoint(value = "/ws-subscribe/{userId}")
  19. public class SubWebSocket {
  20.     /**
  21.      * 线程安全的无序的集合
  22.      */
  23.     private static final CopyOnWriteArraySet<Session> SESSIONS = new CopyOnWriteArraySet<>();
  24.  
  25.     /**
  26.      * 存储在线连接数
  27.      */
  28.     private static final Map<String, Session> SESSION_POOL = new HashMap<>();
  29.  
  30.     @OnOpen
  31.     public void onOpen(Session session, @PathParam(value = "userId") String userId) {
  32.         try {
  33.             SESSIONS.add(session);
  34.             SESSION_POOL.put(userId, session);
  35.         } catch (Exception e) {
  36.             e.printStackTrace();
  37.         }
  38.     }
  39.  
  40.     @OnClose
  41.     public void onClose(Session session) {
  42.         try {
  43.             SESSIONS.remove(session);
  44.         } catch (Exception e) {
  45.             e.printStackTrace();
  46.         }
  47.     }
  48.  
  49.     @OnMessage
  50.     public void onMessage(String message) {
  51.         JSONObject json = JSONObject.parseObject(message);
  52.         JSONArray subUsers = json.getJSONArray("subUser");
  53.         json.remove("subUser");
  54.         sendMoreMessage(subUsers,json.toJSONString());
  55.     }
  56.  
  57.     /**
  58.      * 此为广播消息
  59.      *
  60.      * @param message 消息
  61.      */
  62.     public void sendAllMessage(String message) {
  63.         for (Session session : SESSIONS) {
  64.             try {
  65.             if(session.isOpen() && session != SESSION_POOL.get("jeecgboot")){
  66.                 synchronized (session) {
  67.                     session.getBasicRemote().sendText(message);
  68.                 }
  69.             }
  70.             } catch (Exception e) {
  71.                 e.printStackTrace();
  72.             }
  73.         }
  74.     }
  75.  
  76.     /**
  77.      * 此为单点消息
  78.      *
  79.      * @param userId  用户编号
  80.      * @param message 消息
  81.      */
  82.     public void sendOneMessage(String userId, String message) {
  83.         Session session = SESSION_POOL.get(userId);
  84.         if (session != null && session.isOpen()) {
  85.             try {
  86.                 synchronized (session) {
  87.                     session.getAsyncRemote().sendText(message);
  88.                 }
  89.             } catch (Exception e) {
  90.                 e.printStackTrace();
  91.             }
  92.         }
  93.     }
  94.  
  95.     /**
  96.      * 此为单点消息(多人)
  97.      *
  98.      * @param userIds 用户编号列表
  99.      * @param message 消息
  100.      */
  101.     public void sendMoreMessage(JSONArray userIds, String message) {
  102.         if(userIds.size() != 0) {
  103.             for (Object userId : userIds) {
  104.                 Session session = SESSION_POOL.get(userId.toString());
  105.                 if (session != null && session.isOpen()) {
  106.                     try {
  107.                         synchronized(session) {
  108.                             session.getAsyncRemote().sendText(message);
  109.                         }
  110.                     } catch (Exception e) {
  111.                         e.printStackTrace();
  112.                     }
  113.                 }
  114.             }
  115.         }
  116.     }
  117.  
  118.  
  119. }

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

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

  

闽ICP备14008679号