From ce511a3d55e5f349b1647172e562c1659bbc8172 Mon Sep 17 00:00:00 2001
From: Junjie <DELL@qq.com>
Date: 星期五, 09 一月 2026 15:27:27 +0800
Subject: [PATCH] #
---
src/main/java/com/zy/core/thread/impl/NyShuttleThread.java | 225 ++++++++++++++++++++++++++++++++++++++++++++++++--------
1 files changed, 192 insertions(+), 33 deletions(-)
diff --git a/src/main/java/com/zy/core/thread/impl/NyShuttleThread.java b/src/main/java/com/zy/core/thread/impl/NyShuttleThread.java
index fbb504c..b04f6d2 100644
--- a/src/main/java/com/zy/core/thread/impl/NyShuttleThread.java
+++ b/src/main/java/com/zy/core/thread/impl/NyShuttleThread.java
@@ -6,6 +6,7 @@
import com.zy.common.utils.DateUtils;
import com.zy.common.utils.RedisUtil;
import com.zy.core.News;
+import com.zy.core.enums.RedisKeyType;
import com.zy.core.model.DeviceCommandMsgModel;
import com.zy.core.model.DeviceMsgModel;
import com.zy.core.properties.DeviceConfig;
@@ -13,14 +14,18 @@
import com.zy.core.cache.OutputQueue;
import com.zy.core.enums.SlaveType;
import com.zy.core.thread.ShuttleThread;
-import io.netty.handler.timeout.ReadTimeoutException;
+import com.zy.core.utils.FakeDeviceUtils;
+import com.zy.core.utils.TimeoutExecutor;
import lombok.extern.slf4j.Slf4j;
import java.io.*;
import java.net.InetAddress;
import java.net.Socket;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
import java.text.MessageFormat;
import java.util.*;
+import java.util.concurrent.TimeUnit;
@Slf4j
@SuppressWarnings("all")
@@ -30,6 +35,8 @@
private RedisUtil redisUtil;
private Socket socket;
private boolean stopThread = false;
+ private HashMap<Integer, String> resultKeyMap = new HashMap<Integer, String>();
+ private long lastConnectTime = System.currentTimeMillis();
public NyShuttleThread(DeviceConfig deviceConfig, RedisUtil redisUtil) {
this.deviceConfig = deviceConfig;
@@ -49,9 +56,15 @@
try {
this.connect();
+ // 鍦ㄥ惊鐜腑浣跨敤
+ boolean result = TimeoutExecutor.executeWithTimeout(
+ () -> this.listenSocketMessage(),
+ 30, // 30绉掕秴鏃�
+ TimeUnit.SECONDS
+ );
Thread.sleep(200);
- listenSocketMessage();
} catch (Exception e) {
+ closeSocket();
e.printStackTrace();
}
}
@@ -77,6 +90,8 @@
if (deviceCommandMsg == null) {
continue;
}
+
+ this.connect();
executeCommand(deviceCommandMsg);
Thread.sleep(200);
@@ -94,22 +109,41 @@
return;
}
- Object command = deviceCommandMsg.getCommand();
+ String command = JSON.toJSONString(deviceCommandMsg.getCommand());
+ JSONObject commandObj = JSON.parseObject(command);
+ JSONObject request = commandObj.getJSONObject("request");
+ JSONObject header = request.getJSONObject("header");
+ Integer requestId = header.getInteger("requestId");
+ resultKeyMap.put(requestId, deviceCommandMsg.getResultKey());
// 鑾峰彇杈撳嚭娴�
OutputStreamWriter writer = new OutputStreamWriter(this.socket.getOutputStream());
- writer.write(JSON.toJSONString(command) + "\r\n");
+ writer.write(command + "\r\n");
writer.flush();
// System.out.println("Sent message to server: " + JSON.toJSONString(httpCommand));
- }catch (Exception e) {
+ }catch (SocketException e) {
+ e.printStackTrace();
+ closeSocket();
+ } catch (IOException e) {
e.printStackTrace();
}
}
- private void listenSocketMessage() {
+ private void closeSocket() {
+ try {
+ this.socket.close();
+ } catch (Exception e1) {
+
+ }finally {
+ this.socket = null;
+ }
+ }
+
+ private boolean listenSocketMessage() {
+ StringBuffer sb = new StringBuffer();
try {
if (this.socket == null) {
- return;
+ return false;
}
DeviceMsgUtils deviceMsgUtils = null;
@@ -118,31 +152,54 @@
} catch (Exception e) {
}
if (deviceMsgUtils == null) {
- return;
+ return false;
}
// 鑾峰彇杈撳叆娴�
BufferedReader reader = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));
// 璇诲彇鏈嶅姟鍣ㄧ殑鍝嶅簲
- StringBuffer sb = new StringBuffer();
char[] chars = new char[2048];//缂撳啿鍖�
while (true) {
reader.read(chars);
String trim = new String(chars);
- sb.append(trim);
- if (trim.lastIndexOf("\r\n") != -1) {
+ int lastIndexOf = trim.lastIndexOf("\r\n");
+ if (lastIndexOf != -1) {
+ trim = trim.substring(0, lastIndexOf);
+ sb.append(trim);
break;
}
}
- JSONObject result = JSON.parseObject(sb.toString());//寰楀埌鍝嶅簲缁撴灉闆�
- log.info("鏀跺埌Server Data: {}", JSON.toJSONString(result));
+ JSONObject result = null;
+ try {
+ result = JSON.parseObject(sb.toString());//寰楀埌鍝嶅簲缁撴灉闆�
+ }catch (Exception e){}
+
+ if(result == null) {
+ return false;
+ }
+
+ JSONObject header = null;
+ JSONObject body = null;
+ JSONObject response = result.getJSONObject("response");
+ if (response == null) {
+ JSONObject request = result.getJSONObject("request");
+ header = request.getJSONObject("header");
+ body = request.getJSONObject("body");
+ }else {
+ header = response.getJSONObject("header");
+ body = response.getJSONObject("body");
+ }
String msgType = result.getString("msgType");
+
+// log.info("鏀跺埌Server Data: {}", JSON.toJSONString(result));
if ("responseMsg".equals(msgType)) {
- JSONObject response = result.getJSONObject("response");
- JSONObject body = response.getJSONObject("body");
- if (body.containsKey("workingMode")) {
+ Integer responseId = header.getInteger("responseId");
+ String resultKey = resultKeyMap.get(responseId);
+
+ String responseType = body.getString("responseType");
+ if (responseType.equals("state")) {
//read
JSONObject data = parseSocketResult(body);
@@ -151,20 +208,38 @@
deviceMsgModel.setDeviceMsgType("status");
deviceMsgModel.setDeviceMsg(data);
deviceMsgModel.setDeviceOriginMsg(sb.toString());
+ deviceMsgModel.setResultKey(resultKey);
deviceMsgUtils.sendDeviceMsg(SlaveType.Shuttle, deviceConfig.getDeviceNo(), deviceMsgModel);
- return;
+ }else {
+ log.info("鏀跺埌Rcs Shuttle Command Data: {}", JSON.toJSONString(result));
+ DeviceMsgModel deviceMsgModel = new DeviceMsgModel();
+ deviceMsgModel.setDeviceId(deviceConfig.getDeviceNo());
+ deviceMsgModel.setDeviceMsgType("command");
+ deviceMsgModel.setDeviceMsg(result);
+ deviceMsgModel.setDeviceOriginMsg(sb.toString());
+ deviceMsgModel.setResultKey(resultKey);
+ deviceMsgUtils.sendDeviceMsg(SlaveType.Shuttle, deviceConfig.getDeviceNo(), deviceMsgModel);
+ }
+ } else if ("requestMsg".equals(msgType)) {
+ log.info("鏀跺埌Shuttle Init Data: {}", JSON.toJSONString(result));
+ String requestType = body.getString("requestType");
+ if (requestType.equals("init")) {
+ DeviceMsgModel deviceMsgModel = new DeviceMsgModel();
+ deviceMsgModel.setDeviceId(deviceConfig.getDeviceNo());
+ deviceMsgModel.setDeviceMsgType("shuttleInit");
+ deviceMsgModel.setDeviceMsg(result);
+ deviceMsgModel.setDeviceOriginMsg(sb.toString());
+ deviceMsgUtils.sendDeviceMsg(SlaveType.Shuttle, deviceConfig.getDeviceNo(), deviceMsgModel);
}
}
-
- DeviceMsgModel deviceMsgModel = new DeviceMsgModel();
- deviceMsgModel.setDeviceId(deviceConfig.getDeviceNo());
- deviceMsgModel.setDeviceMsgType("command");
- deviceMsgModel.setDeviceMsg(result);
- deviceMsgModel.setDeviceOriginMsg(sb.toString());
- deviceMsgUtils.sendDeviceMsg(SlaveType.Shuttle, deviceConfig.getDeviceNo(), deviceMsgModel);
- } catch (Exception e) {
-// e.printStackTrace();
+ } catch (SocketTimeoutException e) {
+ return false;
+ } catch (IOException e) {
+ e.printStackTrace();
+ closeSocket();
+ return false;
}
+ return true;
}
public JSONObject parseSocketResult(JSONObject data) {
@@ -223,13 +298,23 @@
return true;
}
- InetAddress address = InetAddress.getByName(deviceConfig.getIp());
- if (address.isReachable(10000)) {
- Socket socket = new Socket(deviceConfig.getIp(), deviceConfig.getPort());
- socket.setSoTimeout(10000);
- socket.setKeepAlive(true);
- this.socket = socket;
- log.info(MessageFormat.format("銆恵0}銆戝洓鍚戠┛姊溅Socket閾炬帴鎴愬姛 ===>> [id:{1}] [ip:{2}] [port:{3}]", DateUtils.convert(new Date()), deviceConfig.getDeviceNo(), deviceConfig.getIp(), deviceConfig.getPort()));
+ if(System.currentTimeMillis() - lastConnectTime < 1000 * 10) {
+ return false;
+ }
+
+ this.lastConnectTime = System.currentTimeMillis();
+
+ if(this.deviceConfig.getFake()) {
+ return fakeConnect();
+ }else {
+ InetAddress address = InetAddress.getByName(deviceConfig.getIp());
+ if (address.isReachable(10000)) {
+ Socket deviceSocket = new Socket(deviceConfig.getIp(), deviceConfig.getPort());
+ deviceSocket.setSoTimeout(10000);
+ deviceSocket.setKeepAlive(true);
+ this.socket = deviceSocket;
+ log.info(MessageFormat.format("銆恵0}銆戝洓鍚戠┛姊溅Socket閾炬帴鎴愬姛 ===>> [id:{1}] [ip:{2}] [port:{3}]", DateUtils.convert(new Date()), deviceConfig.getDeviceNo(), deviceConfig.getIp(), deviceConfig.getPort()));
+ }
}
} catch (Exception e) {
OutputQueue.SHUTTLE.offer(MessageFormat.format("銆恵0}銆戝洓鍚戠┛姊溅Socket閾炬帴澶辫触 ===>> [id:{1}] [ip:{2}] [port:{3}]", DateUtils.convert(new Date()), deviceConfig.getDeviceNo(), deviceConfig.getIp(), deviceConfig.getPort()));
@@ -237,6 +322,76 @@
}
return true;
+ }
+
+ private boolean fakeConnect() {
+ try {
+ FakeDeviceUtils fakeDeviceUtils = SpringUtils.getBean(FakeDeviceUtils.class);
+ if(fakeDeviceUtils == null) {
+ return false;
+ }
+
+ boolean applyResult = fakeDeviceUtils.applyShuttleConnect(deviceConfig.getDeviceNo());
+ if(!applyResult) {
+ return false;
+ }
+
+ InetAddress address = InetAddress.getByName(deviceConfig.getIp());
+ if (address.isReachable(10000)) {
+ Socket deviceSocket = new Socket(deviceConfig.getIp(), deviceConfig.getPort());
+ deviceSocket.setSoTimeout(10000);
+ deviceSocket.setKeepAlive(true);
+
+ //铏氭嫙璁惧闇�瑕佷笂鎶ヨ澶囦俊鎭�
+ HashMap<String, Object> map = new HashMap<>();
+ map.put("msgType", "fakeDeviceFirstConnect");
+ map.put("deviceConfig", this.deviceConfig);
+
+ // 鑾峰彇杈撳嚭娴�
+ OutputStreamWriter writer = new OutputStreamWriter(deviceSocket.getOutputStream());
+ writer.write(JSON.toJSONString(map) + "\r\n");
+ writer.flush();
+
+ // 鑾峰彇杈撳叆娴�
+ BufferedReader reader = new BufferedReader(new InputStreamReader(deviceSocket.getInputStream()));
+ // 璇诲彇鏈嶅姟鍣ㄧ殑鍝嶅簲
+ StringBuffer sb = new StringBuffer();
+ char[] chars = new char[2048];//缂撳啿鍖�
+ while (true) {
+ reader.read(chars);
+ String trim = new String(chars);
+ sb.append(trim);
+ if (trim.lastIndexOf("\r\n") != -1) {
+ break;
+ }
+ }
+
+ JSONObject result = null;
+ try {
+ result = JSON.parseObject(sb.toString());//寰楀埌鍝嶅簲缁撴灉闆�
+ }catch (Exception e){}
+
+ if(result == null) {
+ return false;
+ }
+
+ if(result.getInteger("deviceNo") != deviceConfig.getDeviceNo()) {
+ return false;
+ }
+
+ if(!result.getString("status").equals("success")) {
+ return false;
+ }
+
+ this.socket = deviceSocket;
+
+ log.info(MessageFormat.format("銆恵0}銆戝洓鍚戠┛姊溅Socket閾炬帴鎴愬姛 ===>> [id:{1}] [ip:{2}] [port:{3}]", DateUtils.convert(new Date()), deviceConfig.getDeviceNo(), deviceConfig.getIp(), deviceConfig.getPort()));
+ return true;
+ }
+ }catch (Exception e){
+ e.printStackTrace();
+ }
+ return false;
}
@Override
@@ -254,4 +409,8 @@
this.stopThread = true;
}
+ @Override
+ public Socket getSocket() {
+ return this.socket;
+ }
}
--
Gitblit v1.9.1