From f5e6133809c553cfd9fb28ee61019927c547c374 Mon Sep 17 00:00:00 2001
From: xiaoyong931011 <15274802129@163.com>
Date: Wed, 08 Dec 2021 15:58:33 +0800
Subject: [PATCH] 20211208 fish
---
src/main/java/com/xcong/excoin/rabbit/consumer/ExchangeConsumer.java | 111 ++++++++++++++++++++++++++++++++++---------------------
1 files changed, 68 insertions(+), 43 deletions(-)
diff --git a/src/main/java/com/xcong/excoin/rabbit/consumer/ExchangeConsumer.java b/src/main/java/com/xcong/excoin/rabbit/consumer/ExchangeConsumer.java
index 2ed81da..3cdf0bb 100644
--- a/src/main/java/com/xcong/excoin/rabbit/consumer/ExchangeConsumer.java
+++ b/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);
}
}
--
Gitblit v1.9.1