增加客户端控制

web/src/main/java/com/zhehekeji/web/service/client/ClientChanel.java

@@ -0,0 +1,111 @@
+package com.zhehekeji.web.service.client;
+
+
+import io.netty.channel.Channel;
+import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.time.LocalDateTime;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 所有的客户端的chanel
+ */
+@Slf4j
+public class ClientChanel {
+
+
+    static final Logger tcpLogger = LoggerFactory.getLogger("tcp");
+
+    /**
+     * key : 巷道标识符
+     */
+    static Map<String, Channel> channelMap = new ConcurrentHashMap<>();
+
+
+    /**
+     * key : 巷道标识符
+     */
+    static Map<String, LocalDateTime> channelStringTime = new ConcurrentHashMap<>();
+
+
+    /**
+     * key :IP
+     * value: 巷道标识符
+     */
+    static Map<String,String> IP_SRMNumberMap = new ConcurrentHashMap<>();
+
+    /**
+     * key :巷道标识符
+     * value: IP
+     */
+    static Map<String,String> SRMNumber_IPMap = new ConcurrentHashMap<>();
+
+    public static void putIp(String ip,String ID){
+        IP_SRMNumberMap.put(ip,ID);
+    }
+
+    public static void putSRMNUmber_Ip(String ID,String ip){
+        SRMNumber_IPMap.put(ID,ip);
+        IP_SRMNumberMap.put(ip,ID);
+    }
+
+    public static String getIpFromId(String ID){
+        return SRMNumber_IPMap.get(ID);
+    }
+
+    public static void deleteIp(String ip){
+        IP_SRMNumberMap.remove(ip);
+    }
+
+    public static String getIDFromIp(String ip){
+        return IP_SRMNumberMap.get(ip);
+    }
+
+    public static void connect(String SRMNumber, Channel channel){
+        channelMap.put(SRMNumber,channel);
+        InetSocketAddress socketAddress = (InetSocketAddress) channel.remoteAddress();
+        String clientIp = socketAddress.getAddress().getHostAddress();
+        putSRMNUmber_Ip(SRMNumber, clientIp);
+        channelStringTime.put(SRMNumber,LocalDateTime.now());
+        log.info("connect:{}巷道 ", SRMNumber);
+    }
+//    static {
+//        Timer timer = new Timer();
+//        timer.scheduleAtFixedRate(new TimerTask() {
+//            @Override
+//            public void run() {
+//                for (String key :channelStringTime.keySet()){
+//                    if(LocalDateTime.now().equals(channelStringTime.get(key).plusMinutes(5))) {
+//                        channelStringTime.remove(key);
+//                        disConnect(key);
+//                    }
+//                }
+//            }
+//        },0,60000);
+//    }
+    public static void disConnect(String key){
+        channelMap.remove(key);
+    }
+
+    public static Set<String> keys(){
+        return channelMap.keySet();
+    }
+
+    public static Channel get(String key){
+        return channelMap.get(key);
+    }
+
+    public static void write(String data,String key){
+        if(channelMap.get(key) != null){
+            channelMap.get(key).writeAndFlush(data);
+        }else {
+            tcpLogger.info("no connect client:{}",key);
+        }
+    }
+
+}

web/src/main/java/com/zhehekeji/web/service/client/Decoder.java

