From 2e6cbd3ee1c2d7e1aa73f4392c79bfbfacbcb67d Mon Sep 17 00:00:00 2001
From: Administrator <15274802129@163.com>
Date: Mon, 15 Dec 2025 14:03:36 +0800
Subject: [PATCH] feat(okx): 支持多账号WebSocket连接管理
---
src/main/java/com/xcong/excoin/modules/okxNewPrice/OkxQuantWebSocketClient.java | 190 +++++++++++++++++++++++++++--------------------
1 files changed, 108 insertions(+), 82 deletions(-)
diff --git a/src/main/java/com/xcong/excoin/modules/okxNewPrice/OkxQuantWebSocketClient.java b/src/main/java/com/xcong/excoin/modules/okxNewPrice/OkxQuantWebSocketClient.java
index 0199950..b99d26c 100644
--- a/src/main/java/com/xcong/excoin/modules/okxNewPrice/OkxQuantWebSocketClient.java
+++ b/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.*;
@@ -35,15 +30,11 @@
* @author Administrator
*/
@Slf4j
-@Component
-@ConditionalOnProperty(prefix = "app", name = "quant", havingValue = "true")
public class OkxQuantWebSocketClient {
- @Autowired
- private WangGeService wangGeService;
- @Autowired
- private CaoZuoService caoZuoService;
- @Autowired
- private RedisUtils redisUtils;
+ private final WangGeService wangGeService;
+ private final CaoZuoService caoZuoService;
+ private final RedisUtils redisUtils;
+ private final ExchangeInfoEnum account;
private WebSocketClient webSocketClient;
private ScheduledExecutorService heartbeatExecutor;
@@ -53,13 +44,17 @@
// 连接状态标志
private final AtomicBoolean isConnected = new AtomicBoolean(false);
private final AtomicBoolean isConnecting = new AtomicBoolean(false);
+
+ public OkxQuantWebSocketClient(ExchangeInfoEnum account, WangGeService wangGeService,
+ CaoZuoService caoZuoService, RedisUtils redisUtils) {
+ this.account = account;
+ this.wangGeService = wangGeService;
+ this.caoZuoService = caoZuoService;
+ this.redisUtils = redisUtils;
+ }
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,40 +72,73 @@
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();
}
/**
* 销毁方法，在 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);
- if (reconnectScheduler != null) {
- reconnectScheduler.shutdownNow();
- }
+ log.info("OkxQuantWebSocketClient销毁完成");
}
private void shutdownExecutorGracefully(ExecutorService executor) {
@@ -145,12 +173,12 @@
}
try {
- InstrumentsWs.handleEvent();
+ InstrumentsWs.handleEvent(account.name());
wangGeService.initWangGe();
SSLConfig.configureSSL();
System.setProperty("https.protocols", "TLSv1.2,TLSv1.3");
String WS_URL = WS_URL_MONIPAN;
- if (ExchangeInfoEnum.OKX_UAT.isAccountType()){
+ if (account.isAccountType()){
WS_URL = WS_URL_SHIPAN;
}
URI uri = new URI(WS_URL);
@@ -175,7 +203,7 @@
// 棜查应用是否正在关闭
if (!sharedExecutor.isShutdown()) {
resetHeartbeatTimer();
- websocketLogin();
+ websocketLogin(account);
} else {
log.warn("应用正在关闭，忽略WebSocket连接成功回调");
}
@@ -225,8 +253,8 @@
}
}
- private void websocketLogin() {
- LoginWs.websocketLogin(webSocketClient);
+ private void websocketLogin(ExchangeInfoEnum account) {
+ LoginWs.websocketLogin(webSocketClient, account);
}
private void subscribeBalanceAndPositionChannel(String option) {
@@ -254,7 +282,7 @@
private void handleWebSocketMessage(String message) {
try {
if ("pong".equals(message)) {
- log.debug("收到心跳响应");
+ log.debug("{}: 收到心跳响应", account.name());
cancelPongTimeout();
return;
}
@@ -265,26 +293,49 @@
String code = response.getString("code");
if ("0".equals(code)) {
String connId = response.getString("connId");
- log.info("WebSocket登录成功, connId: {}", connId);
+ log.info("{}: WebSocket登录成功, connId: {}", account.name(), connId);
subscribeAccountChannel(SUBSCRIBE);
subscribeOrderInfoChannel(SUBSCRIBE);
subscribePositionChannel(SUBSCRIBE);
} else {
- log.error("WebSocket登录失败, code: {}, msg: {}", code, response.getString("msg"));
+ log.error("{}: WebSocket登录失败, code: {}, msg: {}", account.name(), code, response.getString("msg"));
}
} else if ("subscribe".equals(event)) {
- log.info("订阅成功: {}", response.getJSONObject("arg"));
+ subscribeEvent(response);
} else if ("error".equals(event)) {
- log.error("订阅错误: code={}, msg={}",
- response.getString("code"), response.getString("msg"));
+ log.error("{}: 订阅错误: code={}, msg={}",
+ account.name(), response.getString("code"), response.getString("msg"));
} else if ("channel-conn-count".equals(event)) {
- log.info("连接限制更新: channel={}, connCount={}",
- response.getString("channel"), response.getString("connCount"));
+ log.info("{}: 连接限制更新: channel={}, connCount={}",
+ account.name(), response.getString("channel"), response.getString("connCount"));
} else {
processPushData(response);
}
} catch (Exception e) {
- log.error("处理WebSocket消息失败: {}", message, e);
+ log.error("{}: 处理WebSocket消息失败: {}", account.name(), message, e);
+ }
+ }
+
+ private void subscribeEvent(JSONObject response) {
+ JSONObject arg = response.getJSONObject("arg");
+ if (arg == null) {
+ log.warn("无效的推送数据，缺少 'arg' 字段 :{}",response);
+ return;
+ }
+
+ String channel = arg.getString("channel");
+ if (channel == null) {
+ log.warn("无效的推送数据，缺少 'channel' 字段{}",response);
+ return;
+ }
+ if (OrderInfoWs.ORDERINFOWS_CHANNEL.equals(channel)) {
+ OrderInfoWs.initEvent(response, account.name());
+ }
+ if (AccountWs.ACCOUNTWS_CHANNEL.equals(channel)) {
+ AccountWs.initEvent(response, account.name());
+ }
+ if (PositionsWs.POSITIONSWS_CHANNEL.equals(channel)) {
+ PositionsWs.initEvent(response, account.name());
}
}
@@ -298,30 +349,34 @@
String op = response.getString("op");
if (op != null){
if (TradeOrderWs.ORDERWS_CHANNEL.equals(op)) {
- log.info("收到下单推动结果: {}", response.getJSONObject("data"));
- return ;
+ // 直接使用Object类型接收，避免强制类型转换
+ Object data = response.get("data");
+ log.info("{}: 收到下单推送结果: {}", account.name(), JSON.toJSONString(data));
+ return;
}
}
JSONObject arg = response.getJSONObject("arg");
if (arg == null) {
- log.warn("无效的推送数据，缺少 'arg' 字段 :{}",response);
+ log.warn("{}: 无效的推送数据，缺少 'arg' 字段 :{}", account.name(), response);
return;
}
String channel = arg.getString("channel");
if (channel == null) {
- log.warn("无效的推送数据，缺少 'channel' 字段{}",response);
+ log.warn("{}: 无效的推送数据，缺少 'channel' 字段{}", account.name(), response);
return;
}
+ // 注意：当前实现中，OrderInfoWs等类使用静态Map存储数据
+ // 这会导致多账号之间的数据冲突。需要进一步修改这些类的设计，让数据存储与特定账号关联
if (OrderInfoWs.ORDERINFOWS_CHANNEL.equals(channel)) {
- OrderInfoWs.handleEvent(response, redisUtils);
+ OrderInfoWs.handleEvent(response, redisUtils, account.name());
}else if (AccountWs.ACCOUNTWS_CHANNEL.equals(channel)) {
- AccountWs.handleEvent(response);
- String side = caoZuoService.caoZuo();
- TradeOrderWs.orderEvent(webSocketClient, side);
+ AccountWs.handleEvent(response, account.name());
+ String side = caoZuoService.caoZuo(account.name());
+ TradeOrderWs.orderEvent(webSocketClient, side, account.name());
} else if (PositionsWs.POSITIONSWS_CHANNEL.equals(channel)) {
- PositionsWs.handleEvent(response);
+ PositionsWs.handleEvent(response, account.name());
} else if (BalanceAndPositionWs.CHANNEL_NAME.equals(channel)) {
BalanceAndPositionWs.handleEvent(response);
}
@@ -346,24 +401,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 方法
/**
* 重置心跳计时器。
@@ -379,19 +417,7 @@
}
}
- /**
- * 执行定时重连任务
- * 每小时强制重连一次以确保连接的新鲜度
- */
- private void performScheduledReconnect() {
- log.info("执行定时重连任务");
- if (webSocketClient != null && webSocketClient.isOpen()) {
- log.info("关闭当前连接准备重连");
- webSocketClient.close();
- }
- // 更新最后重连时间
- lastReconnectTime.set(System.currentTimeMillis());
- }
+ // 移除了 performScheduledReconnect 方法
/**
* 检查心跳超时情况。
--
Gitblit v1.9.1