今天我们来说说我在做Java后端对接币安区块链时，遇到的问题及解决方式。

既然要对接币安区块链K线接口，我们首先必须先了解这个行情api在哪里？

1. 行情K线接口

https://binance-docs.github.io/apidocs/futures/cn/#k-8

注意：这里一定要是连续合约K线，而不是K线接口。这是在看官网K线socket接口对比得到的。

如果使用K线Api，数据是不正确的。

而且有一点我需要表述一下，不能说网络上其他博主的步骤不对，应该是版本升级的问题，其他博主的教程已经无法使用。

2. Java实现websockt客户端

Java实现websocket其实有很多种方式，比如：javax.websocket Java标准库、再如OkHttp、Apache HttpClient还有一个开源的库Java-WebSocket等等

Java-WebSocket在我碰到这个库之后，我基本都没用过其实的库，假如它适用我的业务，必须用它，因为它简单、高效。

来写代码了：

pom.xml先引入Java-WebSocket

   org.java-websocket   Java-WebSocket   1.5.4 

然后我们写一个websocket的工具类。

/**  * 合约站行情推送处理器  */ @Slf4j public class WssMarketHandle implements Cloneable {     private WebSocketClient webSocketClient;     // 合约站行情请求以及订阅地址     private String pushUrl = "";          private WssMarketHandle() {     }      public WssMarketHandle(String pushUrl) {         this.pushUrl = pushUrl;     }     // .... }

由于K线数据时序要实时去接收的，所以我这里需要它连接之后，就让他去订阅接口，所以我的代码只写一个订阅的外部方式:

public void sub(List channels, SubscriptionListener callback) throws URISyntaxException {     doConnect(channels, callback); }

​​​​ List channels 多币种订阅的模式参数

SubscriptionListener callback 回调方法类

来看下doConnect方法，他是实现是主体

private void doConnect(List channels, SubscriptionListener callback) throws URISyntaxException {     webSocketClient = new WebSocketClient(new URI(pushUrl)) {         @Override         public void onOpen(ServerHandshake serverHandshake) {             logger.debug("onOpen Success");             doSub(channels);             dealReconnect();             }         @Override         public void onMessage(String socketJson) {             fixedThreadPool.execute(() -> {                 try {                     JSONObject entries = JSONUtil.parseObj(socketJson);                     String ping = entries.getStr("ping");                     String stream = entries.getStr("stream");                     if (StrUtil.isNotEmpty(ping)) {                         log.info("WssMarketHandle onMessage stream: " + socketJson);                         dealPing();                     } else if (StrUtil.isNotEmpty(stream)) {                          callback.onReceive(socketJson);                     } else {                         log.info("WssMarketHandle onMessage other: " + socketJson);                     }                 } catch (Throwable e) {                     logger.error("onMessage异常", e);                 }             });             }         @Override         public void onMessage(ByteBuffer bytes) {             fixedThreadPool.execute(() -> {                 try {                     byte[] temp = bytes.array();                     // gzip 解压                     byte[] decompress = GZipUtils.decompress(temp);                     String socketJson = new String(decompress, StandardCharsets.UTF_8);                     JSONObject JSONMessage = JSONUtil.parseObj(socketJson);                     String ch = JSONMessage.getStr("ch");                     String ping = JSONMessage.getStr("ping");                     if (StrUtil.isNotEmpty(ch)) {                         callback.onReceive(socketJson);                     }                     if (StrUtil.isNotEmpty(ping)) {                         dealPing();                     }                 } catch (Throwable e) {                     logger.error("onMessage异常", e);                 }             });             }         @Override         public void onClose(int i, String s, boolean b) {             logger.error("onClose i:{},s:{},b:{}", i, s, b);             }         @Override         public void onError(Exception e) {             logger.error("onError:", e);         }         };         webSocketClient.connect(); }

其他的方法，包含取消订阅、断开重连以及Ping处理：

public void close() {   webSocketClient.connect(); }  /** * 开始订阅 * @param channels */ private void doSub(List channels) {   try {       if (IterUtil.isNotEmpty(channels)) {           channels.stream().forEach(e -> {               webSocketClient.send(e);           });       }   } catch (Exception ex) {   } }  /** * 取消订阅 * @param subStr */ public void unSub(String subStr) {   try {       webSocketClient.send(subStr);   } catch (Exception ex) {   } }  /** * Ping处理 */ private void dealPing() {   try {       JSONObject jsonMessage = new JSONObject();       jsonMessage.put("pong", pong.incrementAndGet());       logger.debug("发送pong:{}", jsonMessage.toString());       webSocketClient.send(jsonMessage.toString());   } catch (Throwable t) {       logger.error("dealPing出现了异常");   } }  /** * 重连 */ private void dealReconnect() {   try {       scheduledExecutorService.scheduleAtFixedRate(() -> {           try {               if ((webSocketClient.isClosed() && !webSocketClient.isClosing())) {                   logger.error("isClosed:{},isClosing:{}，准备重连", webSocketClient.isClosed(), webSocketClient.isClosing());                   Boolean reconnectResult = webSocketClient.reconnectBlocking();                   logger.error("重连的结果为：{}", reconnectResult);                   if (!reconnectResult) {                       webSocketClient.closeBlocking();                       logger.error("closeBlocking");                   }                }           } catch (Throwable e) {               logger.error("dealReconnect异常", e);           }        }, 60, 10, TimeUnit.SECONDS);   } catch (Exception e) {       logger.error("dealReconnect scheduledExecutorService异常", e);   } }

​​​​​​​ 至此整个开发过程已然明了。

调用websockt接口实现，写个main方法测试下。

private static final String BIAN_WS_URL = "wss://fstream.binance.com/stream";  public static void main(String[] args) throws Exception {     WssMarketHandle marketHandle = new WssMarketHandle(BIAN_WS_URL);     List channels = new ArrayList<>();     JSONObject object = new JSONObject();     object.set("method", "SUBSCRIBE");     object.set("id", RandomUtil.randomInt(99999));      JSONArray params = new JSONArray();     String pair = ExchangeEnum.BTC.getExchangeName().toLowerCase() + "usdt";     String perpetual = "perpetual";     String urlParam = StrUtil.format("{}_{}@continuousKline_1m", pair, perpetual);     params.add(urlParam);     object.set("params", params);      channels.add(object.toString());     marketHandle.sub(channels, log::info); }

​​​​​​​历史K线我找了好久，都没有适用的，所以使用的是接口方式，因为它本来就是一次性获取的，所以不需要socket长连接获取也是可行的。接口如下：

https://fapi.binance.com/fapi/v1/continuousKlines

有需要帮助的小伙伴，可以关注，联系我，谢谢！