@@ -0,0 +1,131 @@
+package com.zhehekeji.web.service.client;
+
+import com.zhehekeji.web.lib.CameraDelayTask;
+import com.zhehekeji.web.service.EmptyCheckService;
+import com.zhehekeji.web.service.PlcService;
+import com.zhehekeji.web.service.ksec.KsecInfo;
+import com.zhehekeji.web.service.ksec.KsecNettyClient;
+import com.zhehekeji.web.service.putian.*;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.DelimiterBasedFrameDecoder;
+import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.Charset;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 客户端解码器   连接用
+ */
+@Slf4j
+public class Decoder extends DelimiterBasedFrameDecoder {
+
+    private static final Logger tcpLogger = LoggerFactory.getLogger("tcp");
+
+    public static String START_CHECK = "ST";
+    private static String GET_PHOTO = "GP";
+    private static String GET_PHOTO_END = "GPE";
+    private static String RETURN_CHECK = "RTE";
+    private static String FOLLOW_GET_PHOTO_END = "SGPE";
+    private static String END_CHECK = "EN";
+    private static String HEART_BEAT = "HB";
+
+
+
+    private static String END_STRING = "$";
+
+    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5,15,30, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<>(20000));
+
+
+    private PlcService plcService;
+
+    public Decoder(PlcService plcService) {
+
+        super(20000,true,false, Unpooled.copiedBuffer(">".getBytes()),
+                Unpooled.copiedBuffer("$".getBytes()));
+        this.plcService = plcService;
+    }
+
+    @Override
+    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
+
+        in = (ByteBuf) super.decode(ctx, in);
+        if(in == null){
+            log.debug("no data");
+            return null;
+        }
+        ClientRunnable clientRunnable = new ClientRunnable(in,ctx,plcService);
+        threadPoolExecutor.execute(clientRunnable);
+        return null;
+    }
+
+    public static class ClientRunnable implements Runnable {
+
+        private ByteBuf in;
+
+        private ChannelHandlerContext ctx;
+
+        private PlcService plcService;
+
+        public ClientRunnable(ByteBuf in,ChannelHandlerContext ctx,PlcService plcService){
+            this.ctx = ctx;
+            this.in = in;
+            this.plcService = plcService;
+        }
+
+        @Override
+        public void run() {
+            String body = in.toString(Charset.forName("UTF-8"));
+            tcpLogger.info("receive client:{}, data length:{}", body, body.length());
+            TransmissionPojo transmissionPojo = new TransmissionPojo(body);
+            //心跳进行连接
+            if(HEART_BEAT.equals(transmissionPojo.getHeader())){
+                ClientChanel.connect(transmissionPojo.getStreetNumber(),ctx.channel());
+            }
+            //获取照片后发送进行计算逻辑，并在拍照队列中取出拍照发送
+            else if(GET_PHOTO_END.equals(transmissionPojo.getHeader())){
+                if(transmissionPojo.getTaskId() != null && !"0".equals(transmissionPojo.getTaskId())) {
+                    ClientChanel.get(transmissionPojo.getStreetNumber()).writeAndFlush(transmissionPojo.toString(TransmissionType.RTS));
+                }
+                try {
+                    Thread.sleep(50L);
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+                //删除队列的拍照数据
+                GetPhotoDelayExecutor.remove(transmissionPojo.getStreetNumber(),transmissionPojo.toString(TransmissionType.GPS));
+                //发送给上位机
+                CameraDelayTask cameraDelayTask =  GetPhotoDelayExecutor.getNext(transmissionPojo.getStreetNumber());
+                if (cameraDelayTask != null) {
+                    TransmissionPojo pojo = new TransmissionPojo(cameraDelayTask.getCommand());
+                    ClientChanel.get(pojo.getStreetNumber()).writeAndFlush(pojo.toString(TransmissionType.GPS));
+                }
+
+            }
+            //照片和结果保存，并发送给上位机
+            else if(RETURN_CHECK.equals(transmissionPojo.getHeader())){
+               //保存数据
+                plcService.visualCalculationResults(transmissionPojo);
+                //发送给上位机
+                KsecInfo ksecInfo = plcService.getKsecDataInfo(transmissionPojo,"E");
+                ksecInfo.getData().setTypeNum(transmissionPojo.getCategory());
+                ksecInfo.getData().setQuantity(transmissionPojo.getCount());
+                ksecInfo.getData().setCheckRlt(transmissionPojo.getResult());
+                KsecNettyClient.write(ksecInfo);
+
+
+
+            }
+            //照片和结果保存，并发送给上位机
+            else if(FOLLOW_GET_PHOTO_END.equals(transmissionPojo.getHeader())){
+                //保存数据
+                plcService.saveFollowPhoto(transmissionPojo);
+            }
+        }
+    }
+}

web/src/main/java/com/zhehekeji/web/service/client/Encoder.java

@@ -0,0 +1,24 @@
+package com.zhehekeji.web.service.client;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToByteEncoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.StandardCharsets;
+
+/**
+ * 客户端传输协议 以；结尾
+ */
+public class Encoder extends MessageToByteEncoder<String> {
+    private static String END_STRING = ">";
+
+    private static final Logger tcpLogger = LoggerFactory.getLogger("tcp");
+    @Override
+    protected void encode(ChannelHandlerContext channelHandlerContext, String data, ByteBuf byteBuf) throws Exception {
+        data = data + END_STRING;
+        tcpLogger.info("send to client:{}, length:{}",data, data.length());
+        byteBuf.writeBytes(data.getBytes(StandardCharsets.UTF_8));
+    }
+}

web/src/main/java/com/zhehekeji/web/service/client/IntelliBlinkClient.java

@@ -0,0 +1,4 @@
+package com.zhehekeji.web.service.client;
+
+public class IntelliBlinkClient {
+}

web/src/main/java/com/zhehekeji/web/service/client/NettyConnectHandler.java

