盘口优化

heng.zhang1

2020-09-16 2abd0e479bd68cc377d77ac8178cc6b21783fcf1

盘口优化

 src/main/java/com/xcong/excoin/rabbit/consumer/ExchangeConsumer.java

@@ -49,7 +49,6 @@
     */
    @RabbitListener(queues = RabbitMqConfig.QUEUE_TRADE_PLATE)
    public void tradePlate(String content) {
        //log.info("#盘口信息消费者---->{}#", content);
        tradePlateSendWebSocket.sendMessagePlate("NEKK/USDT",content,null);
    }


 src/main/java/com/xcong/excoin/rabbit/producer/ExchangeProducer.java

@@ -24,6 +24,11 @@
        rabbitTemplate.setConfirmCallback(this);
    }

    /**
     * 发送盘口消息
     * @param content
     * @return void
     */
    public void sendPlateMsg(String content) {
        CorrelationData correlationData = new CorrelationData(IdUtil.simpleUUID());
        rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE_B, RabbitMqConfig.ROUTINGKEY_TRADE_PLATE, content, correlationData);

 src/main/java/com/xcong/excoin/trade/CoinTrader.java

@@ -527,7 +527,7 @@
     * @param buyTradePlate sellTradePlate
     */
    public void sendTradePlateMessage(TradePlate buyTradePlate, TradePlate sellTradePlate) {
        //防止并发引起数组越界，造成盘口倒挂 TODO
        //防止并发引起数组越界，造成盘口倒挂
        List<List<BigDecimal>> plate;
        List<BigDecimal> plateItem;
        TradePlateModel tradePlateModel = new TradePlateModel();
@@ -535,10 +535,10 @@
        if (buyTradePlate != null && buyTradePlate.getItems() != null) {
            plate = new ArrayList<>();
            LinkedList<TradePlateItem> items = buyTradePlate.getItems();
            for (TradePlateItem item : items) {
            for (int i = items.size() - 1; i >= 0; i--) {
                plateItem = new ArrayList<>(2);
                BigDecimal price = item.getPrice();
                BigDecimal amount = item.getAmount();
                BigDecimal price = items.get(i).getPrice();
                BigDecimal amount = items.get(i).getAmount();
                plateItem.add(price);
                plateItem.add(amount);
                plate.add(plateItem);
@@ -549,10 +549,10 @@
        if (sellTradePlate != null && sellTradePlate.getItems() != null) {
            plate = new ArrayList<>();
            LinkedList<TradePlateItem> items = sellTradePlate.getItems();
            for (TradePlateItem item : items) {
            for (int i = items.size() - 1; i >= 0; i--) {
                plateItem = new ArrayList<>(2);
                BigDecimal price = item.getPrice();
                BigDecimal amount = item.getAmount();
                BigDecimal price = items.get(i).getPrice();
                BigDecimal amount = items.get(i).getAmount();
                plateItem.add(price);
                plateItem.add(amount);
                plate.add(plateItem);
@@ -560,8 +560,7 @@
            tradePlateModel.setSell(plate);
        }

        // 盘口发生变化通知TODO

        // 盘口发生变化通知
        exchangeProducer.sendPlateMsg(JSON.toJSONString(tradePlateModel));
    }

@@ -571,7 +570,7 @@
     * @param
     */
    public String sendTradePlateMessage() {
        //防止并发引起数组越界，造成盘口倒挂 TODO
        //防止并发引起数组越界，造成盘口倒挂
        List<List<BigDecimal>> plate;
        List<BigDecimal> plateItem;
        TradePlateModel tradePlateModel = new TradePlateModel();
@@ -606,7 +605,6 @@

        // 盘口发生变化通知TODO
        return JSON.toJSONString(tradePlateModel);
        //exchangeProducer.sendPlateMsg(JSON.toJSONString(tradePlateModel));
    }

    /**

 src/main/java/com/xcong/excoin/websocket/TradePlateSendWebSocket.java

@@ -78,13 +78,10 @@
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        //log.info("服务端收到客户端[{}]的消息:{}", session.getId(), message);
        // 订阅方法 {sub: 'market.' + symbol + '.depth.' + this._caculateType(),id: symbol
        //}
        // 盘口订阅方法 {sub: 'market.btcusdt.depth.10,id: symbol}
        JSONObject jsonObject = JSON.parseObject(message);
        // 盘口的判断
        if (jsonObject.containsKey("sub") && jsonObject.get("sub").toString().contains("depth")) {
            //log.info("订阅盘口消息:{}", session.getId());
            String sub = jsonObject.get("sub").toString();
            String symbol = sub.split("\\.")[1];
            symbol = CoinTypeConvert.convert(symbol);
@@ -121,8 +118,7 @@
        }
        // 取消盘口订阅
        if (jsonObject.containsKey("unsub") && jsonObject.get("unsub").toString().contains("depth")) {
            // `market.${symbol}.kline.${strPeriod}
            //log.info("取消订阅盘口消息:{}", session.getId());
            // {unsub:'market.btcusdt.kline.1min'}
            String unsub = jsonObject.get("unsub").toString();
            String[] split = unsub.split("\\.");
            String symbol = split[1];
@@ -133,19 +129,11 @@
            }
        }


        // 最新K线订阅

        // 根据消息判断这个用户需要订阅哪种数据
        // {sub: `market.${symbol}.kline.${strPeriod}`,
        //            symbol: symbol,
        //            period: strPeriod
        //}
        // 取消订阅 {unsub: xxx(标识)}
        if (jsonObject.containsKey("sub") && jsonObject.get("sub").toString().contains("kline")) {
            // 订阅
            // {sub: 'market.btcusdt.kline.1min'}
            String sub = jsonObject.get("sub").toString();
            log.info("订阅最新K线消息:{}", sub);
            String[] split = sub.split("\\.");
            String symbol = split[1];
            symbol = CoinTypeConvert.convert(symbol);
@@ -154,26 +142,22 @@
                period = "1hour";
            }
            String key = symbol + "-" + period;
            log.info("最新K线key:{}", key);
            if (klineClients.containsKey(key)) {
                // 有这个币种K线
                Map<String, Session> stringSessionMap = klineClients.get(key);
                if (!stringSessionMap.containsKey(session.getId())) {
                    stringSessionMap.put(session.getId(), session);
                    log.info("放入最新K线Map:{}", key);
                }
            } else {
                Map<String, Session> stringSessionMap = new HashMap<>();
                stringSessionMap.put(session.getId(), session);
                klineClients.put(key, stringSessionMap);
                log.info("放入最新K线Map:{}", key);
            }
        }

        // 取消订阅
        if (jsonObject.containsKey("unsub") && jsonObject.get("unsub").toString().contains("kline")) {
            // `market.${symbol}.kline.${strPeriod}
            //log.info("取消订阅最新K消息:{}", session.getId());
            // {unsub:'market.${symbol}.kline.${strPeriod}'}
            String unsub = jsonObject.get("unsub").toString();
            String[] split = unsub.split("\\.");
            String strPeriod = split[3];
@@ -185,7 +169,6 @@
            String key = symbol + "-" + strPeriod;
            if (klineClients.containsKey(key)) {
                klineClients.get(key).remove(session.getId());
                //session.getAsyncRemote().sendText(message);
            }
        }

@@ -239,7 +222,7 @@
    }

    /**
     * 群发消息
     * 发送盘口消息
     *
     * @param message 消息内容
     */
@@ -249,38 +232,40 @@
            for (Map.Entry<String, Session> sessionEntry : nekk.entrySet()) {
                Session toSession = sessionEntry.getValue();
                // 排除掉自己
                //if (!fromSession.getId().equals(toSession.getId())) {
               // log.info("服务端给客户端[{}]发送盘口消息{}", toSession.getId(), message);
                boolean open = toSession.isOpen();
                if (open) {
                    toSession.getAsyncRemote().sendText(message);
                }

                //  }
            }
        }

    }

    /**
     * 发送历史K线
     * @param message, toSession
     * @return void
     */
    public void sendMessageHistory(String message, Session toSession) {
      //  log.info("服务端给客户端[{}]发送历史K线", toSession.getId());
        boolean open = toSession.isOpen();
        if (open) {
            toSession.getAsyncRemote().sendText(message);
        }
    }

    /**
     * 发送最新K线
     * @param symbol, period, message, fromSession
     * @return void
     */
    public void sendMessageKline(String symbol, String period, String message, Session fromSession) {

        String key = symbol + "-" + period;
        //log.info("发送最新K线[{}],数据[{}]",key,message);
        if (klineClients.containsKey(key)) {
            Map<String, Session> stringSessionMap = klineClients.get(key);
            for (Map.Entry<String, Session> sessionEntry : stringSessionMap.entrySet()) {
                Session toSession = sessionEntry.getValue();
                boolean open = toSession.isOpen();
                if (open) {
                    log.info("服务端给客户端[{}]发送最新K线消息{}", toSession.getId(), message);
                    toSession.getAsyncRemote().sendText(message);
                }
            }