#

Junjie

2025-09-25 2b6156951c18ea9ff60ed5578db633cebbe99fa7

 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,13 +14,18 @@
import com.zy.core.cache.OutputQueue;
import com.zy.core.enums.SlaveType;
import com.zy.core.thread.ShuttleThread;
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 +36,7 @@
    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,8 +56,13 @@

                try {
                    this.connect();
                    // 在循环中使用
                    boolean result = TimeoutExecutor.executeWithTimeout(
                            () -> this.listenSocketMessage(),
                            30,  // 30秒超时
                            TimeUnit.SECONDS
                    );
                    Thread.sleep(200);
                    listenSocketMessage();
                } catch (Exception e) {
                    e.printStackTrace();
                }
@@ -77,6 +89,8 @@
                    if (deviceCommandMsg == null) {
                        continue;
                    }

                    this.connect();
                    executeCommand(deviceCommandMsg);

                    Thread.sleep(200);
@@ -106,15 +120,29 @@
            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;
@@ -123,27 +151,44 @@
            } 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());//得到响应结果集
            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");
            JSONObject header = response.getJSONObject("header");
            JSONObject body = response.getJSONObject("body");
            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");

@@ -186,9 +231,14 @@
                    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) {
@@ -247,13 +297,22 @@
                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;
            }

            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;
                    this.lastConnectTime = System.currentTimeMillis();
                    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()));
@@ -261,6 +320,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
@@ -278,4 +407,8 @@
        this.stopThread = true;
    }

    @Override
    public Socket getSocket() {
        return this.socket;
    }
}