package com.zy.core.thread.impl; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; 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.enums.SlaveType; import com.zy.core.thread.ShuttleThread; import com.zy.core.utils.FakeDeviceUtils; 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.*; @Slf4j @SuppressWarnings("all") public class NyShuttleThread implements ShuttleThread { private DeviceConfig deviceConfig; private RedisUtil redisUtil; private Socket socket; private boolean stopThread = false; private HashMap resultKeyMap = new HashMap(); private long lastConnectTime = System.currentTimeMillis(); public NyShuttleThread(DeviceConfig deviceConfig, RedisUtil redisUtil) { this.deviceConfig = deviceConfig; this.redisUtil = redisUtil; } @Override public void run() { 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(); } } }); innerThread.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; } 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() { 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) { break; } } JSONObject result = null; try { result = JSON.parseObject(sb.toString());//得到响应结果集 }catch (Exception e){} if(result == null) { return; } JSONObject response = result.getJSONObject("response"); JSONObject header = response.getJSONObject("header"); JSONObject body = response.getJSONObject("body"); String msgType = result.getString("msgType"); // log.info("收到Server Data: {}", JSON.toJSONString(result)); if ("responseMsg".equals(msgType)) { Integer responseId = header.getInteger("responseId"); String resultKey = resultKeyMap.get(responseId); String responseType = body.getString("responseType"); if (responseType.equals("state")) { //read 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) { } catch (IOException e) { e.printStackTrace(); closeSocket(); } } 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 { 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()), 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 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 public void close() { } @Override public DeviceConfig getDeviceConfig() { return this.deviceConfig; } @Override public void stopThread() { this.stopThread = true; } @Override public Socket getSocket() { return this.socket; } }