批量订阅

market/src/main/java/com/market/client/BinanceWebSocketClient.java

@@ -70,14 +70,57 @@ public class BinanceWebSocketClient {
         return channel != null && channel.isActive();
     }
 
-    /** 首次连接：以初始 streams 拼入 URL 建立连接 */
+    /** 单次 SUBSCRIBE 消息携带的最大 stream 数，避免单条消息过大 */
+    private static final int SUBSCRIBE_BATCH_SIZE = 100;
+
+    /** 分批 SUBSCRIBE 之间的间隔毫秒数，规避币安 10 条/秒 的消息频率限制 */
+    private static final long SUBSCRIBE_BATCH_INTERVAL_MS = 200L;
+
+    /**
+     * 首次连接：仅用第一个 stream 引导建立连接，握手成功后通过 SUBSCRIBE 分批补订阅其余 stream。
+     *
+     * <p>不再将全部 stream 拼入 URL，避免 stream 过多导致 URL 超长触发 414 Request-URI Too Large。
+     *
+     * @param streams 需要订阅的全部 stream 列表
+     */
     public void connect(List<String> streams) {
+        // 1. 校验入参
         if (streams == null || streams.isEmpty()) {
             log.warn("没有要订阅的 stream");
             return;
         }
-        String url = "wss://fstream.binance.com/market/stream?streams=" + String.join("/", streams);
+
+        // 2. 仅用第一个 stream 引导建立连接，保证 URL 短小
+        String url = "wss://fstream.binance.com/market/stream?streams=" + streams.get(0);
         connectInternal(url);
+
+        // 3. 握手成功后分批补订阅其余 stream
+        if (isConnected() && streams.size() > 1) {
+            subscribeInBatches(streams.subList(1, streams.size()));
+        }
+    }
+
+    /**
+     * 将 stream 列表按 {@link #SUBSCRIBE_BATCH_SIZE} 分批发送 SUBSCRIBE 消息
+     *
+     * @param streams 待订阅的 stream 列表
+     */
+    private void subscribeInBatches(List<String> streams) {
+        // 按批次切分并逐批 SUBSCRIBE
+        for (int i = 0; i < streams.size(); i += SUBSCRIBE_BATCH_SIZE) {
+            int end = Math.min(i + SUBSCRIBE_BATCH_SIZE, streams.size());
+            subscribe(streams.subList(i, end));
+
+            // 批次之间稍作停顿，规避消息频率限制
+            if (end < streams.size()) {
+                try {
+                    Thread.sleep(SUBSCRIBE_BATCH_INTERVAL_MS);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    break;
+                }
+            }
+        }
     }
 
     private void connectInternal(String url) {

market/src/main/java/com/market/manager/BinanceSpotWebSocketClient.java

@@ -10,6 +10,7 @@ import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.http.DefaultHttpHeaders;
 import io.netty.handler.codec.http.HttpClientCodec;
 import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
 import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
 import io.netty.handler.codec.http.websocketx.WebSocketVersion;
@@ -20,8 +21,11 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.URI;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * 现货 Binance WebSocket 客户端（Netty 实现）
@@ -37,9 +41,16 @@ public class BinanceSpotWebSocketClient {
     private static final Logger log = LoggerFactory.getLogger(BinanceSpotWebSocketClient.class);
     private static final String COMBINED_BASE_URL = "wss://stream.binance.com:443/stream?streams=";
 
+    /** 单次 SUBSCRIBE 消息携带的最大 stream 数，避免单条消息过大 */
+    private static final int SUBSCRIBE_BATCH_SIZE = 100;
+
+    /** 分批 SUBSCRIBE 之间的间隔毫秒数，规避币安 10 条/秒 的消息频率限制 */
+    private static final long SUBSCRIBE_BATCH_INTERVAL_MS = 200L;
+
     /** 现货行情处理服务 */
     private final SpotDataProcessService dataProcessService;
     private final ObjectMapper objectMapper = new ObjectMapper();
+    private final AtomicInteger reqId = new AtomicInteger(1);
 
     private NioEventLoopGroup group;
     private Channel channel;
@@ -49,17 +60,73 @@ public class BinanceSpotWebSocketClient {
     }
 
     /**
-     * 连接合并流
+     * 连接合并流：仅用第一个 stream 引导建立连接，握手成功后通过 SUBSCRIBE 分批补订阅其余 stream。
+     *
+     * <p>不再将全部 stream 拼入 URL，避免 stream 过多导致 URL 超长触发 414 Request-URI Too Large。
      *
      * @param streams 流列表（如 btcusdt@kline_1m）
      */
     public void connect(List<String> streams) {
+        // 1. 校验入参
         if (streams == null || streams.isEmpty()) {
             log.warn("现货 stream 列表为空，跳过连接");
             return;
         }
-        String url = COMBINED_BASE_URL + String.join("/", streams);
+
+        // 2. 仅用第一个 stream 引导建立连接，保证 URL 短小
+        String url = COMBINED_BASE_URL + streams.get(0);
         connectInternal(url, streams.size());
+
+        // 3. 握手成功后分批补订阅其余 stream
+        if (isConnected() && streams.size() > 1) {
+            subscribeInBatches(streams.subList(1, streams.size()));
+        }
+    }
+
+    /**
+     * 将 stream 列表按 {@link #SUBSCRIBE_BATCH_SIZE} 分批发送 SUBSCRIBE 消息
+     *
+     * @param streams 待订阅的 stream 列表
+     */
+    private void subscribeInBatches(List<String> streams) {
+        // 按批次切分并逐批 SUBSCRIBE
+        for (int i = 0; i < streams.size(); i += SUBSCRIBE_BATCH_SIZE) {
+            int end = Math.min(i + SUBSCRIBE_BATCH_SIZE, streams.size());
+            sendCommand("SUBSCRIBE", streams.subList(i, end));
+
+            // 批次之间稍作停顿，规避消息频率限制
+            if (end < streams.size()) {
+                try {
+                    Thread.sleep(SUBSCRIBE_BATCH_INTERVAL_MS);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    break;
+                }
+            }
+        }
+    }
+
+    /**
+     * 向已建立的连接发送订阅/取消订阅命令
+     *
+     * @param method SUBSCRIBE 或 UNSUBSCRIBE
+     * @param params stream 列表
+     */
+    private void sendCommand(String method, List<String> params) {
+        // 连接不存活或参数为空时直接跳过
+        if (!isConnected() || params == null || params.isEmpty()) {
+            return;
+        }
+        try {
+            Map<String, Object> msg = new LinkedHashMap<>();
+            msg.put("method", method);
+            msg.put("params", params);
+            msg.put("id", reqId.getAndIncrement());
+            channel.writeAndFlush(new TextWebSocketFrame(objectMapper.writeValueAsString(msg)));
+            log.info("现货 {} {} 个 stream", method, params.size());
+        } catch (Exception e) {
+            log.error("现货发送 {} 失败", method, e);
+        }
     }
 
     /**