修复冻结问题

zainali5120

2020-10-08 0f25ef5fe5076be73cf7d40b2f0f9c3f7293eadb

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

@@ -7,9 +7,13 @@
import com.xcong.excoin.modules.coin.service.OrderCoinService;
import com.xcong.excoin.modules.exchange.service.HandleKlineService;
import com.xcong.excoin.trade.ExchangeTrade;
import com.xcong.excoin.utils.CoinTypeConvert;
import com.xcong.excoin.utils.RedisUtils;
import com.xcong.excoin.websocket.CandlestickModel;
import com.xcong.excoin.websocket.NewCandlestick;
import com.xcong.excoin.websocket.TradePlateSendWebSocket;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

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

    /**
@@ -55,7 +58,7 @@
     */
    @RabbitListener(queues = RabbitMqConfig.QUEUE_HANDLE_TRADE)
    public void handleTradeExchange(String content) {
        log.info("#---->{}#", content);
       // log.info("#处理订单---->{}#", content);
        List<ExchangeTrade> exchangeTrades = JSONObject.parseArray(content, ExchangeTrade.class);
        // 去掉空的  暂时这样
        Iterator<ExchangeTrade> iterator = exchangeTrades.iterator();
@@ -63,6 +66,9 @@
            if(iterator.next()==null){
                iterator.remove();
            }
        }
        if(CollectionUtils.isEmpty(exchangeTrades)){
            return;
        }
        // 处理K线 并更新最新价
        handleKlineService.handleExchangeOrderToKline(exchangeTrades);
@@ -77,8 +83,33 @@
        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){
            tradePlateSendWebSocket.sendMessageKline(symbolUsdt,map.getKey(),JSONObject.toJSONString(map.getValue()),null);
            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);