From 2abd0e479bd68cc377d77ac8178cc6b21783fcf1 Mon Sep 17 00:00:00 2001
From: heng.zhang1 <512061637@qq.com>
Date: Wed, 16 Sep 2020 16:54:24 +0800
Subject: [PATCH] 盘口优化
---
src/main/java/com/xcong/excoin/rabbit/producer/ExchangeProducer.java | 5 ++
src/main/java/com/xcong/excoin/websocket/TradePlateSendWebSocket.java | 45 +++++++---------------
src/main/java/com/xcong/excoin/rabbit/consumer/ExchangeConsumer.java | 1
src/main/java/com/xcong/excoin/trade/CoinTrader.java | 20 ++++-----
4 files changed, 29 insertions(+), 42 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 df5900f..cf45f32 100644
--- a/src/main/java/com/xcong/excoin/rabbit/consumer/ExchangeConsumer.java
+++ b/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);
}
diff --git a/src/main/java/com/xcong/excoin/rabbit/producer/ExchangeProducer.java b/src/main/java/com/xcong/excoin/rabbit/producer/ExchangeProducer.java
index 0d08c61..5d6e4f6 100644
--- a/src/main/java/com/xcong/excoin/rabbit/producer/ExchangeProducer.java
+++ b/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);
diff --git a/src/main/java/com/xcong/excoin/trade/CoinTrader.java b/src/main/java/com/xcong/excoin/trade/CoinTrader.java
index fc4a8fc..fd62c5f 100644
--- a/src/main/java/com/xcong/excoin/trade/CoinTrader.java
+++ b/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));
}
/**
diff --git a/src/main/java/com/xcong/excoin/websocket/TradePlateSendWebSocket.java b/src/main/java/com/xcong/excoin/websocket/TradePlateSendWebSocket.java
index 15d4168..fc1ccab 100644
--- a/src/main/java/com/xcong/excoin/websocket/TradePlateSendWebSocket.java
+++ b/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);
}
}
--
Gitblit v1.9.1