| src/main/java/com/nq/enums/EStockType.java | ●●●●● patch | view | raw | blame | history | |
| src/main/java/com/nq/service/impl/StockServiceImpl.java | ●●●●● patch | view | raw | blame | history | |
| src/main/java/com/nq/utils/task/stock/StockTask.java | ●●●●● patch | view | raw | blame | history | |
| src/main/java/com/nq/ws/WebSocketClientBeanConfig.java | ●●●●● patch | view | raw | blame | history | |
| src/main/java/com/nq/ws/WebsocketRunClient.java | ●●●●● patch | view | raw | blame | history | |
| src/main/java/com/nq/ws/WsClientConfig.java | ●●●●● patch | view | raw | blame | history | |
| src/main/resources/application.properties | ●●●●● patch | view | raw | blame | history |
src/main/java/com/nq/enums/EStockType.java
@@ -15,7 +15,7 @@ HK("HK","香港股票","39",PropertiesUtil.getProperty("HK_HTTP_API"),PropertiesUtil.getProperty("HK_KEY"),"HKD","HK$"), MAS("MAS","马来西亚股票","42",PropertiesUtil.getProperty("MAS_HTTP_API"),PropertiesUtil.getProperty("MAS_KEY"),"MYR","RM"), IN("IN","印度股票","14", PropertiesUtil.getProperty("IN_HTTP_API"),PropertiesUtil.getProperty("JS_IN_KEY"),"INR","₹"); IN("IN","印度股票","14", PropertiesUtil.getProperty("JS_IN_HTTP_URL"),PropertiesUtil.getProperty("JS_IN_KEY"),"INR","₹"); // TH("TH","泰国股票","41",PropertiesUtil.getProperty("TH_HTTP_API"),PropertiesUtil.getProperty("TH_KEY")), // HG("HG","韩国股票","11",PropertiesUtil.getProperty("HG_HTTP_API"),PropertiesUtil.getProperty("HG_KEY")), // SZHB("SZHB","数字货币","41",PropertiesUtil.getProperty("SZHB_HTTP_API"),PropertiesUtil.getProperty("SZHB_KEY")); src/main/java/com/nq/service/impl/StockServiceImpl.java
@@ -453,25 +453,21 @@ @Override public Object getKData(String pid, String interval, String stockType) { EStockType eStockType = EStockType.getEStockTypeByCode(stockType); if(eStockType == EStockType.IN){ Object object = HttpUtil.get(eStockType.stockUrl + "api/all/getKData.do?pid=" + pid + "&interval=" + interval + "&stockType=in"); Gson gson = new Gson(); List<kData> dataList = gson.fromJson(object.toString(), new TypeToken<List<kData>>(){}.getType()); Object object = HttpUtil.get(eStockType.stockUrl + "kline?pid=" + pid + "&interval=" + interval + "&key=" + eStockType.stockKey); Gson gson = new Gson(); List<kData> dataList = gson.fromJson(object.toString(), new TypeToken<List<kData>>(){}.getType()); Stock stock = stockMapper.selectOne(new LambdaQueryWrapper<Stock>().eq(Stock::getStockCode, pid).eq(Stock::getStockType, "IN")); BigDecimal nowPrice = iPriceServices.getNowPrice(stock.getStockCode()); Map singleStock = getSingleStock(stock.getStockCode()); StockVO stockVO = (StockVO)singleStock.get("stock"); // 修改 List 中的最后一条数据 kData lastData = dataList.get(dataList.size() - 1); lastData.setC(nowPrice.toString()); lastData.setO(stockVO.getOpen_px()); lastData.setH(stockVO.getToday_max()); lastData.setL(stockVO.getToday_min()); return gson.toJson(dataList); } return HttpUtil.get(eStockType.stockUrl + "kline?pid=" + pid + "&interval=" + interval + "&key=" + eStockType.stockKey); Stock stock = stockMapper.selectOne(new LambdaQueryWrapper<Stock>().eq(Stock::getStockCode, pid).eq(Stock::getStockType, "IN")); BigDecimal nowPrice = iPriceServices.getNowPrice(stock.getStockCode()); Map singleStock = getSingleStock(stock.getStockCode()); StockVO stockVO = (StockVO)singleStock.get("stock"); // 修改 List 中的最后一条数据 kData lastData = dataList.get(dataList.size() - 1); lastData.setC(nowPrice.toString()); lastData.setO(stockVO.getOpen_px()); lastData.setH(stockVO.getToday_max()); lastData.setL(stockVO.getToday_min()); return gson.toJson(dataList); } @Override src/main/java/com/nq/utils/task/stock/StockTask.java
@@ -61,7 +61,7 @@ /** * 同步系统所需要的股票 */ @Scheduled(cron = "0/5 * * * * ?") @Scheduled(cron = "0 0/30 * * * ?") public void syncINStockData() { if (syncINStockData.get()) { // 判断任务是否在处理中 return; src/main/java/com/nq/ws/WebSocketClientBeanConfig.java
@@ -1,48 +1,48 @@ package com.nq.ws; import com.nq.enums.EStockType; import com.nq.utils.PropertiesUtil; import lombok.extern.slf4j.Slf4j; import org.java_websocket.client.WebSocketClient; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.net.URI; import java.util.HashMap; import java.util.Map; @Slf4j @Configuration public class WebSocketClientBeanConfig { @Bean public Map<String, WebSocketClient> websocketRunClientMap() { Map<String, WebSocketClient> retMap = new HashMap<>(2); try { WebsocketRunClient websocketRunClient = new WebsocketRunClient(new URI(PropertiesUtil.getProperty("IN_WS_URL")),EStockType.IN); websocketRunClient.connect(); websocketRunClient.setConnectionLostTimeout(0); new Thread(() -> { while (true) { try { Thread.sleep(8000); websocketRunClient.send("heartbeat".getBytes()); } catch (Exception e) { websocketRunClient.reconnect(); websocketRunClient.setConnectionLostTimeout(0); } } }).start(); } catch (Exception e) { } return retMap; } } //package com.nq.ws; // // //import com.nq.enums.EStockType; //import com.nq.utils.PropertiesUtil; //import lombok.extern.slf4j.Slf4j; //import org.java_websocket.client.WebSocketClient; //import org.springframework.context.annotation.Bean; //import org.springframework.context.annotation.Configuration; // //import java.net.URI; //import java.util.HashMap; //import java.util.Map; // //@Slf4j //@Configuration //public class WebSocketClientBeanConfig { // // // @Bean // public Map<String, WebSocketClient> websocketRunClientMap() { // // Map<String, WebSocketClient> retMap = new HashMap<>(2); // try { // WebsocketRunClient websocketRunClient = new WebsocketRunClient(new URI(PropertiesUtil.getProperty("IN_WS_URL")),EStockType.IN); // websocketRunClient.connect(); // websocketRunClient.setConnectionLostTimeout(0); // new Thread(() -> { // while (true) { // try { // Thread.sleep(8000); // websocketRunClient.send("heartbeat".getBytes()); // } catch (Exception e) { // websocketRunClient.reconnect(); // websocketRunClient.setConnectionLostTimeout(0); // } // } // }).start(); // } catch (Exception e) { // } // // // // return retMap; // } // // //} src/main/java/com/nq/ws/WebsocketRunClient.java
@@ -1,63 +1,63 @@ package com.nq.ws; import com.google.gson.Gson; import com.google.gson.reflect.TypeToken; import com.nq.enums.EStockType; import com.nq.pojo.StockRealTimeBean; import com.nq.service.IMandatoryLiquidationService; import com.nq.service.impl.MandatoryLiquidationService; import com.nq.utils.ApplicationContextRegisterUtil; import com.nq.utils.redis.RedisKeyUtil; import lombok.extern.slf4j.Slf4j; import org.java_websocket.client.WebSocketClient; import org.java_websocket.handshake.ServerHandshake; import org.springframework.context.ApplicationContext; import java.lang.reflect.Type; import java.net.URI; import java.util.Map; @Slf4j public class WebsocketRunClient extends WebSocketClient { private EStockType eStockType; public WebsocketRunClient(URI serverUri, EStockType eStockType ) { super(serverUri); this.eStockType = eStockType; } @Override public void onOpen(ServerHandshake serverHandshake) { send(("key:"+ eStockType.getStockKey()+":"+eStockType.getContryId()).getBytes()); } @Override public void onMessage(String s) { try { ApplicationContext act = ApplicationContextRegisterUtil.getApplicationContext(); MandatoryLiquidationService liquidationService = (MandatoryLiquidationService) act.getBean(IMandatoryLiquidationService.class); StockRealTimeBean stockDetailBean = new Gson().fromJson(s, StockRealTimeBean.class); liquidationService.RealTimeDataProcess(eStockType,stockDetailBean); }catch (Exception e){ } } public static Map<String, Object> jsonToMap(String json) { Gson gson = new Gson(); Type type = new TypeToken<Map<String, Object>>(){}.getType(); return gson.fromJson(json, type); } @Override public void onClose(int i, String s, boolean b) { log.info("websocket 印度股票 关闭"+1); } @Override public void onError(Exception e) { log.info("websocket 错误"); } } //package com.nq.ws; // //import com.google.gson.Gson; //import com.google.gson.reflect.TypeToken; //import com.nq.enums.EStockType; //import com.nq.pojo.StockRealTimeBean; //import com.nq.service.IMandatoryLiquidationService; //import com.nq.service.impl.MandatoryLiquidationService; //import com.nq.utils.ApplicationContextRegisterUtil; //import com.nq.utils.redis.RedisKeyUtil; //import lombok.extern.slf4j.Slf4j; //import org.java_websocket.client.WebSocketClient; //import org.java_websocket.handshake.ServerHandshake; //import org.springframework.context.ApplicationContext; // //import java.lang.reflect.Type; //import java.net.URI; //import java.util.Map; // //@Slf4j //public class WebsocketRunClient extends WebSocketClient { // // private EStockType eStockType; // public WebsocketRunClient(URI serverUri, // EStockType eStockType // ) { // super(serverUri); // this.eStockType = eStockType; // } // // @Override // public void onOpen(ServerHandshake serverHandshake) { // send(("key:"+ eStockType.getStockKey()+":"+eStockType.getContryId()).getBytes()); // } // // @Override // public void onMessage(String s) { // try { // ApplicationContext act = ApplicationContextRegisterUtil.getApplicationContext(); // MandatoryLiquidationService liquidationService = (MandatoryLiquidationService) act.getBean(IMandatoryLiquidationService.class); // StockRealTimeBean stockDetailBean = new Gson().fromJson(s, StockRealTimeBean.class); // liquidationService.RealTimeDataProcess(eStockType,stockDetailBean); // }catch (Exception e){ // // } // } // // public static Map<String, Object> jsonToMap(String json) { // Gson gson = new Gson(); // Type type = new TypeToken<Map<String, Object>>(){}.getType(); // return gson.fromJson(json, type); // } // // @Override // public void onClose(int i, String s, boolean b) { // log.info("websocket 印度股票 关闭"+1); // } // // @Override // public void onError(Exception e) { // log.info("websocket 错误"); // } //} src/main/java/com/nq/ws/WsClientConfig.java
@@ -1,149 +1,166 @@ //package com.nq.ws; // //import com.google.gson.Gson; //import com.google.gson.reflect.TypeToken; //import com.nq.enums.EStockType; //import com.nq.pojo.StockRealTimeBean; //import com.nq.service.IMandatoryLiquidationService; //import com.nq.service.impl.MandatoryLiquidationService; //import com.nq.utils.ApplicationContextRegisterUtil; //import com.nq.utils.PropertiesUtil; //import com.nq.ws.client.IO; //import com.nq.ws.client.Socket; //import io.socket.emitter.Emitter; //import lombok.extern.slf4j.Slf4j; //import org.apache.http.HttpResponse; //import org.apache.http.client.HttpClient; //import org.apache.http.client.entity.UrlEncodedFormEntity; //import org.apache.http.client.methods.HttpPost; //import org.apache.http.impl.client.HttpClients; //import org.apache.http.message.BasicNameValuePair; //import org.json.JSONException; //import org.json.JSONObject; //import org.springframework.context.ApplicationContext; //import org.springframework.context.annotation.Bean; //import org.springframework.context.annotation.Configuration; // //import java.io.IOException; //import java.lang.reflect.Type; //import java.net.URI; //import java.net.URISyntaxException; //import java.nio.charset.StandardCharsets; //import java.text.SimpleDateFormat; //import java.util.ArrayList; //import java.util.Date; //import java.util.List; //import java.util.Map; package com.nq.ws; import com.google.gson.Gson; import com.google.gson.reflect.TypeToken; import com.nq.enums.EStockType; import com.nq.pojo.StockRealTimeBean; import com.nq.service.IMandatoryLiquidationService; import com.nq.service.impl.MandatoryLiquidationService; import com.nq.utils.ApplicationContextRegisterUtil; import com.nq.utils.PropertiesUtil; import com.nq.ws.client.IO; import com.nq.ws.client.Socket; import io.socket.emitter.Emitter; import lombok.extern.slf4j.Slf4j; import org.apache.http.HttpResponse; import org.apache.http.client.HttpClient; import org.apache.http.client.entity.UrlEncodedFormEntity; import org.apache.http.client.methods.HttpPost; import org.apache.http.impl.client.HttpClients; import org.apache.http.message.BasicNameValuePair; import org.json.JSONException; import org.json.JSONObject; import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.io.IOException; import java.lang.reflect.Type; import java.net.URI; import java.net.URISyntaxException; import java.nio.charset.StandardCharsets; import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.Map; /** * soket-io */ //@Slf4j //@Configuration //public class WsClientConfig { // // private static final String SERVER_URL = PropertiesUtil.getProperty("JS_IN_HTTP_API"); // private static final String AUTH_KEY = PropertiesUtil.getProperty("JS_IN_KEY"); // private static final String ROOM_ID = "14"; // private static Socket socket; // // @Bean // public void websocketRunClientMap() { // connectToServer(); // } // // private static void connectToServer() { // IO.Options options = new IO.Options(); // options.reconnection = true; // options.reconnectionDelay = 1000; // // try { // socket = IO.socket(new URI(SERVER_URL), options); // } catch (URISyntaxException e) { // log.error("Invalid URI", e); // return; // } // // socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() { // @Override // public void call(Object... args) { // authenticate(); // } // }); // // socket.on("marketData", new Emitter.Listener() { // @Override // public void call(Object... args) { // Map<String, Object> map = jsonToMap(args[0].toString()); // if(map.get("pid").equals("00000001")){ // System.out.println("接收时间：" + new SimpleDateFormat("HH:mm:ss").format(new Date()) + " " + args[0].toString()); // } // try { // ApplicationContext act = ApplicationContextRegisterUtil.getApplicationContext(); // MandatoryLiquidationService liquidationService = (MandatoryLiquidationService) act.getBean(IMandatoryLiquidationService.class); // StockRealTimeBean stockDetailBean = new Gson().fromJson(args[0].toString(), StockRealTimeBean.class); // liquidationService.RealTimeDataProcess(EStockType.IN,stockDetailBean); // }catch (Exception e){ // log.error("socket数据存入缓存错误：", e.getMessage()); // } // } // }); // // socket.on(Socket.EVENT_CONNECT_ERROR, new Emitter.Listener() { // @Override // public void call(Object... args) { // log.error("socketIo连接错误: " + args[0]); // } // }); // // socket.on(Socket.EVENT_DISCONNECT, new Emitter.Listener() { // @Override // public void call(Object... args) { // log.error("socketIo 断开连接: " + args[0]); // if (!socket.connected()) { // log.error("socketIo 开始重连: " + args[0]); // reconnect(); // } // } // }); // // socket.connect(); // } // // public static Map<String, Object> jsonToMap(String json) { // Gson gson = new Gson(); // Type type = new TypeToken<Map<String, Object>>(){}.getType(); // return gson.fromJson(json, type); // } // // private static void reconnect() { // new Thread(new Runnable() { // @Override // public void run() { // while (!socket.connected()) { // try { // log.error("socketIo 开始重连"); // socket.connect(); // Thread.sleep(5000); // 重连间隔，单位毫秒 // } catch (InterruptedException e) { // log.error("重连被中断", e); // } // } // } // }).start(); // } // // private static void authenticate() { // log.info("socketIo---->开始连接"); // if (socket != null && socket.connected()) { // JSONObject authData = new JSONObject(); // try { // authData.put("key", AUTH_KEY); // authData.put("roomId", ROOM_ID); // socket.emit("authenticate", authData); // log.info("socketIo---->连接成功"); // } catch (JSONException e) { // log.error("socketIo认证错误："+e.getMessage(), e); // } // } // } //} @Slf4j @Configuration public class WsClientConfig { private static final String SERVER_URL = PropertiesUtil.getProperty("JS_IN_HTTP_API"); private static final String AUTH_KEY = PropertiesUtil.getProperty("JS_IN_KEY"); private static final String ROOM_ID = "14"; private static Socket socket; private static HttpClient httpClient = HttpClients.createDefault(); // 单例化 HttpClient private static HttpPost httpPost; static { httpPost = new HttpPost("http://127.0.0.1:8001/api/sendNotification"); // 初始化 HttpPost } @Bean public void websocketRunClientMap() { connectToServer(); } private static void connectToServer() { IO.Options options = new IO.Options(); options.reconnection = true; options.reconnectionDelay = 1000; try { socket = IO.socket(new URI(SERVER_URL), options); } catch (URISyntaxException e) { log.error("Invalid URI", e); return; } socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() { @Override public void call(Object... args) { authenticate(); } }); socket.on("marketData", new Emitter.Listener() { @Override public void call(Object... args) { send(args[0].toString()); } }); socket.on(Socket.EVENT_CONNECT_ERROR, new Emitter.Listener() { @Override public void call(Object... args) { log.error("socketIo连接错误: " + args[0]); } }); socket.on(Socket.EVENT_DISCONNECT, new Emitter.Listener() { @Override public void call(Object... args) { log.error("socketIo 断开连接: " + args[0]); if (!socket.connected()) { log.error("socketIo 开始重连: " + args[0]); reconnect(); } } }); socket.connect(); } public static Map<String, Object> jsonToMap(String json) { Gson gson = new Gson(); Type type = new TypeToken<Map<String, Object>>(){}.getType(); return gson.fromJson(json, type); } private static void reconnect() { new Thread(new Runnable() { @Override public void run() { while (!socket.connected()) { try { log.error("socketIo 开始重连"); socket.connect(); Thread.sleep(5000); // 重连间隔，单位毫秒 } catch (InterruptedException e) { log.error("重连被中断", e); } } } }).start(); } private static void authenticate() { log.info("socketIo---->开始连接"); if (socket != null && socket.connected()) { JSONObject authData = new JSONObject(); try { authData.put("key", AUTH_KEY); authData.put("roomId", ROOM_ID); socket.emit("authenticate", authData); log.info("socketIo---->连接成功"); } catch (JSONException e) { log.error("socketIo认证错误："+e.getMessage(), e); } } } private static void send(String message) { try { // 准备 form-data 参数 List<BasicNameValuePair> params = new ArrayList<>(); params.add(new BasicNameValuePair("message", message)); // 设置编码格式为 UTF-8 UrlEncodedFormEntity entity = new UrlEncodedFormEntity(params, StandardCharsets.UTF_8); httpPost.setEntity(entity); // 设置 HttpPost 对象的参数 // 发送请求 HttpResponse response = httpClient.execute(httpPost); // 处理响应 int statusCode = response.getStatusLine().getStatusCode(); } catch (IOException e) { log.error("Http 请求错误", e); } } } src/main/resources/application.properties
@@ -41,13 +41,14 @@ #股票key的地址 IN_HTTP_API = https://data.is4vc.com/v1 IN_WS_URL =ws://ws.is4vc.com:8001/websocket-server/v1 IN_KEY = r3ZAgtcYzuBizmqge2hK IN_NEW_HTTP_API = http://api-in-2.js-stock.top/ IN_NEW_KEY = eVKtHt7aG4m6ozwWL9qG #IN_HTTP_API = https://data.is4vc.com/v1 #IN_WS_URL =ws://ws.is4vc.com:8001/websocket-server/v1 #IN_KEY = r3ZAgtcYzuBizmqge2hK #IN_NEW_HTTP_API = http://api-in-2.js-stock.top/ #IN_NEW_KEY = eVKtHt7aG4m6ozwWL9qG JS_IN_HTTP_API = http://api-in-2-socket.js-stock.top JS_IN_HTTP_URL = http://api-in-2.js-stock.top/ JS_IN_WS_URL =ws://api-in-2-ws.js-stock.top JS_IN_KEY = eVKtHt7aG4m6ozwWL9qG
