package org.example.bitgetsclient.wsClient; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.reflect.TypeToken; import com.google.gson.Gson; import com.google.gson.JsonSyntaxException; import lombok.extern.slf4j.Slf4j; import org.example.bitgetsclient.util.RedisUtil; import org.json.JSONException; import org.springframework.context.annotation.Bean; import org.springframework.util.CollectionUtils; import javax.websocket.*; import java.io.IOException; import java.lang.reflect.Type; import java.math.BigDecimal; import java.net.URI; import java.nio.ByteBuffer; import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @ClientEndpoint @Slf4j public class BitgetClient { private static final String WS_ENDPOINT = "wss://ws.bitget.com/v2/ws/public"; // WebSocket 接口地址 private static final long PING_INTERVAL = 20000; // 心跳间隔，单位毫秒 private static final String PING_MESSAGE = "ping"; // 心跳消息 private static final int MAX_RECONNECT_ATTEMPTS = 5; // 最大重连次数 private final String subscriptions; // 订阅内容 private final ScheduledExecutorService executorService; // 定义调度任务执行服务 private Session session; // WebSocket 会话 private final Object lock = new Object(); // 使用锁对象以确保线程安全 private boolean reconnecting = false; // 重连状态 private int reconnectAttempts = 0; // 当前重连次数 // 静态单例 ObjectMapper 实例，避免重复创建 private static final ObjectMapper objectMapper = new ObjectMapper(); private static final Gson gson = new Gson(); // 静态 Gson 实例 // 构造方法，初始化订阅内容和调度服务 public BitgetClient(String subscriptions) { this.subscriptions = subscriptions; // 初始化订阅内容 this.executorService = Executors.newScheduledThreadPool(1); // 初始化调度线程池 } public void start() { try { connect(); // 尝试连接 if (session == null) { log.info("无法在超时内连接到服务器。"); // 记录连接失败的信息 return; // 如果连接失败，直接返回 } // 定期发送心跳 executorService.scheduleAtFixedRate(this::sendPing, PING_INTERVAL, PING_INTERVAL, TimeUnit.MILLISECONDS); // 发送订阅消息 session.getBasicRemote().sendText(subscriptions); // 发送订阅信息 synchronized (this) { this.wait(); // 等待 WebSocket 消息到来 } } catch (InterruptedException e) { Thread.currentThread().interrupt(); // 恢复中断状态 log.error("线程被中断: " + e.getMessage(), e); // 记录线程中断的错误 } catch (Exception e) { log.error("bitget ws 连接过程中发生异常: " + e.getMessage(), e); // 记录连接过程中发生的异常 } finally { executorService.shutdownNow(); // 尝试立即关闭调度服务 } } private void connect() throws Exception { WebSocketContainer container = ContainerProvider.getWebSocketContainer(); container.connectToServer(this, new URI(WS_ENDPOINT)); // 连接 WebSocket 服务 } @OnOpen public void onOpen(Session session) { log.info("bitget ws 已连接到服务器。"); // 记录连接成功的信息 this.session = session; // 存储会话 synchronized (this) { this.notify(); // 通知等待的线程 } reconnectAttempts = 0; // 连接成功重置重连次数 } @OnMessage public void onMessage(String message) { try { // 将消息解析为 Map Map map = parseMessage(message); // 调用共用方法解析 if (map != null && map.containsKey("arg") && map.containsKey("data")) { // 确保 map 不为 null 且包含必要的键 JsonNode dataNode = getDataNode(message); // 获取数据节点 // 确保 dataNode 不为 null，以避免空指针异常 if (dataNode != null) { Map hashMap = new HashMap<>(); // 变量命名更具描述性 String bids = dataNode.get("bids").toString(); String asks = dataNode.get("asks").toString(); Type listType = new TypeToken>>(){}.getType(); List> bidsList = gson.fromJson(bids, listType); List> asksList = gson.fromJson(asks, listType); if(!bidsList.isEmpty() && !bidsList.get(0).isEmpty() && !asksList.isEmpty() && !asksList.get(0).isEmpty()){ if (!bidsList.isEmpty() && !bidsList.get(0).isEmpty()) { List bidsStringList = bidsList.get(0); HashMap pvMap = new HashMap<>(); // 创建新的 HashMap 保存 bids 和 asks pvMap.put("p", new BigDecimal(bidsStringList.get(0)).toPlainString()); pvMap.put("v", new BigDecimal(bidsStringList.get(1)).toPlainString()); hashMap.put("bids", pvMap); // 获取并存储 bids } if (!asksList.isEmpty() && !asksList.get(0).isEmpty()) { List asksStringList = asksList.get(0); // 修正此处为 asksList HashMap pvMap = new HashMap<>(); // 创建新的 HashMap 保存 bids 和 asks pvMap.put("p", new BigDecimal(asksStringList.get(0)).toPlainString()); pvMap.put("v", new BigDecimal(asksStringList.get(1)).toPlainString()); hashMap.put("asks", pvMap); // 获取并存储 asks } Map argMap = gson.fromJson(map.get("arg").toString(), new TypeToken>() {}.getType()); // 解析 arg 为 Map String key = "bitget" + argMap.get("instId"); // 构建 Redis 存储的键 // 存储到 Redis，使用 ObjectMapper 转换为 JSON 字符串 String jsonData = objectMapper.writeValueAsString(hashMap); // 先将 HashMap 转换为 JSON 字符串 RedisUtil.set(key, jsonData); // 存储数据 } } } } catch (JsonSyntaxException e) { log.error("JSON 解析错误: " + e.getMessage(), e); // 记录 JSON 解析错误 } catch (IOException e) { log.error("对象转换时发生 I/O 异常: " + e.getMessage(), e); // 记录对象转换 I/O 异常 } catch (Exception e) { log.error("处理消息时发生异常: " + e.getMessage(), e); // 记录处理消息时发生的异常 } } // 解析消息的共用方法 private Map parseMessage(String message) { if(!message.equals("pong")){ return gson.fromJson(message, new TypeToken>() {}.getType()); // 将消息解析为 Map } return null; } // 获取数据节点的共用方法 private JsonNode getDataNode(String message) throws JsonProcessingException { JsonNode jsonNode = objectMapper.readTree(message); // 使用静态 ObjectMapper，避免重复创建 return jsonNode.get("data").get(0); // 获取第一个数据节点 } @OnClose public void onClose() { log.info("bitget ws 连接已关闭，尝试重新连接..."); // 记录连接关闭的信息 handleConnectionClosedOrError(); // 处理连接关闭事件 } @OnError public void onError(Throwable throwable) { log.error("bitget ws 发生错误: " + throwable.getMessage(), throwable); // 记录错误信息 handleConnectionClosedOrError(); // 处理错误事件 } private void handleConnectionClosedOrError() { synchronized (lock) { if (!reconnecting) { reconnecting = true; // 开始重连 executorService.execute(this::attemptReconnect); // 执行重连操作 } } } private void attemptReconnect() { if (reconnectAttempts < MAX_RECONNECT_ATTEMPTS) { // 检查重连次数是否超过限制 try { log.info("bitget ws 开始重连"); // 记录开始重连的信息 connect(); // 尝试重连 log.info("bitget ws 重连成功"); // 成功重连的日志 reconnectAttempts = 0; // 重连成功，重置重连次数 } catch (Exception e) { reconnectAttempts++; // 增加重连尝试次数 log.warn("bitget ws 重连失败，尝试次数: " + reconnectAttempts, e); // 记录重连失败的警告信息 // 采用指数退避策略，增加重连间隔 long waitTime = Math.min(60, (long) Math.pow(2, reconnectAttempts)) * 1000; // 最大等待时间为 60 秒 scheduleReconnect(waitTime); // 调度下次重连 } } else { log.error("超过最大重连次数，停止重连"); // 超过最大重连次数 reconnecting = false; // 重连状态重置 } } private void scheduleReconnect(long waitTime) { // 接受等待时间参数 if (!executorService.isShutdown()) { executorService.schedule(this::attemptReconnect, waitTime, TimeUnit.MILLISECONDS); // 调度重连 } } private void sendPing() { try { if (session != null) { session.getBasicRemote().sendText(PING_MESSAGE); // 发送心跳消息 } } catch (Exception e) { log.error("发送心跳失败", e); // 记录发送心跳失败的错误 } } public String getParameter(String symbol) throws JsonProcessingException { // 使用 ObjectMapper 构建 JSON 请求 Map jsonMap = new HashMap<>(); jsonMap.put("op", "subscribe"); // 设置操作类型 Map argsMap = new HashMap<>(); jsonMap.put("args", argsMap); // 参数放入请求 // 设置请求参数 argsMap.put("instType", "SPOT"); // 交易类型 argsMap.put("channel", "books15"); // 渠道类型 argsMap.put("instId", symbol); // 交易对符号 // 返回 JSON 字符串 return objectMapper.writeValueAsString(jsonMap); // 使用单一的 ObjectMapper 实例 } }