@@ -0,0 +1,44 @@
+package com.zhehekeji.web.service.client;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import lombok.extern.slf4j.Slf4j;
+
+import java.net.InetSocketAddress;
+
+/**
+ *    客户端的上下线
+ *
+ * @author Administrator
+ *
+ */
+@Slf4j
+public class NettyConnectHandler extends ChannelInboundHandlerAdapter {
+
+    /**
+     * 建立连接时
+     */
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
+        String clientIp = socketAddress.getAddress().getHostAddress();
+        int clientPort = socketAddress.getPort();
+        log.info("ip:{} port:{} connected",clientIp, clientPort);
+        ctx.fireChannelActive();
+    }
+
+    /**
+     * 关闭连接时
+     */
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
+        String clientIp = socketAddress.getAddress().getHostAddress();
+        int clientPort = socketAddress.getPort();
+        log.info("ip:{} port:{} disconnected",clientIp, clientPort);
+        String ID = ClientChanel.getIDFromIp(clientIp);
+        //设置客户端下线
+        ClientChanel.disConnect(ID);
+    }
+
+}

web/src/main/java/com/zhehekeji/web/service/client/NettyServer.java

@@ -0,0 +1,60 @@
+package com.zhehekeji.web.service.client;
+
+import com.zhehekeji.web.service.EmptyCheckService;
+import com.zhehekeji.web.service.PlcService;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import lombok.SneakyThrows;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+@Component
+public class NettyServer {
+
+    @Resource
+    private PlcService plcService;
+
+    public void CreateNettyServer(int port) throws InterruptedException {
+        Thread thread = new Thread(new Runnable() {
+            @SneakyThrows
+            @Override
+            public void run() {
+                EventLoopGroup bossGroup = new NioEventLoopGroup();
+                // 从线程组, 老板线程组会把任务丢给他，让手下线程组去做任务
+                EventLoopGroup workerGroup = new NioEventLoopGroup();
+
+                try {
+                    // netty服务器的创建, 辅助工具类，用于服务器通道的一系列配置
+                    ServerBootstrap serverBootstrap = new ServerBootstrap();
+                    serverBootstrap.group(bossGroup, workerGroup)           //绑定两个线程组
+                            .channel(NioServerSocketChannel.class)   //指定NIO的模式
+
+                            .childHandler(new ChannelInitializer<SocketChannel>() {
+                                @Override
+                                public void initChannel(SocketChannel ch) {
+                                    ch.pipeline().addLast(new NettyConnectHandler());
+                                    ch.pipeline().addLast(new Decoder(plcService));
+                                    ch.pipeline().addLast(new Encoder());
+                                }}); // 子处理器，用于处理workerGroup
+
+                    // 启动server，并且设置8088为启动的端口号，同时启动方式为同步
+                    ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
+
+                    // 监听关闭的channel，设置位同步方式
+                    channelFuture.channel().closeFuture().sync();
+                } finally {
+                    //退出线程组
+                    bossGroup.shutdownGracefully();
+                    workerGroup.shutdownGracefully();
+                }
+            }
+        });
+        thread.start();
+    }
+}

web/src/main/java/com/zhehekeji/web/service/client/TransmissionPojo.java

