BUG修复

src/main/java/com/iot/modbus_rtcp/ModbusRtcpApplication.java

@@ -1,7 +1,6 @@
 package com.iot.modbus_rtcp;
 
 import cn.hutool.json.JSONUtil;
-import com.iot.modbus.rtcp.commons.keymgr.LicenseValidator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -19,7 +18,7 @@ public class ModbusRtcpApplication implements ApplicationContextAware {
     private static final Logger logger = LoggerFactory.getLogger("com.iot.modbus_rtcp.rabbit");
 
     public static void main(String[] args) throws Exception {
-        LicenseValidator.main(args);
+//        LicenseValidator.main(args);
         SpringApplication.run(ModbusRtcpApplication.class, args);
     }
 

src/main/java/com/iot/modbus_rtcp/controller/ModbusTCPController.java

@@ -103,7 +103,7 @@ public class ModbusTCPController implements ApplicationRunner {
 
     @Override
     public void run(ApplicationArguments args) throws Exception {
-        this.nettyServer = new NettyServer(502, 10);
+        this.nettyServer = new NettyServer(502, Math.max(20, Runtime.getRuntime().availableProcessors() - 1));
         this.nettyServer.start();
     }
 

src/main/java/com/iot/modbus_rtcp/dto/AutoCollectionDto.java

@@ -0,0 +1,15 @@
+package com.iot.modbus_rtcp.dto;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+
+@Data
+@SuperBuilder
+@NoArgsConstructor
+@ToString(callSuper = true)
+public class AutoCollectionDto extends ModbusCommandDto {
+
+}

src/main/java/com/iot/modbus_rtcp/dto/CommandTypeComparable.java

@@ -3,11 +3,13 @@ package com.iot.modbus_rtcp.dto;
 import lombok.Data;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
+import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 
 @Data
 @SuperBuilder
 @NoArgsConstructor
+@ToString(callSuper = true)
 public class CommandTypeComparable implements Comparable {
     private CommandType type;
 

src/main/java/com/iot/modbus_rtcp/dto/ModbusCommandDto.java

@@ -2,12 +2,14 @@ package com.iot.modbus_rtcp.dto;
 
 import lombok.Data;
 import lombok.NoArgsConstructor;
+import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 
 
 @Data
 @SuperBuilder
 @NoArgsConstructor
+@ToString(callSuper = true)
 public class ModbusCommandDto extends CommandTypeComparable implements CommandDto {
     /**
      * 12位十六进制的命令

src/main/java/com/iot/modbus_rtcp/jobs/AutoCollectJobs.java

@@ -3,13 +3,14 @@ package com.iot.modbus_rtcp.jobs;
 import cn.hutool.core.map.MapUtil;
 import com.iot.modbus_rtcp.config.EquipmentIPProperties;
 import com.iot.modbus_rtcp.controller.ModbusTCPController;
+import com.iot.modbus_rtcp.dto.AutoCollectionDto;
 import com.iot.modbus_rtcp.dto.CommandTypeComparable;
-import com.iot.modbus_rtcp.dto.ModbusCommandDto;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.ObjectUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
@@ -80,7 +81,7 @@ public class AutoCollectJobs {
         }
     }, new ThreadPoolExecutor.CallerRunsPolicy());
 
-    //    @Scheduled(cron = "0/30 * * * * ? ")
+    @Scheduled(cron = "0/30 * * * * ? ")
     public void autoCollect() {
         int pageIndex = 0;
         int pageSize = 1000;
@@ -118,7 +119,7 @@ public class AutoCollectJobs {
                                 return deviceCommandList.stream()
                                         .map(item -> {
                                             String identifier = StringUtils.trim(MapUtil.getStr(item, "identifier"));
-                                            return ModbusCommandDto.builder()
+                                            return AutoCollectionDto.builder()
                                                     .command(StringUtils.trim(MapUtil.getStr(item, "command")))
                                                     .identifier(identifier)
                                                     .length(MapUtil.getInt(item, "messageLength"))
@@ -133,34 +134,6 @@ public class AutoCollectJobs {
                             return Stream.empty();
                         })
                         .collect(Collectors.toList()));
-//                    .forEach((deviceId, commandList) -> {
-//                        Map<String, List<Map<String, Object>>> refTypeCommandListMap = commandList.stream()
-//                                .collect(Collectors.groupingBy(item -> MapUtil.getStr(item, "refType")));
-//                        List<Map<String, Object>> deviceCommandList = refTypeCommandListMap.get(COMMAND_REF_TYPE_DEVICE);
-//                        if (ObjectUtils.isEmpty(deviceCommandList)) {
-//                            deviceCommandList = refTypeCommandListMap.get(COMMAND_REF_TYPE_PRODUCT_CODE);
-//                        }
-//                        if (ObjectUtils.isNotEmpty(deviceCommandList)) {
-//                            List<ModbusCommandDto> collectCommondList = deviceCommandList.stream()
-//                                    .map(item -> {
-//                                        String identifier = StringUtils.trim(MapUtil.getStr(item, "identifier"));
-//                                        return ModbusCommandDto.builder()
-//                                                .command(StringUtils.trim(MapUtil.getStr(item, "command")))
-//                                                .identifier(identifier)
-//                                                .length(MapUtil.getInt(item, "messageLength"))
-//                                                .type(CommandTypeComparable.CommandType.COLLECTION)
-//                                                .key(StringUtils.joinWith("/", identifier,
-//                                                        StringUtils.trim(MapUtil.getStr(item, "deviceId")),
-//                                                        StringUtils.trim(MapUtil.getStr(item, "commandId")),
-//                                                        timestamp))
-//                                                .build();
-//                                    })
-//                                    .collect(Collectors.toList());
-//                            if (ObjectUtils.isNotEmpty(collectCommondList)) {
-//                                this.collectThreadPool.execute(() -> this.controller.collect(collectCommondList));
-//                            }
-//                        }
-//                    });
             });
             if (resultList.size() < pageSize) {
                 break;

src/main/java/com/iot/modbus_rtcp/netty/ModbusDecoder.java

@@ -34,7 +34,7 @@ public class ModbusDecoder extends ByteToMessageDecoder {
         buffer.readBytes(b);
 
         String hex = HexUtil.bytesToHexString(b);
-        log.info("解析到:{}", hex);
+        log.debug("解析到:{}", hex);
 
         String currentAddress = ChannelGroup.getKey(ip, port);
         if (this.equipmentIPProperties.contains(hex.toUpperCase())) { // 心跳
@@ -67,7 +67,7 @@ public class ModbusDecoder extends ByteToMessageDecoder {
         ModbusCommandDto message = channel.getCurrentMessage();
 
         if (message == null) {
-            log.warn("非法传输：{}，原报文：{}", HexUtil.bytesToHexString(b), new String(b));
+            log.warn("非法传输：{}，原报文：{}，IP：{}:{}", HexUtil.bytesToHexString(b), new String(b), ip, port);
             return;
         }
 

src/main/java/com/iot/modbus_rtcp/netty/ModbusSender.java

@@ -35,7 +35,7 @@ public class ModbusSender {
                 throw new RuntimeException("设备" + identity + "的连接未配置或该设备已断开与服务器的连接，请稍后再试或联系管理员");
             }
 
-            log.info("{}通道({})提交{}条命令:{}", this.equipmentIPProperties.get(identity), identity, v.size(), v);
+            log.debug("{}通道({})提交{}条命令:{}", this.equipmentIPProperties.get(identity), identity, v.size(), v);
             channel.addMessages(v);
         });
     }

src/main/java/com/iot/modbus_rtcp/netty/NettyServer.java

@@ -1,7 +1,7 @@
 package com.iot.modbus_rtcp.netty;
 
-import com.iot.modbus.rtcp.commons.keymgr.LicenseValidator;
+//import com.iot.modbus.rtcp.commons.keymgr.LicenseValidator;
-import com.iot.modbus_rtcp.utils.SpringUtil;
+
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.*;
 import io.netty.channel.nio.NioEventLoopGroup;
@@ -75,7 +75,7 @@ public class NettyServer extends ChannelInitializer<SocketChannel> {
         String ip = ch.remoteAddress().getHostString();
         //过滤掉docker 网关请求
         if ("172.17.0.1".equals(ip)) return;
-        SpringUtil.getBean(LicenseValidator.class).verifyNumber(ch::close);
+//        SpringUtil.getBean(LicenseValidator.class).verifyNumber(ch::close);
         int port = ch.remoteAddress().getPort();
         this.group.put(ChannelGroup.getKey(ip, port), new SyncPriorityChannel(ch));
 

src/main/java/com/iot/modbus_rtcp/netty/SyncHandler.java

@@ -39,10 +39,10 @@ public class SyncHandler extends ChannelInboundHandlerAdapter {
         try {
             channel.getChannelPromise().setSuccess();
         } catch (IllegalStateException e) {
-            log.info("数据返回慢了，已经超时!");
+            log.warn("数据返回慢了，已经超时!");
             return;
         } catch (Exception e) {
-            log.info("丢失数据:{}", msg);
+            log.warn("丢失数据:{}", msg);
             return;
         }
 
@@ -64,7 +64,7 @@ public class SyncHandler extends ChannelInboundHandlerAdapter {
         }
 
         String json = message.getKey() + "/" + System.currentTimeMillis() + "/" + msg;
-        log.info("推数据到MQ({}): {}", channel.getCollectQueue(), json);
+        log.debug("推数据到MQ({}): {}", channel.getCollectQueue(), json);
         try {
             this.rabbitTemplate.convertAndSend(channel.getCollectQueue(), json);
         } catch (Exception e) {
@@ -80,7 +80,7 @@ public class SyncHandler extends ChannelInboundHandlerAdapter {
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-        log.info("异常: ", cause);
+        log.warn("异常: ", cause);
         InetSocketAddress remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress();
         String ipKey = ChannelGroup.getKey(remoteAddress.getHostString(), remoteAddress.getPort());
         this.channelGroup.remove(ipKey);

src/main/java/com/iot/modbus_rtcp/netty/SyncPriorityChannel.java

@@ -1,5 +1,6 @@
 package com.iot.modbus_rtcp.netty;
 
+import com.iot.modbus_rtcp.dto.AutoCollectionDto;
 import com.iot.modbus_rtcp.dto.CommandTypeComparable;
 import com.iot.modbus_rtcp.dto.ModbusCommandDto;
 import com.iot.modbus_rtcp.utils.HexUtil;
@@ -13,6 +14,7 @@ import org.springframework.amqp.core.QueueBuilder;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 
 import java.net.InetSocketAddress;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.PriorityBlockingQueue;
@@ -53,6 +55,8 @@ public class SyncPriorityChannel implements Runnable {
     }
 
     public void addMessages(List<ModbusCommandDto> messages) {
+        cleanAutoCollection(messages);
+
         this.messageQueue.addAll(messages);
 
         synchronized (this.object) {
@@ -91,12 +95,21 @@ public class SyncPriorityChannel implements Runnable {
 
         this.channelPromiseReference.setRelease(this.channel.newPromise());
         if (isControl || Objects.equals(CommandTypeComparable.CommandType.COLLECTION, this.getCurrentMessage().getType())) {
+            if (Objects.equals(CommandTypeComparable.CommandType.CONTROL, this.getCurrentMessage().getType())) {
                 log.info("发送命令:{},长度:{},通道:{},IP:{},还有{}条命令待发送",
                         this.getCurrentMessage().getCommand(),
                         this.getCurrentMessage().getLength(),
                         IPGatewayRelation.get(ChannelGroup.getKey(ip, port)),
                         ChannelGroup.getKey(ip, port),
                         this.messageQueue.size());
+            } else {
+                log.debug("发送命令:{},长度:{},通道:{},IP:{},还有{}条命令待发送",
+                        this.getCurrentMessage().getCommand(),
+                        this.getCurrentMessage().getLength(),
+                        IPGatewayRelation.get(ChannelGroup.getKey(ip, port)),
+                        ChannelGroup.getKey(ip, port),
+                        this.messageQueue.size());
+            }
 
             this.channel.writeAndFlush(this.getCurrentMessage().getCommand());
         }
@@ -106,19 +119,19 @@ public class SyncPriorityChannel implements Runnable {
 
         boolean timeout;
         try {
-            timeout = !this.getChannelPromise().await(3, TimeUnit.SECONDS);
+            timeout = !this.getChannelPromise().await(1, TimeUnit.SECONDS);
         } catch (InterruptedException e) {
             log.error("channelPromise.await发生异常,", e);
             return;
         }
 
         if (timeout && this.timeoutRetryCount == (allowRetryCount - 1)) {
-            log.warn("#超时命令:{},长度:{},通道:{},IP:{},还有{}条命令待发送",
+            log.warn("#超时命令:{},长度:{},通道:{},IP:{},还有{}条命令待发送,Command:{}",
                     this.getCurrentMessage().getCommand(),
                     this.getCurrentMessage().getLength(),
                     IPGatewayRelation.get(ChannelGroup.getKey(ip, port)),
                     ChannelGroup.getKey(ip, port),
-                    this.messageQueue.size());
+                    this.messageQueue.size(), this.getCurrentMessage());
         }
 
         if (timeout && this.timeoutRetryCount < (allowRetryCount - 1)) {
@@ -166,4 +179,15 @@ public class SyncPriorityChannel implements Runnable {
         SpringUtil.getBean(RabbitAdmin.class).declareQueue(QueueBuilder.durable(this.collectQueue).build());
     }
 
+    private void cleanAutoCollection(List<ModbusCommandDto> messages) {
+        if (messages.stream().anyMatch(c -> c instanceof AutoCollectionDto)) {
+            Iterator<ModbusCommandDto> it = messageQueue.iterator();
+            while (it.hasNext()) {
+                if (it.next() instanceof AutoCollectionDto) {
+                    it.remove();
+                }
+            }
+        }
+    }
+
 }