fix(gateApi): 修复止损订单数量参数类型及优化WebSocket心跳机制

Administrator

8 hours ago bba9786444b39d1c84a1291605426bd2afbbe849

 src/main/java/com/xcong/excoin/modules/gateApi/GateKlineWebSocketClient.java

@@ -15,6 +15,7 @@
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

/**
@@ -39,12 +40,15 @@
 *     4. update/all           → 遍历 channelHandlers → handler.handleMessage(response)
 * </pre>
 *
 * <h3>心跳与断线检测（双层）</h3>
 * <h3>心跳与断线检测（主动健康检查，依据 Gate.io 官方 ping/pong 语义）</h3>
 * <ol>
 *   <li>TCP 层：Java-WebSocket 自带 ping/pong 保活（{@code connectionLostTimeout = 30s}），
 *       半开连接（对端静默丢包、无 FIN/RST）会在 30 秒内被库判定丢失并触发 onClose。</li>
 *   <li>应用层：{@code futures.ping}/{@code futures.pong}。10 秒未收到任何消息 → 发 ping；
 *       45 秒未收到任何消息 → 判定连接死亡，主动 close → onClose → 重连。</li>
 *       半开连接（对端静默丢包、无 FIN/RST）会被库判定丢失并触发 onClose。</li>
 *   <li>应用层：{@code futures.ping}/{@code futures.pong}。
 *       静默 10 秒 → 发 futures.ping 探测（可重置服务端关闭计数，官方文档语义）；
 *       静默 20 秒（ping 也无任何回应）→ 判定连接不健康，主动 close → onClose → 重连。</li>
 *   <li>主动换线：连接存活满 55 分钟主动优雅重连，绕开链路中间设备约 1 小时的强制断连，
 *       把被动盲区变为计划内切换。</li>
 * </ol>
 *
 * <h3>重连策略</h3>
@@ -63,12 +67,21 @@

    private static final String FUTURES_PING = "futures.ping";
    private static final String FUTURES_PONG = "futures.pong";
    /** 静默超过该秒数 → 发送 futures.ping */
    /** 静默超过该秒数 → 发送 futures.ping 探测（官方文档：客户端 ping 可重置服务端关闭计数） */
    private static final int HEARTBEAT_TIMEOUT = 10;
    /** 静默超过该秒数 → 判定连接死亡，主动断开重连 */
    private static final int DEAD_TIMEOUT_SECONDS = 45;
    /** Java-WebSocket 库层 ping/pong 保活超时（秒），用于检测半开 TCP 连接 */
    /**
     * 静默超过该秒数（且已发出过探测 ping）→ 判定连接不健康，主动断开重连。
     * 即：10s 静默发 ping，再等 10s 仍无任何消息（连 pong 都没有）→ 判定死亡。
     */
    private static final int DEAD_TIMEOUT_SECONDS = 20;
    /** Java-WebSocket 库层 ping/pong 保活超时（秒），用于检测半开 TCP 连接（兜底） */
    private static final int CONNECTION_LOST_TIMEOUT_SECONDS = 30;
    /**
     * 连接最大存活时长（秒）：到达后主动优雅断开重连（"主动换线"）。
     * 用于绕开链路中间设备（腾讯云出口 NAT / 国际出口）约 1 小时的连接强制断连，
     * 把被动盲区变成计划内换线。设为 0 可禁用。
     */
    private static final int MAX_CONNECTION_AGE_SECONDS = 55 * 60;
    /** 重连初始延迟（毫秒） */
    private static final long INITIAL_RECONNECT_DELAY_MS = 5000;
    /** 重连最大延迟（毫秒），指数退避封顶值 */
@@ -85,6 +98,8 @@
    private volatile ScheduledFuture<?> pongTimeoutFuture;
    /** 最后收到消息的时间戳（毫秒） */
    private final AtomicReference<Long> lastMessageTime = new AtomicReference<>(System.currentTimeMillis());
    /** 当前连接建立时间（毫秒），用于主动换线 */
    private final AtomicLong connectionOpenedTime = new AtomicLong(System.currentTimeMillis());

    /** 连接状态 */
    private final AtomicBoolean isConnected = new AtomicBoolean(false);
