package com.xcong.excoin.rabbit.consumer;
|
|
import cn.hutool.core.util.StrUtil;
|
import com.alibaba.fastjson.JSONObject;
|
import com.huobi.client.model.Candlestick;
|
import com.xcong.excoin.configurations.RabbitMqConfig;
|
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.RedisUtils;
|
import com.xcong.excoin.websocket.TradePlateSendWebSocket;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
import org.springframework.stereotype.Component;
|
|
import javax.annotation.Resource;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.Set;
|
|
/**
|
* @author wzy
|
* @date 2020-05-25
|
**/
|
@Slf4j
|
@Component
|
public class ExchangeConsumer {
|
|
@Resource
|
private TradePlateSendWebSocket tradePlateSendWebSocket;
|
|
@Resource
|
private RedisUtils redisUtils;
|
|
@Resource
|
private HandleKlineService handleKlineService;
|
|
@Resource
|
private OrderCoinService orderCoinService;
|
|
/**
|
* 发送盘口信息
|
* @param content
|
*/
|
@RabbitListener(queues = RabbitMqConfig.QUEUE_TRADE_PLATE)
|
public void tradePlate(String content) {
|
log.info("#---->{}#", content);
|
tradePlateSendWebSocket.sendMessagePlate(content,null);
|
}
|
|
/**
|
* 处理订单
|
* @param content
|
*/
|
@RabbitListener(queues = RabbitMqConfig.QUEUE_HANDLE_TRADE)
|
public void handleTradeExchange(String content) {
|
log.info("#---->{}#", content);
|
List<ExchangeTrade> exchangeTrades = JSONObject.parseArray(content, ExchangeTrade.class);
|
// 处理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){
|
tradePlateSendWebSocket.sendMessageKline(symbolUsdt,map.getKey(),JSONObject.toJSONString(map.getValue()),null);
|
}
|
// 处理用户订单
|
orderCoinService.handleOrder(exchangeTrades);
|
}
|
|
/**
|
* 更新最新K线
|
* @param content
|
*/
|
// @RabbitListener(queues = RabbitMqConfig.QUEUE_TRADE_PLATE)
|
// public void newKling(String content) {
|
// log.info("#---->{}#", content);
|
// // 最新K线的币种
|
// String key = "NEW_KINE_{}";
|
// key = StrUtil.format(key, content);
|
// Object o = redisUtils.get(key);
|
// Map<String, Candlestick> currentKlineMap = (Map<String, Candlestick>)o;
|
// // 推送最新K线
|
// Set<Map.Entry<String, Candlestick>> entries = currentKlineMap.entrySet();
|
// for(Map.Entry<String, Candlestick> map : entries){
|
// tradePlateSendWebSocket.sendMessageKline(content,map.getKey(),JSONObject.toJSONString(map.getValue()),null);
|
// }
|
//
|
// }
|
}
|
