#

Junjie

2025-09-23 5b556e4e6744a57dcbc372f0277a071d75e300d3

 src/main/java/com/zy/core/thread/impl/NyShuttleThread.java

@@ -2,20 +2,26 @@

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.core.common.DateUtils;
import com.zy.common.SpringUtils;
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;
import com.zy.core.utils.DeviceMsgUtils;
import com.zy.core.cache.OutputQueue;
import com.zy.core.model.ShuttleSlave;
import com.zy.core.enums.SlaveType;
import com.zy.core.thread.ShuttleThread;
import com.zy.core.utils.FakeDeviceUtils;
import lombok.extern.slf4j.Slf4j;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
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.*;

@@ -23,29 +29,32 @@
@SuppressWarnings("all")
public class NyShuttleThread implements ShuttleThread {

    private ShuttleSlave slave;
    private DeviceConfig deviceConfig;
    private RedisUtil redisUtil;
    private Socket socket;
    private boolean stopThread = false;
    private HashMap<Integer, String> resultKeyMap = new HashMap<Integer, String>();
    private long lastConnectTime = System.currentTimeMillis();

    private static final boolean DEBUG = false;//调试模式

    private List<JSONObject> socketReadResults = new ArrayList<>();
    private List<JSONObject> socketResults = new ArrayList<>();