@@ -221,6 +236,8 @@
                    isConnected.set(true);
                    isConnecting.set(false);
                    reconnectAttempt.set(0);
                    connectionOpenedTime.set(System.currentTimeMillis());
                    lastMessageTime.set(System.currentTimeMillis());
                    if (sharedExecutor != null && !sharedExecutor.isShutdown()) {
                        for (GateChannelHandler handler : channelHandlers) {
                            handler.setSubscribed(false);
@@ -357,24 +374,49 @@
    }

    /**
     * 心跳检查（由 25s 周期任务 + 每次收到消息后 10s 的一次性任务触发）：
     * 心跳检查（由 25s 周期任务 + 每次收到消息后 10s 的一次性任务触发）。
     * 主动健康检查逻辑（依据 Gate.io 官方文档 ping/pong 语义）：
     * <ol>
     *   <li>静默 ≥ {@link #DEAD_TIMEOUT_SECONDS} 秒 → 判定连接死亡（半开 TCP / 对端失联），主动断开触发重连；</li>
     *   <li>静默 ≥ {@link #HEARTBEAT_TIMEOUT} 秒 → 发送 futures.ping 探测。</li>
     *   <li>连接存活 ≥ {@link #MAX_CONNECTION_AGE_SECONDS} 秒 → 主动优雅断开重连（主动换线，
     *       绕开链路中间设备约 1 小时的强制断连，把被动盲区变为计划内切换）；</li>
     *   <li>静默 ≥ {@link #HEARTBEAT_TIMEOUT} 秒 → 发送 futures.ping 探测，并安排 10 秒后复查；</li>
     *   <li>静默 ≥ {@link #DEAD_TIMEOUT_SECONDS} 秒（探测 ping 也得不到任何回应，包括 pong）→
     *       判定连接不健康，主动断开触发重连。</li>
     * </ol>
     */
    private void checkHeartbeatTimeout() {
        if (!isConnected.get() || destroyed.get()) return;
        long silentMs = System.currentTimeMillis() - lastMessageTime.get();
        if (silentMs >= DEAD_TIMEOUT_SECONDS * 1000L) {
            log.warn("[WS] 已 {} 秒未收到任何消息，判定连接死亡，主动断开重连", silentMs / 1000);
            if (webSocketClient != null) {
                try { webSocketClient.close(); } catch (Exception e) { log.warn("[WS] 主动断开失败", e); }
            }
        long now = System.currentTimeMillis();
        long silentMs = now - lastMessageTime.get();

        // 1) 主动换线：到达连接最大存活时长
        long ageMs = now - connectionOpenedTime.get();
        if (MAX_CONNECTION_AGE_SECONDS > 0 && ageMs >= MAX_CONNECTION_AGE_SECONDS * 1000L) {
            log.warn("[WS] 连接已存活 {} 秒，达到主动换线阈值，主动断开重连", ageMs / 1000);
            closeSocket();
            return;
        }
        if (silentMs >= HEARTBEAT_TIMEOUT * 1000L) {

        // 2) 探测：静默 10s → 发 futures.ping，并安排复查
        if (silentMs >= HEARTBEAT_TIMEOUT * 1000L && silentMs < DEAD_TIMEOUT_SECONDS * 1000L) {
            sendPing();
            resetHeartbeatTimer();
            return;
        }

        // 3) 判定不健康：20s 静默（ping 后仍无 pong/任何消息）→ 断开重连
        if (silentMs >= DEAD_TIMEOUT_SECONDS * 1000L) {
            log.warn("[WS] 已 {} 秒未收到任何消息（ping 无回应），判定连接不健康，断开重连", silentMs / 1000);
            closeSocket();
        }
    }

    /**
     * 主动关闭 WebSocket（非阻塞），触发 onClose → 重连流程。
     */
    private void closeSocket() {
        if (webSocketClient != null) {
            try { webSocketClient.close(); } catch (Exception e) { log.warn("[WS] 主动断开失败", e); }
        }
    }