| New file |
| | |
|---|
| | | package com.xcong.excoin.modules.okxNewPrice; |
|---|
| | | |
|---|
| | | import com.alibaba.fastjson.JSON; |
|---|
| | | import com.alibaba.fastjson.JSONArray; |
|---|
| | | import com.alibaba.fastjson.JSONObject; |
|---|
| | | import com.xcong.excoin.modules.okxNewPrice.celue.CaoZuoService; |
|---|
| | | import com.xcong.excoin.modules.okxNewPrice.okxWs.TradeOrderWs; |
|---|
| | | import com.xcong.excoin.modules.okxNewPrice.okxWs.enums.CoinEnums; |
|---|
| | | import com.xcong.excoin.modules.okxNewPrice.okxWs.param.TradeRequestParam; |
|---|
| | | import com.xcong.excoin.modules.okxNewPrice.okxWs.wanggeList.WangGeListEnum; |
|---|
| | | import com.xcong.excoin.modules.okxNewPrice.okxWs.wanggeList.WangGeListService; |
|---|
| | | import com.xcong.excoin.modules.okxNewPrice.utils.SSLConfig; |
|---|
| | | import com.xcong.excoin.utils.RedisUtils; |
|---|
| | | import java.math.BigDecimal; |
|---|
| | | import lombok.extern.slf4j.Slf4j; |
|---|
| | | import org.java_websocket.client.WebSocketClient; |
|---|
| | | import org.java_websocket.handshake.ServerHandshake; |
|---|
| | | |
|---|
| | | import java.net.URI; |
|---|
| | | import java.net.URISyntaxException; |
|---|
| | | import java.util.Collection; |
|---|
| | | import java.util.concurrent.*; |
|---|
| | | import java.util.concurrent.atomic.AtomicBoolean; |
|---|
| | | import java.util.concurrent.atomic.AtomicReference; |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * OKX 新价格 WebSocket 客户端类，用于连接 OKX 的 WebSocket 接口， |
|---|
| | | * 实时获取并处理标记价格（mark price）数据，并将价格信息存储到 Redis 中。 |
|---|
| | | * 同时支持心跳检测、自动重连以及异常恢复机制。 |
|---|
| | | * @author Administrator |
|---|
| | | */ |
|---|
| | | @Slf4j |
|---|
| | | public class OkxNewPriceWebSocketClient { |
|---|
| | | private final RedisUtils redisUtils; |
|---|
| | | private final CaoZuoService caoZuoService; |
|---|
| | | private final OkxWebSocketClientManager clientManager; |
|---|
| | | private final WangGeListService wangGeListService; |
|---|
| | | |
|---|
| | | private WebSocketClient webSocketClient; |
|---|
| | | private ScheduledExecutorService heartbeatExecutor; |
|---|
| | | private volatile ScheduledFuture<?> pongTimeoutFuture; |
|---|
| | | private final AtomicReference<Long> lastMessageTime = new AtomicReference<>(System.currentTimeMillis()); |
|---|
| | | |
|---|
| | | // 连接状态标志 |
|---|
| | | private final AtomicBoolean isConnected = new AtomicBoolean(false); |
|---|
| | | private final AtomicBoolean isConnecting = new AtomicBoolean(false); |
|---|
| | | private final AtomicBoolean isInitialized = new AtomicBoolean(false); |
|---|
| | | |
|---|
| | | private static final String CHANNEL = "mark-price"; |
|---|
| | | |
|---|
| | | // 心跳超时时间(秒)，小于30秒 |
|---|
| | | private static final int HEARTBEAT_TIMEOUT = 10; |
|---|
| | | |
|---|
| | | // 共享线程池用于重连等异步任务 |
|---|
| | | private final ExecutorService sharedExecutor = Executors.newCachedThreadPool(r -> { |
|---|
| | | Thread t = new Thread(r, "okx-ws-shared-worker"); |
|---|
| | | t.setDaemon(true); |
|---|
| | | return t; |
|---|
| | | }); |
|---|
| | | |
|---|
| | | public OkxNewPriceWebSocketClient(RedisUtils redisUtils, |
|---|
| | | CaoZuoService caoZuoService, OkxWebSocketClientManager clientManager, |
|---|
| | | WangGeListService wangGeListService) { |
|---|
| | | this.redisUtils = redisUtils; |
|---|
| | | this.caoZuoService = caoZuoService; |
|---|
| | | this.clientManager = clientManager; |
|---|
| | | this.wangGeListService = wangGeListService; |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 初始化方法，创建并初始化WebSocket客户端实例 |
|---|
| | | */ |
|---|
| | | public void init() { |
|---|
| | | if (!isInitialized.compareAndSet(false, true)) { |
|---|
| | | log.warn("OkxNewPriceWebSocketClient 已经初始化过，跳过重复初始化"); |
|---|
| | | return; |
|---|
| | | } |
|---|
| | | connect(); |
|---|
| | | startHeartbeat(); |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 销毁方法，关闭WebSocket连接和相关资源 |
|---|
| | | */ |
|---|
| | | public void destroy() { |
|---|
| | | log.info("开始销毁OkxNewPriceWebSocketClient"); |
|---|
| | | |
|---|
| | | // 设置关闭标志，避免重连 |
|---|
| | | if (sharedExecutor != null && !sharedExecutor.isShutdown()) { |
|---|
| | | sharedExecutor.shutdown(); |
|---|
| | | } |
|---|
| | | |
|---|
| | | if (webSocketClient != null && webSocketClient.isOpen()) { |
|---|
| | | try { |
|---|
| | | webSocketClient.closeBlocking(); |
|---|
| | | } catch (InterruptedException e) { |
|---|
| | | Thread.currentThread().interrupt(); |
|---|
| | | log.warn("关闭WebSocket连接时被中断"); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | shutdownExecutorGracefully(heartbeatExecutor); |
|---|
| | | if (pongTimeoutFuture != null) { |
|---|
| | | pongTimeoutFuture.cancel(true); |
|---|
| | | } |
|---|
| | | shutdownExecutorGracefully(sharedExecutor); |
|---|
| | | |
|---|
| | | log.info("OkxNewPriceWebSocketClient销毁完成"); |
|---|
| | | } |
|---|
| | | |
|---|
| | | private static final String WS_URL_MONIPAN = "wss://wspap.okx.com:8443/ws/v5/public"; |
|---|
| | | private static final String WS_URL_SHIPAN = "wss://ws.okx.com:8443/ws/v5/public"; |
|---|
| | | private static final boolean isAccountType = true; |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 建立与 OKX WebSocket 服务器的连接。 |
|---|
| | | * 设置回调函数以监听连接打开、接收消息、关闭和错误事件。 |
|---|
| | | */ |
|---|
| | | private void connect() { |
|---|
| | | // 避免重复连接 |
|---|
| | | if (isConnecting.get()) { |
|---|
| | | log.info("连接已在进行中，跳过重复连接请求"); |
|---|
| | | return; |
|---|
| | | } |
|---|
| | | |
|---|
| | | if (!isConnecting.compareAndSet(false, true)) { |
|---|
| | | log.info("连接已在进行中，跳过重复连接请求"); |
|---|
| | | return; |
|---|
| | | } |
|---|
| | | |
|---|
| | | try { |
|---|
| | | SSLConfig.configureSSL(); |
|---|
| | | System.setProperty("https.protocols", "TLSv1.2,TLSv1.3"); |
|---|
| | | String WS_URL = WS_URL_MONIPAN; |
|---|
| | | if (isAccountType){ |
|---|
| | | WS_URL = WS_URL_SHIPAN; |
|---|
| | | } |
|---|
| | | URI uri = new URI(WS_URL); |
|---|
| | | |
|---|
| | | // 关闭之前的连接（如果存在） |
|---|
| | | if (webSocketClient != null) { |
|---|
| | | try { |
|---|
| | | webSocketClient.closeBlocking(); |
|---|
| | | } catch (InterruptedException e) { |
|---|
| | | Thread.currentThread().interrupt(); |
|---|
| | | log.warn("关闭之前连接时被中断"); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | webSocketClient = new WebSocketClient(uri) { |
|---|
| | | @Override |
|---|
| | | public void onOpen(ServerHandshake handshake) { |
|---|
| | | log.info("OKX New Price WebSocket连接成功"); |
|---|
| | | isConnected.set(true); |
|---|
| | | isConnecting.set(false); |
|---|
| | | |
|---|
| | | // 检查应用是否正在关闭 |
|---|
| | | if (sharedExecutor != null && !sharedExecutor.isShutdown()) { |
|---|
| | | resetHeartbeatTimer(); |
|---|
| | | subscribeChannels(); |
|---|
| | | } else { |
|---|
| | | log.warn("应用正在关闭，忽略WebSocket连接成功回调"); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | @Override |
|---|
| | | public void onMessage(String message) { |
|---|
| | | lastMessageTime.set(System.currentTimeMillis()); |
|---|
| | | handleWebSocketMessage(message); |
|---|
| | | resetHeartbeatTimer(); |
|---|
| | | } |
|---|
| | | |
|---|
| | | @Override |
|---|
| | | public void onClose(int code, String reason, boolean remote) { |
|---|
| | | log.warn("OKX New Price WebSocket连接关闭: code={}, reason={}", code, reason); |
|---|
| | | isConnected.set(false); |
|---|
| | | isConnecting.set(false); |
|---|
| | | cancelPongTimeout(); |
|---|
| | | |
|---|
| | | if (sharedExecutor != null && !sharedExecutor.isShutdown() && !sharedExecutor.isTerminated()) { |
|---|
| | | sharedExecutor.execute(() -> { |
|---|
| | | try { |
|---|
| | | reconnectWithBackoff(); |
|---|
| | | } catch (InterruptedException e) { |
|---|
| | | Thread.currentThread().interrupt(); |
|---|
| | | log.error("重连线程被中断", e); |
|---|
| | | } catch (Exception e) { |
|---|
| | | log.error("重连失败", e); |
|---|
| | | } |
|---|
| | | }); |
|---|
| | | } else { |
|---|
| | | log.warn("共享线程池已关闭，无法执行重连任务"); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | @Override |
|---|
| | | public void onError(Exception ex) { |
|---|
| | | log.error("OKX New Price WebSocket发生错误", ex); |
|---|
| | | isConnected.set(false); |
|---|
| | | } |
|---|
| | | }; |
|---|
| | | |
|---|
| | | webSocketClient.connect(); |
|---|
| | | } catch (URISyntaxException e) { |
|---|
| | | log.error("WebSocket URI格式错误", e); |
|---|
| | | isConnecting.set(false); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 订阅指定交易对的价格通道。 |
|---|
| | | * 构造订阅请求并发送给服务端。 |
|---|
| | | */ |
|---|
| | | private void subscribeChannels() { |
|---|
| | | JSONObject subscribeMsg = new JSONObject(); |
|---|
| | | subscribeMsg.put("op", "subscribe"); |
|---|
| | | |
|---|
| | | JSONArray argsArray = new JSONArray(); |
|---|
| | | JSONObject arg = new JSONObject(); |
|---|
| | | arg.put("channel", CHANNEL); |
|---|
| | | arg.put("instId", CoinEnums.HE_YUE.getCode()); |
|---|
| | | argsArray.add(arg); |
|---|
| | | |
|---|
| | | subscribeMsg.put("args", argsArray); |
|---|
| | | webSocketClient.send(subscribeMsg.toJSONString()); |
|---|
| | | log.info("已发送价格订阅请求，订阅通道数: {}", argsArray.size()); |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 处理从 WebSocket 收到的消息。 |
|---|
| | | * 包括订阅确认、错误响应、心跳响应以及实际的数据推送。 |
|---|
| | | * |
|---|
| | | * @param message 来自 WebSocket 的原始字符串消息 |
|---|
| | | */ |
|---|
| | | private void handleWebSocketMessage(String message) { |
|---|
| | | try { |
|---|
| | | JSONObject response = JSON.parseObject(message); |
|---|
| | | String event = response.getString("event"); |
|---|
| | | |
|---|
| | | if ("subscribe".equals(event)) { |
|---|
| | | log.info("价格订阅成功: {}", response.getJSONObject("arg")); |
|---|
| | | } else if ("error".equals(event)) { |
|---|
| | | log.error("价格订阅错误: code={}, msg={}", |
|---|
| | | response.getString("code"), response.getString("msg")); |
|---|
| | | } else if ("pong".equals(event)) { |
|---|
| | | log.debug("收到pong响应"); |
|---|
| | | cancelPongTimeout(); |
|---|
| | | } else { |
|---|
| | | processPushData(response); |
|---|
| | | } |
|---|
| | | } catch (Exception e) { |
|---|
| | | log.error("处理WebSocket消息失败: {}", message, e); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 解析并处理价格推送数据。 |
|---|
| | | * 将最新的标记价格存入 Redis 并触发后续业务逻辑比较处理。 |
|---|
| | | * 当价格变化时，调用CaoZuoService的caoZuo方法，触发所有账号的量化操作 |
|---|
| | | * |
|---|
| | | * @param response 包含价格数据的 JSON 对象 |
|---|
| | | */ |
|---|
| | | private void processPushData(JSONObject response) { |
|---|
| | | try { |
|---|
| | | JSONArray dataArray = response.getJSONArray("data"); |
|---|
| | | if (dataArray != null && !dataArray.isEmpty()) { |
|---|
| | | for (int i = 0; i < dataArray.size(); i++) { |
|---|
| | | try { |
|---|
| | | JSONObject priceData = dataArray.getJSONObject(i); |
|---|
| | | String instId = priceData.getString("instId"); |
|---|
| | | String markPx = priceData.getString("markPx"); |
|---|
| | | // 保存价格到Redis |
|---|
| | | redisUtils.set(CoinEnums.HE_YUE.getCode(), markPx); |
|---|
| | | |
|---|
| | | log.debug("更新最新价格: {} = {}, 币种: {}", CoinEnums.HE_YUE.getCode(), markPx, instId); |
|---|
| | | |
|---|
| | | // 价格变化时，触发所有账号的量化操作 |
|---|
| | | triggerQuantOperations(markPx); |
|---|
| | | } catch (Exception innerEx) { |
|---|
| | | log.warn("处理单条价格数据失败", innerEx); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } catch (Exception e) { |
|---|
| | | log.error("处理价格推送数据失败", e); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 触发所有账号的量化操作 |
|---|
| | | * @param markPx 当前标记价格 |
|---|
| | | */ |
|---|
| | | private void triggerQuantOperations(String markPx) { |
|---|
| | | try { |
|---|
| | | // 1. 判断当前价格属于哪个网格 |
|---|
| | | WangGeListEnum gridByPriceNew = WangGeListEnum.getGridByPrice(new BigDecimal(markPx)); |
|---|
| | | if (gridByPriceNew == null) { |
|---|
| | | log.error("当前价格{}不在任何网格范围内，无法触发量化操作", markPx); |
|---|
| | | return; |
|---|
| | | } |
|---|
| | | /** |
|---|
| | | * 获取当前网格信息 |
|---|
| | | * 根据当前网格的持仓方向获取反方向是否存在持仓 |
|---|
| | | * 如果持有，直接止损 |
|---|
| | | */ |
|---|
| | | Collection<OkxQuantWebSocketClient> allClients = clientManager.getAllClients(); |
|---|
| | | //如果为空，则直接返回 |
|---|
| | | if (allClients.isEmpty()) { |
|---|
| | | return; |
|---|
| | | } |
|---|
| | | // 获取所有OkxQuantWebSocketClient实例 |
|---|
| | | for (OkxQuantWebSocketClient client : clientManager.getAllClients()) { |
|---|
| | | String accountName = client.getAccountName(); |
|---|
| | | if (accountName != null) { |
|---|
| | | /** |
|---|
| | | * 处理历史网格的订单 |
|---|
| | | * 根据历史网格的开单方向，是否需要止损处理 |
|---|
| | | * 如果方向一致就不需要处理 |
|---|
| | | * 如果不一致则需要处理 |
|---|
| | | */ |
|---|
| | | String fangXiang = gridByPriceNew.getFang_xiang(); |
|---|
| | | String fangXiangOld = CoinEnums.POSSIDE_LONG.getCode().equals(fangXiang) ? CoinEnums.POSSIDE_SHORT.getCode() : CoinEnums.POSSIDE_LONG.getCode(); |
|---|
| | | log.info("历史网格方向为：{}", fangXiangOld); |
|---|
| | | if (!fangXiang.equals(fangXiangOld)){ |
|---|
| | | TradeRequestParam tradeRequestParamOld = caoZuoService.caoZuoZhiSunEvent(accountName, markPx, fangXiangOld); |
|---|
| | | TradeOrderWs.orderEvent(client.getWebSocketClient(), tradeRequestParamOld); |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 处理当前网格的订单，触发量化操作 |
|---|
| | | */ |
|---|
| | | log.info("当前价格{}属于网格: {}-{}({}-{})", markPx, gridByPriceNew.getName(),gridByPriceNew.getFang_xiang(), gridByPriceNew.getJiage_xiaxian(), gridByPriceNew.getJiage_shangxian()); |
|---|
| | | wangGeListService.initWangGe(markPx); |
|---|
| | | TradeRequestParam tradeRequestParam = caoZuoService.caoZuoHandler(accountName, markPx, gridByPriceNew.getFang_xiang()); |
|---|
| | | TradeOrderWs.orderEvent(client.getWebSocketClient(), tradeRequestParam); |
|---|
| | | log.info("价格变化触发量化操作: 账号={}, 价格={}", accountName, markPx); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } catch (Exception e) { |
|---|
| | | log.error("触发量化操作失败", e); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 构建 Redis Key |
|---|
| | | */ |
|---|
| | | private String buildRedisKey(String instId) { |
|---|
| | | return "PRICE_" + instId.replace("-", ""); |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 启动心跳检测任务。 |
|---|
| | | * 使用 ScheduledExecutorService 定期检查是否需要发送 ping 请求来维持连接。 |
|---|
| | | */ |
|---|
| | | private void startHeartbeat() { |
|---|
| | | if (heartbeatExecutor != null && !heartbeatExecutor.isTerminated()) { |
|---|
| | | heartbeatExecutor.shutdownNow(); |
|---|
| | | } |
|---|
| | | |
|---|
| | | heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(r -> { |
|---|
| | | Thread t = new Thread(r, "okx-newprice-heartbeat"); |
|---|
| | | t.setDaemon(true); |
|---|
| | | return t; |
|---|
| | | }); |
|---|
| | | |
|---|
| | | heartbeatExecutor.scheduleWithFixedDelay(this::checkHeartbeatTimeout, 25, 25, TimeUnit.SECONDS); |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 重置心跳计时器。 |
|---|
| | | * 当收到新消息或发送 ping 后取消当前超时任务并重新安排下一次超时检查。 |
|---|
| | | */ |
|---|
| | | private synchronized void resetHeartbeatTimer() { |
|---|
| | | cancelPongTimeout(); |
|---|
| | | |
|---|
| | | if (heartbeatExecutor != null && !heartbeatExecutor.isShutdown()) { |
|---|
| | | pongTimeoutFuture = heartbeatExecutor.schedule(this::checkHeartbeatTimeout, |
|---|
| | | HEARTBEAT_TIMEOUT, TimeUnit.SECONDS); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 检查心跳超时情况。 |
|---|
| | | * 若长时间未收到任何消息则主动发送 ping 请求保持连接活跃。 |
|---|
| | | */ |
|---|
| | | private void checkHeartbeatTimeout() { |
|---|
| | | // 只有在连接状态下才检查心跳 |
|---|
| | | if (!isConnected.get()) { |
|---|
| | | return; |
|---|
| | | } |
|---|
| | | |
|---|
| | | long currentTime = System.currentTimeMillis(); |
|---|
| | | long lastTime = lastMessageTime.get(); |
|---|
| | | |
|---|
| | | if (currentTime - lastTime >= HEARTBEAT_TIMEOUT * 1000L) { |
|---|
| | | sendPing(); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 发送 ping 请求至 WebSocket 服务端。 |
|---|
| | | * 用于维持长连接有效性。 |
|---|
| | | */ |
|---|
| | | private void sendPing() { |
|---|
| | | try { |
|---|
| | | if (webSocketClient != null && webSocketClient.isOpen()) { |
|---|
| | | JSONObject ping = new JSONObject(); |
|---|
| | | ping.put("op", "ping"); |
|---|
| | | webSocketClient.send(ping.toJSONString()); |
|---|
| | | log.debug("发送ping请求"); |
|---|
| | | } |
|---|
| | | } catch (Exception e) { |
|---|
| | | log.warn("发送ping失败", e); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 取消当前的心跳超时任务。 |
|---|
| | | * 在收到 pong 或其他有效消息时调用此方法避免不必要的断开重连。 |
|---|
| | | */ |
|---|
| | | private synchronized void cancelPongTimeout() { |
|---|
| | | if (pongTimeoutFuture != null && !pongTimeoutFuture.isDone()) { |
|---|
| | | pongTimeoutFuture.cancel(true); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 执行 WebSocket 重连操作。 |
|---|
| | | * 在连接意外中断后尝试重新建立连接。 |
|---|
| | | */ |
|---|
| | | private void reconnectWithBackoff() throws InterruptedException { |
|---|
| | | int attempt = 0; |
|---|
| | | int maxAttempts = 3; |
|---|
| | | long delayMs = 5000; |
|---|
| | | |
|---|
| | | while (attempt < maxAttempts) { |
|---|
| | | try { |
|---|
| | | Thread.sleep(delayMs); |
|---|
| | | connect(); |
|---|
| | | return; |
|---|
| | | } catch (Exception e) { |
|---|
| | | log.warn("第{}次重连失败", attempt + 1, e); |
|---|
| | | delayMs *= 2; |
|---|
| | | attempt++; |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | | log.error("超过最大重试次数({})仍未连接成功", maxAttempts); |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 优雅关闭线程池 |
|---|
| | | */ |
|---|
| | | private void shutdownExecutorGracefully(ExecutorService executor) { |
|---|
| | | if (executor == null || executor.isTerminated()) { |
|---|
| | | return; |
|---|
| | | } |
|---|
| | | try { |
|---|
| | | executor.shutdown(); |
|---|
| | | if (!executor.awaitTermination(5, TimeUnit.SECONDS)) { |
|---|
| | | executor.shutdownNow(); |
|---|
| | | } |
|---|
| | | } catch (InterruptedException e) { |
|---|
| | | Thread.currentThread().interrupt(); |
|---|
| | | executor.shutdownNow(); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } |
|---|
