diff --git a/src/main/java/com/iot/modbus_rtcp/config/EquipmentIPProperties.java b/src/main/java/com/iot/modbus_rtcp/config/EquipmentIPProperties.java
index 1280aab..27e5f8d 100644
--- a/src/main/java/com/iot/modbus_rtcp/config/EquipmentIPProperties.java
+++ b/src/main/java/com/iot/modbus_rtcp/config/EquipmentIPProperties.java
@@ -36,6 +36,8 @@ public class EquipmentIPProperties {
if (gatewayHeartbeat.length() > 100) {
return false;
}
+ // 解决心跳包中可能存在的空格符和换行符
+ gatewayHeartbeat = HexUtil.bytesToHexString(StringUtils.trim(new String(HexUtil.hexStringToBytes(gatewayHeartbeat))).getBytes()).toUpperCase();
// 每分钟从数据库入库一次
if (this.lastSyncEquipment.getAcquire() < System.currentTimeMillis() - 60_000) {
this.jdbcTemplate.queryForList("select gateway_sn from device").forEach(map -> {
diff --git a/src/main/java/com/iot/modbus_rtcp/jobs/AutoCollectJobs.java b/src/main/java/com/iot/modbus_rtcp/jobs/AutoCollectJobs.java
index 4e2ff2d..702e837 100644
--- a/src/main/java/com/iot/modbus_rtcp/jobs/AutoCollectJobs.java
+++ b/src/main/java/com/iot/modbus_rtcp/jobs/AutoCollectJobs.java
@@ -22,6 +22,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
/**
* @author 王仕龙
@@ -66,10 +67,10 @@ public class AutoCollectJobs {
private final JdbcTemplate jdbcTemplate;
private final ModbusTCPController controller;
private final EquipmentIPProperties equipmentIPProperties;
- private final ThreadPoolExecutor collectThreadPool = new ThreadPoolExecutor(Math.min(10, Runtime.getRuntime().availableProcessors())
+ private final ThreadPoolExecutor collectThreadPool = new ThreadPoolExecutor(Math.min(50, Runtime.getRuntime().availableProcessors())
, Math.min(100, Runtime.getRuntime().availableProcessors()), 60, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(5000), new ThreadFactory() {
- AtomicInteger index = new AtomicInteger(1);
+ final AtomicInteger index = new AtomicInteger(1);
@Override
public Thread newThread(Runnable runnable) {
@@ -85,53 +86,86 @@ public class AutoCollectJobs {
int pageIndex = 0;
int pageSize = 1000;
long timestamp = System.currentTimeMillis();
- List> resultList = null;
- while (Objects.isNull(resultList) || resultList.size() >= pageSize) {
- resultList = this.jdbcTemplate.queryForList(SQL + " LIMIT " + (pageIndex++ * pageSize) + ", " + pageSize);
+ while (true) {
+ List> resultList = this.jdbcTemplate.queryForList(SQL + " LIMIT " + (pageIndex++ * pageSize) + ", " + pageSize);
if (ObjectUtils.isEmpty(resultList)) {
- continue;
+ break;
}
- resultList.stream()
- .filter(item -> {
- String gatewaySn = StringUtils.trim(MapUtil.getStr(item, "identifier"));
- String gatewayIdentifier = this.equipmentIPProperties.put(gatewaySn);
- if (Objects.nonNull(ModbusTCPController.nettyServer.getGroup()
- .get(gatewayIdentifier))) {
- item.put("identifier", gatewayIdentifier);
- return true;
- }
- log.warn("Gateway {} is disconnected and does not collect data", gatewaySn);
- return false;
- })
- .collect(Collectors.groupingBy(item -> StringUtils.trim(MapUtil.getStr(item, "deviceId"))))
- .forEach((deviceId, commandList) -> {
- Map>> refTypeCommandListMap = commandList.stream()
- .collect(Collectors.groupingBy(item -> MapUtil.getStr(item, "refType")));
- List> deviceCommandList = refTypeCommandListMap.get(COMMAND_REF_TYPE_DEVICE);
- if (ObjectUtils.isEmpty(deviceCommandList)) {
- deviceCommandList = refTypeCommandListMap.get(COMMAND_REF_TYPE_PRODUCT_CODE);
- }
- if (ObjectUtils.isNotEmpty(deviceCommandList)) {
- List 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));
+ this.collectThreadPool.execute(() -> {
+ this.controller.collect(resultList.stream()
+ .filter(item -> {
+ String gatewaySn = StringUtils.trim(MapUtil.getStr(item, "identifier"));
+ String gatewayIdentifier = this.equipmentIPProperties.put(gatewaySn);
+ if (Objects.nonNull(ModbusTCPController.nettyServer.getGroup()
+ .get(gatewayIdentifier))) {
+ item.put("identifier", gatewayIdentifier);
+ return true;
}
- }
- });
+ log.warn("Gateway {} is disconnected and does not collect data", gatewaySn);
+ return false;
+ })
+ .collect(Collectors.groupingBy(item -> StringUtils.trim(MapUtil.getStr(item, "deviceId"))))
+ .entrySet()
+ .stream()
+ .flatMap(entry -> {
+ List> commandList = entry.getValue();
+ Map>> refTypeCommandListMap = commandList.stream()
+ .collect(Collectors.groupingBy(item -> MapUtil.getStr(item, "refType")));
+ List> deviceCommandList = refTypeCommandListMap.get(COMMAND_REF_TYPE_DEVICE);
+ if (ObjectUtils.isEmpty(deviceCommandList)) {
+ deviceCommandList = refTypeCommandListMap.get(COMMAND_REF_TYPE_PRODUCT_CODE);
+ }
+ if (ObjectUtils.isNotEmpty(deviceCommandList)) {
+ return 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();
+ });
+ }
+ return Stream.empty();
+ })
+ .collect(Collectors.toList()));
+// .forEach((deviceId, commandList) -> {
+// Map>> refTypeCommandListMap = commandList.stream()
+// .collect(Collectors.groupingBy(item -> MapUtil.getStr(item, "refType")));
+// List> deviceCommandList = refTypeCommandListMap.get(COMMAND_REF_TYPE_DEVICE);
+// if (ObjectUtils.isEmpty(deviceCommandList)) {
+// deviceCommandList = refTypeCommandListMap.get(COMMAND_REF_TYPE_PRODUCT_CODE);
+// }
+// if (ObjectUtils.isNotEmpty(deviceCommandList)) {
+// List 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;
+ }
}
}
}
diff --git a/src/main/java/com/iot/modbus_rtcp/netty/SyncPriorityChannel.java b/src/main/java/com/iot/modbus_rtcp/netty/SyncPriorityChannel.java
index 5ae63a3..be25fc4 100644
--- a/src/main/java/com/iot/modbus_rtcp/netty/SyncPriorityChannel.java
+++ b/src/main/java/com/iot/modbus_rtcp/netty/SyncPriorityChannel.java
@@ -8,6 +8,7 @@ import io.netty.channel.ChannelPromise;
import io.netty.channel.socket.SocketChannel;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
@@ -150,8 +151,14 @@ public class SyncPriorityChannel implements Runnable {
}
public void setIdentifier(String identifier) {
- this.identifier = identifier;
- this.setCollectQueue();
+ if (StringUtils.isBlank(this.identifier)) {
+ this.identifier = identifier;
+ this.setCollectQueue();
+ } else if (!StringUtils.equals(this.identifier, identifier)) {
+ throw new RuntimeException("多个设备对应同一心跳包，当前已绑定心跳信息："
+ + new String(HexUtil.hexStringToBytes(identifier)) + "[" + identifier + "] --> "
+ + this.channel.remoteAddress().getHostString() + ":" + this.channel.remoteAddress().getPort());
+ }
}
public void setCollectQueue() {