20211208 fish

xiaoyong931011

2021-12-08 f5e6133809c553cfd9fb28ee61019927c547c374

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

@@ -4,6 +4,7 @@
import com.alibaba.fastjson.JSONObject;
import com.huobi.client.model.Candlestick;
import com.xcong.excoin.configurations.RabbitMqConfig;
import com.xcong.excoin.modules.coin.entity.OrderCoinsEntity;
import com.xcong.excoin.modules.coin.service.OrderCoinService;
import com.xcong.excoin.modules.exchange.service.HandleKlineService;
import com.xcong.excoin.trade.ExchangeTrade;
@@ -50,7 +51,8 @@
     */
    @RabbitListener(queues = RabbitMqConfig.QUEUE_TRADE_PLATE)
    public void tradePlate(String content) {
        tradePlateSendWebSocket.sendMessagePlate("ROC/USDT",content,null);
        //log.info("--发送盘口消息--");
        tradePlateSendWebSocket.sendMessagePlate("XCC/USDT",content,null);
    }

    /**
@@ -61,6 +63,9 @@
    public void handleTradeExchange(String content) {
       // log.info("#处理订单---->{}#", content);
        List<ExchangeTrade> exchangeTrades = JSONObject.parseArray(content, ExchangeTrade.class);
        if(CollectionUtils.isEmpty(exchangeTrades)){
            return;
        }
        // 去掉空的  暂时这样
        Iterator<ExchangeTrade> iterator = exchangeTrades.iterator();
        while (iterator.hasNext()){
@@ -71,49 +76,69 @@
        if(CollectionUtils.isEmpty(exchangeTrades)){
            return;
        }
        // 处理K线 并更新最新价
        handleKlineService.handleExchangeOrderToKline(exchangeTrades);
        // 推送最新K线
        String symbol = exchangeTrades.get(0).getSymbol();
        String symbolUsdt = symbol;
        if(!symbol.contains("USDT")){
            symbolUsdt = symbol+"/USDT";
        }
        String key = "NEW_KINE_{}";
        key = StrUtil.format(key, symbolUsdt);
        Object o = redisUtils.get(key);
        Map<String, Candlestick> currentKlineMap = (Map<String, Candlestick> )o;
        Set<Map.Entry<String, Candlestick>> entries = currentKlineMap.entrySet();


        for(Map.Entry<String, Candlestick> map : entries){
            String ch = "market.{}.kline.{}";
            Candlestick value = map.getValue();
            String key1 = map.getKey();
            String chKey = key1;
            if(key1.equals("1hour")){
                chKey = "60min";
            }
            // 转换
            NewCandlestick newCandlestick= new NewCandlestick();
            String nekkusdt = CoinTypeConvert.convertReverse(symbolUsdt);
            ch = StrUtil.format(ch, nekkusdt,chKey);
            newCandlestick.setCh(ch);
            CandlestickModel model = new CandlestickModel();
            model.setVol(value.getVolume());
            model.setLow(value.getLow());
            model.setOpen(value.getOpen());
            model.setHigh(value.getHigh());
            model.setCount(value.getCount());
            model.setAmount(value.getAmount());
            model.setId(value.getTimestamp()/1000);
            model.setTimestamp(value.getTimestamp()/1000);
            model.setClose(value.getClose());
            newCandlestick.setTick(model);
            tradePlateSendWebSocket.sendMessageKline(symbolUsdt,key1,JSONObject.toJSONString(newCandlestick),null);
        }
        // 处理用户订单
        // 先处理处理用户订单
        orderCoinService.handleOrder(exchangeTrades);
        try{
            // 处理K线 并更新最新价
            handleKlineService.handleExchangeOrderToKline(exchangeTrades);
            // 推送最新K线
            String symbol = exchangeTrades.get(0).getSymbol();
            String symbolUsdt = symbol;
            if(!symbol.contains("USDT")){
                symbolUsdt = symbol+"/USDT";
            }
            String key = "NEW_KINE_{}";
            key = StrUtil.format(key, symbolUsdt);
            Object o = redisUtils.get(key);
            Map<String, Candlestick> currentKlineMap = (Map<String, Candlestick> )o;
            Set<Map.Entry<String, Candlestick>> entries = currentKlineMap.entrySet();


            for(Map.Entry<String, Candlestick> map : entries){
                String ch = "market.{}.kline.{}";
                Candlestick value = map.getValue();
                String key1 = map.getKey();
                String chKey = key1;
                if(key1.equals("1hour")){
                    chKey = "60min";
                }
                // 转换
                NewCandlestick newCandlestick= new NewCandlestick();
                String nekkusdt = CoinTypeConvert.convertReverse(symbolUsdt);
                ch = StrUtil.format(ch, nekkusdt,chKey);
                newCandlestick.setCh(ch);
                CandlestickModel model = new CandlestickModel();
                model.setVol(value.getVolume());
                model.setLow(value.getLow());
                model.setOpen(value.getOpen());
                model.setHigh(value.getHigh());
                model.setCount(value.getCount());
                model.setAmount(value.getAmount());
                model.setId(value.getTimestamp()/1000);
                model.setTimestamp(value.getTimestamp()/1000);
                model.setClose(value.getClose());
                newCandlestick.setTick(model);
                tradePlateSendWebSocket.sendMessageKline(symbolUsdt,key1,JSONObject.toJSONString(newCandlestick),null);
            }

        }catch (Exception e){
            e.printStackTrace();
        }

    }

    /**
     *  撮合交易订单全部完成
     * @param content
     */
    @RabbitListener(queues = RabbitMqConfig.QUEUE_ROC_ORDER_COMPLETE)
    public void doComplete(String content) {
        log.debug("#完成的订单---->{}#", content);
        List<OrderCoinsEntity> exchangeTrades = JSONObject.parseArray(content, OrderCoinsEntity.class);
        if(CollectionUtils.isEmpty(exchangeTrades)){
            return;
        }
        orderCoinService.completeOrder(exchangeTrades);
    }

}