feat(okxNewPrice): 添加防止重复初始化的机制

Administrator

2025-12-15 7981067b68ee94b27d22dd4619c98acdddc3379b

 src/main/java/com/xcong/excoin/modules/okxNewPrice/OkxQuantWebSocketClient.java

@@ -1,14 +1,10 @@
package com.xcong.excoin.modules.okxNewPrice;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xcong.excoin.modules.okxNewPrice.celue.CaoZuoService;
import com.xcong.excoin.modules.okxNewPrice.okxWs.*;
import com.xcong.excoin.modules.okxNewPrice.okxWs.enums.CoinEnums;
import com.xcong.excoin.modules.okxNewPrice.okxWs.enums.ExchangeInfoEnum;
import com.xcong.excoin.modules.okxNewPrice.okxWs.enums.OrderParamEnums;
import com.xcong.excoin.modules.okxNewPrice.utils.SSLConfig;
import com.xcong.excoin.modules.okxNewPrice.wangge.WangGeService;
import com.xcong.excoin.utils.RedisUtils;
@@ -21,7 +17,6 @@

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.math.BigDecimal;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.concurrent.*;
@@ -57,10 +52,6 @@
    private static final String WS_URL_MONIPAN = "wss://wspap.okx.com:8443/ws/v5/private";
    private static final String WS_URL_SHIPAN = "wss://ws.okx.com:8443/ws/v5/private";

    private ScheduledExecutorService reconnectScheduler;
    private final AtomicReference<Long> lastReconnectTime = new AtomicReference<>(System.currentTimeMillis());


    /**
     * 订阅频道指令
     */
@@ -77,17 +68,23 @@
        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();
        
        // 添加每小时重连的定时任务
        schedulePeriodicReconnect();
    }

    /**
@@ -108,9 +105,7 @@
        }
        shutdownExecutorGracefully(sharedExecutor);

        if (reconnectScheduler != null) {
            reconnectScheduler.shutdownNow();
        }
        // 移除了 reconnectScheduler 的关闭操作
    }

    private void shutdownExecutorGracefully(ExecutorService executor) {
@@ -343,13 +338,13 @@
            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);
    }

    /**
@@ -371,24 +366,7 @@
                HEARTBEAT_TIMEOUT, HEARTBEAT_TIMEOUT, TimeUnit.SECONDS);
    }

    /**
     * 安排定期重连任务
     * 每小时执行一次重连以保持连接新鲜度
     */
    private void schedulePeriodicReconnect() {
        if (reconnectScheduler != null && !reconnectScheduler.isTerminated()) {
            reconnectScheduler.shutdownNow();
        }

        reconnectScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread t = new Thread(r, "okx-scheduled-reconnect");
            t.setDaemon(true);
            return t;
        });

        // 每小时执行一次重连
        reconnectScheduler.scheduleWithFixedDelay(this::performScheduledReconnect, 60, 60, TimeUnit.MINUTES);
    }
    // 移除了 schedulePeriodicReconnect 方法

    /**
     * 重置心跳计时器。
@@ -404,19 +382,7 @@
        }
    }

    /**
     * 执行定时重连任务
     * 每小时强制重连一次以确保连接的新鲜度
     */
    private void performScheduledReconnect() {
        log.info("执行定时重连任务");
        if (webSocketClient != null && webSocketClient.isOpen()) {
            log.info("关闭当前连接准备重连");
            webSocketClient.close();
        }
        // 更新最后重连时间
        lastReconnectTime.set(System.currentTimeMillis());
    }
    // 移除了 performScheduledReconnect 方法

    /**
     * 检查心跳超时情况。