| | |
| | | return t; |
| | | }); |
| | | |
| | | // 在 OkxQuantWebSocketClient 中添加初始化标记 |
| | | private final AtomicBoolean isInitialized = new AtomicBoolean(false); |
| | | |
| | | /** |
| | | * 初始化方法,在 Spring Bean 构造完成后执行。 |
| | | * 负责建立 WebSocket 连接并启动心跳检测任务。 |
| | | */ |
| | | @PostConstruct |
| | | public void init() { |
| | | // 防止重复初始化 |
| | | if (!isInitialized.compareAndSet(false, true)) { |
| | | log.warn("OkxQuantWebSocketClient 已经初始化过,跳过重复初始化"); |
| | | return; |
| | | } |
| | | |
| | | connect(); |
| | | startHeartbeat(); |
| | | |
| | | // 移除了每小时重连的定时任务 |
| | | } |
| | | |
| | | /** |
| | | * 销毁方法,在 Spring Bean 销毁前执行。 |
| | | * 关闭 WebSocket 连接、停止心跳定时器及相关的线程资源。 |
| | | */ |
| | | // @PreDestroy |
| | | // public void destroy() { |
| | | // if (webSocketClient != null && webSocketClient.isOpen()) { |
| | | // subscribeAccountChannel(UNSUBSCRIBE); |
| | | // subscribePositionChannel(UNSUBSCRIBE); |
| | | // subscribeOrderInfoChannel(UNSUBSCRIBE); |
| | | // webSocketClient.close(); |
| | | // } |
| | | // shutdownExecutorGracefully(heartbeatExecutor); |
| | | // if (pongTimeoutFuture != null) { |
| | | // pongTimeoutFuture.cancel(true); |
| | | // } |
| | | // shutdownExecutorGracefully(sharedExecutor); |
| | | // |
| | | // // 移除了 reconnectScheduler 的关闭操作 |
| | | // } |
| | | @PreDestroy |
| | | public void destroy() { |
| | | if (webSocketClient != null && webSocketClient.isOpen()) { |
| | | subscribeAccountChannel(UNSUBSCRIBE); |
| | | subscribePositionChannel(UNSUBSCRIBE); |
| | | subscribeOrderInfoChannel(UNSUBSCRIBE); |
| | | webSocketClient.close(); |
| | | log.info("开始销毁OkxQuantWebSocketClient"); |
| | | |
| | | // 设置关闭标志,避免重连 |
| | | if (sharedExecutor != null && !sharedExecutor.isShutdown()) { |
| | | sharedExecutor.shutdown(); |
| | | } |
| | | |
| | | if (webSocketClient != null && webSocketClient.isOpen()) { |
| | | try { |
| | | subscribeAccountChannel(UNSUBSCRIBE); |
| | | subscribePositionChannel(UNSUBSCRIBE); |
| | | subscribeOrderInfoChannel(UNSUBSCRIBE); |
| | | webSocketClient.closeBlocking(); |
| | | } catch (InterruptedException e) { |
| | | Thread.currentThread().interrupt(); |
| | | log.warn("关闭WebSocket连接时被中断"); |
| | | } |
| | | } |
| | | |
| | | shutdownExecutorGracefully(heartbeatExecutor); |
| | | if (pongTimeoutFuture != null) { |
| | | pongTimeoutFuture.cancel(true); |
| | | } |
| | | shutdownExecutorGracefully(sharedExecutor); |
| | | |
| | | // 移除了 reconnectScheduler 的关闭操作 |
| | | log.info("OkxQuantWebSocketClient销毁完成"); |
| | | } |
| | | |
| | | private void shutdownExecutorGracefully(ExecutorService executor) { |
| | |
| | | OrderInfoWs.handleEvent(response, redisUtils); |
| | | }else if (AccountWs.ACCOUNTWS_CHANNEL.equals(channel)) { |
| | | AccountWs.handleEvent(response); |
| | | String side = caoZuoService.caoZuo(); |
| | | TradeOrderWs.orderEvent(webSocketClient, side); |
| | | } else if (PositionsWs.POSITIONSWS_CHANNEL.equals(channel)) { |
| | | PositionsWs.handleEvent(response); |
| | | } else if (BalanceAndPositionWs.CHANNEL_NAME.equals(channel)) { |
| | | BalanceAndPositionWs.handleEvent(response); |
| | | } |
| | | String side = caoZuoService.caoZuo(); |
| | | TradeOrderWs.orderEvent(webSocketClient, side); |
| | | } |
| | | |
| | | /** |