    public NyShuttleThread(ShuttleSlave slave, RedisUtil redisUtil) {
        this.slave = slave;
    public NyShuttleThread(DeviceConfig deviceConfig, RedisUtil redisUtil) {
        this.deviceConfig = deviceConfig;
        this.redisUtil = redisUtil;
    }

    @Override
    public void run() {
        News.info("{}号四向车线程启动", slave.getId());
        this.connect();
        News.info("{}号四向车线程启动", deviceConfig.getDeviceNo());

        //监听消息并存储
        //监听消息
        Thread innerThread = new Thread(() -> {
            while (true) {
                if(stopThread) {
                    break;
                }

                try {
                    this.connect();
                    Thread.sleep(200);
                    listenSocketMessage();
                } catch (Exception e) {
                    e.printStackTrace();
@@ -54,147 +63,324 @@
        });
        innerThread.start();

//        //设备执行
//        Thread executeThread = new Thread(() -> {
//            while (true) {
//                try {
//                    ShuttleAction shuttleAction = SpringUtils.getBean(ShuttleAction.class);
//                    if (shuttleAction == null) {
//                        continue;
//                    }
//
//                    Object object = redisUtil.get(RedisKeyType.SHUTTLE_FLAG.key + slave.getId());
//                    if (object == null) {
//                        continue;
//                    }
//
//                    Integer taskNo = Integer.valueOf(String.valueOf(object));
//                    if (taskNo != 0) {
//                        //存在任务需要执行
//                        boolean result = shuttleAction.executeWork(slave.getId(), taskNo);
//                    }
//
////                    //小车空闲且有跑库程序
////                    shuttleAction.moveLoc(slave.getId());
//
//                    //演示模式
//                    shuttleAction.demo(slave.getId());
//
//                    Thread.sleep(200);
//                } catch (Exception e) {
//                    e.printStackTrace();
//                }
//            }
//        });
//        executeThread.start();
        //执行指令
        Thread executeThread = new Thread(() -> {
            while (true) {
                if(stopThread) {
                    break;
                }

                try {
                    DeviceMsgUtils deviceMsgUtils = null;
                    try {
                        deviceMsgUtils = SpringUtils.getBean(DeviceMsgUtils.class);
                    }catch (Exception e){}
                    if (deviceMsgUtils == null) {
                        continue;
                    }
                    DeviceCommandMsgModel deviceCommandMsg = deviceMsgUtils.getDeviceCommandMsg(SlaveType.Shuttle, deviceConfig.getDeviceNo());
                    if (deviceCommandMsg == null) {
                        continue;
                    }

                    this.connect();
                    executeCommand(deviceCommandMsg);

                    Thread.sleep(200);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        executeThread.start();
    }

    private void executeCommand(DeviceCommandMsgModel deviceCommandMsg) {
        try {
            if (this.socket == null) {
                return;
            }

            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(command + "\r\n");
            writer.flush();
//            System.out.println("Sent message to server: " + JSON.toJSONString(httpCommand));
        }catch (SocketException e) {
            e.printStackTrace();
            closeSocket();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void closeSocket() {
        try {
            this.socket.close();
        } catch (Exception e1) {

        }finally {
            this.socket = null;
        }
    }

    private void listenSocketMessage() {
        StringBuffer sb = new StringBuffer();
        try {
            if (this.socket == null) {
                return;
            }

            DeviceMsgUtils deviceMsgUtils = null;
            try {
                deviceMsgUtils = SpringUtils.getBean(DeviceMsgUtils.class);
            } catch (Exception e) {
            }
            if (deviceMsgUtils == null) {
                return;
            }

            // 获取输入流
            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;
            }

            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
                    socketReadResults.add(body);
                    return;
                    JSONObject data = parseSocketResult(body);

                    DeviceMsgModel deviceMsgModel = new DeviceMsgModel();
                    deviceMsgModel.setDeviceId(deviceConfig.getDeviceNo());
                    deviceMsgModel.setDeviceMsgType("status");
                    deviceMsgModel.setDeviceMsg(data);
                    deviceMsgModel.setDeviceOriginMsg(sb.toString());
                    deviceMsgModel.setResultKey(resultKey);
                    deviceMsgUtils.sendDeviceMsg(SlaveType.Shuttle, deviceConfig.getDeviceNo(), deviceMsgModel);
                }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);
                }
            }
        } catch (SocketTimeoutException e) {

            if (!socketResults.isEmpty() && socketResults.size() >= 20) {
                socketResults.remove(0);//清理头节点
            }
            socketResults.add(result);//添加数据
        } catch (Exception e) {
//            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
            closeSocket();
        }
    }

//    public JSONObject getRequestBody(String type, String taskId) {
//        try {
//            // 获取服务器响应
//            JSONObject result = null;
//            if (type.equals("readState")) {
//                type = "state";
//            }
//
//            for (int i = 0; i < socketResults.size(); i++) {
//                JSONObject socketResult = socketResults.get(i);
//                if (!socketResult.get("msgType").equals("responseMsg")) {//不是响应内容
//                    continue;
//                }
//
//                JSONObject resultResponse = JSON.parseObject(socketResult.get("response").toString());
//                JSONObject resultBody = JSON.parseObject(resultResponse.get("body").toString());
//                String responseType = resultBody.get("responseType").toString();
//                if (DEBUG) {
//                    result = socketResult;
//                    break;
//                }
//
//                if (!responseType.equals(type)) {
//                    continue;//响应类型与请求类型不一致，不在调试模式下
//                }
//
//                if (taskId != null) {
//                    String responseTaskId = resultBody.get("taskId").toString();
//                    if (!responseTaskId.equals(taskId)) {
//                        continue;//响应ID与请求ID不一致，不在调试模式下
//                    }
//                }
//
//                result = socketResult;
//                break;
//            }
//
//            if (result == null) {
//                return null;//无响应结果
//            }
//
//            return filterBodyData(result);//返回Body结果集
//        } catch (Exception e) {
//            return null;
//        }
//    }
    public JSONObject parseSocketResult(JSONObject data) {
        JSONObject device = new JSONObject();

        //小车设备状态
        device.put("deviceStatus", data.getInteger("free"));
        //小车模式
        device.put("mode", data.getInteger("workingMode"));
        //当前二维码
        device.put("currentCode", data.getString("point"));
        //电池电量
        device.put("batteryPower", data.getString("powerPercent"));
        //电池电压
        device.put("batteryVoltage", data.getInteger("voltage"));
        //故障
        device.put("errorCode", data.getJSONArray("errCode").getString(0));

        //是否顶升
        device.put("hasLift", data.getInteger("liftPosition") == 2 ? true : false);
        //是否有托盘
        device.put("hasPallet", data.getInteger("loadState") == 1 ? true : false);
        //行驶方向
        device.put("runDirection", data.getString("runDir") == null ? "none" : data.getString("runDir"));
        //是否为充电状态
        device.put("hasCharge", data.getInteger("chargState") == 1 ? true : false);
        //运行速度
        device.put("speed", data.getInteger("speed"));

        //*********读取扩展字段**********

        JSONObject extend = new JSONObject();
        device.put("extend", extend);

        //管制状态
        extend.put("suspendState", data.getInteger("suspendState"));
        //最高电芯电压(mV)
        extend.put("maxCellVoltage", data.getInteger("maxCellVoltage"));
        //最低电芯电压(mV)
        extend.put("minCellVoltage", data.getInteger("minCellVoltage"));
        //电池电压
        extend.put("voltage", data.getInteger("voltage"));
        //充放电循环次数
        extend.put("chargeCycleTimes", data.getInteger("chargeCycleTimes"));
        //剩余电量
        extend.put("surplusQuantity", data.getInteger("surplusQuantity"));
        //总电量
        extend.put("countQuantity", data.getInteger("countQuantity"));
        return device;
    }

    @Override
    public boolean connect() {
        try {
            InetAddress address = InetAddress.getByName(slave.getIp());
            if (address.isReachable(10000)) {
                Socket socket = new Socket(slave.getIp(), slave.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()), slave.getId(), slave.getIp(), slave.getPort()));
            if(this.socket != null) {
                return true;
            }

            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()), slave.getId(), slave.getIp(), slave.getPort()));
            OutputQueue.SHUTTLE.offer(MessageFormat.format("【{0}】四向穿梭车Socket链接失败 ===>> [id:{1}] [ip:{2}] [port:{3}]", DateUtils.convert(new Date()), deviceConfig.getDeviceNo(), deviceConfig.getIp(), deviceConfig.getPort()));
            return false;
        }

        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
@@ -202,91 +388,18 @@

    }

//    //发出请求
//    private JSONObject requestCommand(NyShuttleHttpCommand httpCommand) throws IOException {
//        try {
//            if (this.socket == null) {
//                return null;
//            }
//
//            //压缩数据包
//            JSONObject data = JSON.parseObject(JSON.toJSONString(httpCommand));
//            data.remove("nodes");
//
//            // 获取输出流
//            OutputStreamWriter writer = new OutputStreamWriter(this.socket.getOutputStream());
//            writer.write(JSON.toJSONString(data) + "\r\n");
//            writer.flush();
////            System.out.println("Sent message to server: " + JSON.toJSONString(httpCommand));
//
//            String requestType = null;
//            String taskId = null;
//            try {
//                requestType = httpCommand.getRequest().getBody().get("requestType").toString();
//                taskId = httpCommand.getRequest().getBody().get("taskId").toString();
//            } catch (Exception e) {
////            return null;
//                //taskId可能取空，不报错，正常情况
//            }
//
//            // 获取服务器响应
//            // 尝试10次
//            JSONObject result = null;
//            for (int i = 0; i < 10; i++) {
//                result = getRequestBody(requestType, taskId);
//                if (result == null) {
//                    try {
//                        Thread.sleep(100);
//                    } catch (Exception e) {
//                        e.printStackTrace();
//                    }
//                }else {
//                    break;
//                }
//            }
//            return result;//返回Body结果集
//        }catch (Exception e) {
//            e.printStackTrace();
//        }
//        return null;
//    }
//
//    private void requestCommandAsync(NyShuttleHttpCommand httpCommand) throws IOException {
//        if (this.socket == null) {
//            return;
//        }
//
//       try {
//           //压缩数据包
//           JSONObject data = JSON.parseObject(JSON.toJSONString(httpCommand));
//           data.remove("nodes");
//
//           // 获取输出流
//           OutputStreamWriter writer = new OutputStreamWriter(this.socket.getOutputStream());
//           writer.write(JSON.toJSONString(data) + "\r\n");
//           writer.flush();
////            System.out.println("Sent message to server: " + JSON.toJSONString(httpCommand));
//       }catch (Exception e) {
////           e.printStackTrace();
////           System.out.println("socket write error");
//           this.socket.close();
//           this.socket = null;
//       }
//    }
//
//    private JSONObject filterBodyData(JSONObject data) {
//        Object response = data.get("response");
//        if (response == null) {
//            return null;
//        }
//
//        JSONObject result = JSON.parseObject(response.toString());
//        Object body = result.get("body");
//        if (body == null) {
//            return null;
//        }
//        JSONObject jsonBody = JSON.parseObject(body.toString());
//        return jsonBody;
//    }
    @Override
    public DeviceConfig getDeviceConfig() {
        return this.deviceConfig;
    }

    @Override
    public void stopThread() {
        this.stopThread = true;
    }

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