@@ -0,0 +1,156 @@
+package com.zhehekeji.web.service.client;
+
+import com.zhehekeji.web.entity.CheckLog;
+import com.zhehekeji.web.entity.Street;
+import lombok.Data;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+@Data
+public class TransmissionPojo {
+    private String data;
+    private String[] dataArray;
+    private String header;
+    private String streetNumber;//巷道标识符
+
+    private Integer row;//行
+    private Integer column;//列
+
+    private String pcd;//点位文件路径
+    private String taskId;
+    private String checkId;
+    private Integer direction;//左右
+    private Integer count = 0; //数量
+    private String category = ""; //品规
+    private Set<String> picsPath; //照片
+
+
+    private Integer result;
+    public TransmissionPojo(CheckLog checkLog , Street street){
+        this.checkId = checkLog.getId().toString();
+        this.streetNumber = street.getPlcId();
+        this.taskId = checkLog.getLotnum();
+        this.row = checkLog.getRow();
+        this.column = checkLog.getColumn();
+        this.direction = checkLog.getDirection();
+        this.count = checkLog.getCount();
+        if(checkLog.getCategory() != null && checkLog.getCategory().contains("/")) {
+            this.category = checkLog.getCategory().split("/")[0];
+        }else {
+            this.category = checkLog.getCategory();
+        }
+
+    }
+
+    /**
+     * 将给定数字转换为指定长度（默认3位）的字符，不足部分用“0”补全。
+     *
+     * @param number 需要转换的数字
+     * @param length 目标字符长度（默认3）
+     * @return 符合要求的字符串
+     */
+    public static String convertNumberToFixedLengthChars(int number, int length) {
+        if (length < 1) {
+            throw new IllegalArgumentException("Length must be a positive integer.");
+        }
+
+        String numStr = Integer.toString(number);
+        while (numStr.length() < length) {
+            numStr = "0" + numStr;
+        }
+        return numStr;
+    }
+    public TransmissionPojo( Street street,Integer row , Integer column,Integer direction,String orderId){
+        this.streetNumber = street.getPlcId();
+        this.row = row;
+
+        this.taskId = "0";
+        this.direction = direction;
+        this.column = column;
+        this.checkId =  convertNumberToFixedLengthChars(row, 3) + convertNumberToFixedLengthChars(column, 3);
+
+    }
+    public String toString(TransmissionType type) {
+        this.header = type.toString();
+        if (type == TransmissionType.ST) {
+            return "ST&" + streetNumber + "&" + checkId + "&" + taskId + "&" + direction ;
+        }else if (type == TransmissionType.GPS) {
+            return "GPS&" + streetNumber +"/"+row +"/"+column+ "&" + checkId + "&" + taskId + "&" + direction+ "&" + category + "&" + count;
+        }else if (type == TransmissionType.RTS) {
+            return "RTS&" +  streetNumber +"/"+row +"/"+column+  "&" + checkId + "&" + taskId + "&" + direction  + "&" + category + "&" + count;
+        }
+        return "";
+    }
+
+    public TransmissionPojo(String data) {
+        this.data = data;
+        data = data.replace(">", "");
+        this.dataArray = data.split("&");
+        this.header = dataArray[0];
+        if(dataArray[1].contains("/")) {
+            String[] dataArr = dataArray[1].split("/");
+            this.streetNumber = dataArr[0];
+            this.row = Integer.valueOf(dataArr[1]);
+            if(dataArr.length > 2) {
+                this.column = Integer.valueOf(dataArr[2]);
+            }
+        }else {
+            this.streetNumber = dataArray[1];
+        }
+        if (dataArray.length > 2) {
+            this.checkId = dataArray[2];
+            this.taskId = (dataArray[3]);
+            if(dataArray[4] != null && !"".equals(dataArray[4]) && !"null".equals(dataArray[4])) {
+                this.direction = Integer.valueOf(dataArray[4]);
+            }
+        }
+
+        if (dataArray.length > 5) {
+            if(dataArray[5] != null &&!"".equals(dataArray[5]) &&  !"null".equals(dataArray[5]) ) {
+
+                this.category = dataArray[5];
+            }
+            if( dataArray.length > 6 && dataArray[6] != null && !"".equals(dataArray[6]) && ! "null".equals(dataArray[5])) {
+                this.count = Integer.valueOf(dataArray[6]);
+            }
+        }
+        if (dataArray.length > 7) {
+            this.result = Integer.parseInt(dataArray[7])==0 ?0:1;
+            //不记录2d图像
+            //String path2D = "/fileData/"+this.taskId+"/"+this.checkId+"/"+this.direction+".jpg";
+            String path3D = "/fileData/"+this.taskId+"/"+this.checkId+"/"+(this.direction+2)+".PNG";
+            Set<String> strings = new HashSet<>();
+            //strings.add(path2D);
+            strings.add(path3D);
+
+            // 将数组中的元素添加到 Set 集合中
+            if(dataArray.length > 8 && dataArray[8].split(";").length>0) {
+                String[] s =  dataArray[8].split(";");
+                for (int i = 0; i < s.length; i++) {
+                    s[i] = s[i].replace("\\\\","/");
+                    s[i] = s[i].replace("\\","/");
+                    if(s[i].startsWith(".")) {
+                        s[i] = s[i].replaceFirst(".", ""); // 移除单个字符
+                    }
+                    // 或者
+                    // stringArray[i] = stringArray[i].replaceAll(regexToRemove, ""); // 使用正则表达式移除字符类别
+                }
+                Collections.addAll(strings,s);
+                strings.remove("");
+            }
+            this.picsPath = strings;
+
+        }
+        if(this.taskId != null && !"".equals(this.taskId) && this.checkId != null && !"".equals(this.checkId))
+            this.pcd = "/fileData/"+this.taskId+"/"+this.checkId+"/"+(this.direction+2)+".pcd";
+
+    }
+
+    public static void main(String[] args) {
+        TransmissionPojo transmissionPojo = new TransmissionPojo("GPE&002/1/18&7&2&123&41111&30&0&fff;.\\\\f.png;/f.png>");
+        System.out.println(transmissionPojo);
+    }
+
+}

web/src/main/java/com/zhehekeji/web/service/client/TransmissionType.java

@@ -0,0 +1,14 @@
+package com.zhehekeji.web.service.client;
+
+public enum TransmissionType {
+    ST   ,
+    GPS  ,
+    GPE  ,
+    SGPS  ,
+    SGPE  ,
+    RTS  ,
+    RTE   ,
+    EN   ,
+    ZHB  ;
+
+}