Skip to content

Z
zhmes-agecal
Commit 9406e0a9 authored by wanghao's avatar wanghao
Browse files
Options

1 测试 上电后通信 和 最终完成 定时任务功能

parent f5156dc3
Hide whitespace changes
Inline Side-by-side
Showing with 66 additions and 36 deletions
zhmes-agecal-system/src/main/java/com/zehong/system/modbus/business/DeviceStatusReaderAndTimeSetter.java
View file @ 9406e0a9
...@@ -171,39 +171,73 @@ public class DeviceStatusReaderAndTimeSetter { ...@@ -171,39 +171,73 @@ public class DeviceStatusReaderAndTimeSetter {
TcpMaster tcpMaster = (TcpMaster) master; TcpMaster tcpMaster = (TcpMaster) master;
try { try {
// 1. 获取TcpMaster的private字段"transport"（类型：TcpTransport） // 1. 关键：内部类路径用 "$" 连接外部类（必须与你的Modbus4j版本匹配）
String transportClassName = "sun.rmi.transport.tcp.TcpTransport";
Class<?> transportClass = Class.forName(transportClassName);
log.debug("成功加载TcpMaster内部类：{}", transportClassName);
// 2. 获取TcpMaster的private字段 "transport"（存储内部类实例）
Field transportField = TcpMaster.class.getDeclaredField("transport"); Field transportField = TcpMaster.class.getDeclaredField("transport");
transportField.setAccessible(true); // 取消私有字段访问检查 transportField.setAccessible(true); // 取消私有字段访问限制
Object transport = transportField.get(tcpMaster); Object transport = transportField.get(tcpMaster);
if (transport == null) { if (transport == null) {
log.error("TcpMaster的transport字段为null"); log.error("TcpMaster的transport字段为null（连接未初始化？）");
return null; return null;
} }
// 2. 获取TcpTransport的private字段"socket"（类型：Socket） // 3. 获取内部类TcpTransport的private字段 "socket"（底层Socket）
// 注意：TcpTransport是Modbus4j的内部类，包路径为com.serotonin.modbus4j.ip.tcp.TcpTransport // 注意：若报错"NoSuchFieldException"，需打开Modbus4j源码确认字段名（如"clientSocket"）
Class<?> transportClass = Class.forName("com.serotonin.modbus4j.ip.tcp.TcpTransport");
Field socketField = transportClass.getDeclaredField("socket"); Field socketField = transportClass.getDeclaredField("socket");
socketField.setAccessible(true); // 取消私有字段访问检查 socketField.setAccessible(true); // 取消私有字段访问限制
Socket socket = (Socket) socketField.get(transport); Socket socket = (Socket) socketField.get(transport);
log.debug("成功获取Socket：{}", socket); log.debug("成功获取Socket：IP={}, 本地端口={}", socket.getInetAddress(), socket.getLocalPort());
return socket; return socket;
} catch (ClassNotFoundException e) {
log.error("=== 致命错误：TcpTransport类未找到！===\n" +
"原因：1. Modbus4j版本不兼容；2. 内部类路径错误（需确认是否为 TcpMaster$TcpTransport）\n" +
"解决方案：1. 打开Modbus4j源码，找到TcpMaster的内部类TcpTransport，复制完整类名；\n" +
" 2. 将 transportClassName 改为实际路径（如 com.xxx.TcpMaster$TcpTransport）", e);
} catch (NoSuchFieldException e) { } catch (NoSuchFieldException e) {
log.error("反射获取字段失败（可能Modbus4j版本不兼容），请检查字段名", e); log.error("=== 致命错误：字段未找到！===\n" +
"原因：1. TcpMaster的transport字段名错误；2. TcpTransport的socket字段名错误\n" +
"解决方案：1. 打开TcpMaster源码，确认传输字段名（如'tcpTransport'）；\n" +
" 2. 打开TcpTransport源码，确认Socket字段名（如'clientSocket'）", e);
} catch (IllegalAccessException e) { } catch (IllegalAccessException e) {
log.error("反射访问字段权限失败", e); log.error("反射访问字段权限失败（可能被安全管理器拦截）", e);
} catch (ClassNotFoundException e) {
log.error("TcpTransport类未找到（Modbus4j版本不兼容）", e);
} catch (Exception e) { } catch (Exception e) {
log.error("获取底层Socket异常", e); log.error("获取底层Socket异常", e);
} }
return null; return null;
} }
/**
* 销毁ModbusMaster并关闭底层Socket（确保资源彻底释放）
*/
private static void destroyModbusMaster(ModbusMaster master, int deviceId) {
if (master == null) return;
try {
// 1. 先关闭底层Socket（反射获取）
Socket socket = getUnderlyingSocket(master);
if (socket != null && !socket.isClosed()) {
socket.close();
log.debug("设备{}: 底层Socket已强制关闭", deviceId);
}
} catch (Exception e) {
log.error("设备{}: 关闭底层Socket异常", deviceId, e);
} finally {
// 2. 再销毁ModbusMaster
try {
master.destroy();
log.debug("设备{}: ModbusMaster已销毁", deviceId);
} catch (Exception e) {
log.error("设备{}: 销毁ModbusMaster异常", deviceId, e);
}
}
}
/** /**
* 启动多设备监控（无反射，兼容所有版本） * 启动多设备监控（反射方案，确保资源释放）
*/ */
public void startMultiDeviceMonitoring( public void startMultiDeviceMonitoring(
String ip, int port, List<Integer> deviceIds, String ip, int port, List<Integer> deviceIds,
...@@ -215,7 +249,7 @@ public class DeviceStatusReaderAndTimeSetter { ...@@ -215,7 +249,7 @@ public class DeviceStatusReaderAndTimeSetter {
} }
final CountDownLatch latch = new CountDownLatch(deviceIds.size()); final CountDownLatch latch = new CountDownLatch(deviceIds.size());
log.info("开始监控设备：IP={}, 端口={}, 设备数={}（活跃线程：{}）", log.info("开始监控设备：IP={}, 端口={}, 设备数={}（线程池活跃线程：{}）",
ip, port, deviceIds.size(), ((ThreadPoolExecutor) FIXED_THREAD_POOL).getActiveCount()); ip, port, deviceIds.size(), ((ThreadPoolExecutor) FIXED_THREAD_POOL).getActiveCount());
for (int deviceId : deviceIds) { for (int deviceId : deviceIds) {
...@@ -223,60 +257,56 @@ public class DeviceStatusReaderAndTimeSetter { ...@@ -223,60 +257,56 @@ public class DeviceStatusReaderAndTimeSetter {
FIXED_THREAD_POOL.submit(() -> { FIXED_THREAD_POOL.submit(() -> {
ModbusMaster threadMaster = null; ModbusMaster threadMaster = null;
try { try {
// 1. 创建连接（每次任务独立连接，避免复用导致泄漏） // 1. 创建独立连接（每个设备任务一个连接，避免复用泄漏）
threadMaster = createModbusMaster(ip, port); threadMaster = createModbusMaster(ip, port);
if (threadMaster == null) { if (threadMaster == null) {
log.error("设备{}: Modbus连接创建失败", devId); log.error("设备{}: Modbus连接创建失败，终止任务", devId);
recordAlarm("03", "ip:" + ip + ",port:" + port + ",deviceId:" + devId, "连接创建失败"); recordAlarm("03", "ip:" + ip + ",port:" + port + ",deviceId:" + devId, "连接创建失败");
return; return;
} }
// 2. 读取数据（带超时控制的重试） // 2. 读取数据（带重试和异常处理）
int[] result = readWithConditionalRetry(threadMaster, ip, port, devId, stopCondition); int[] result = readWithConditionalRetry(threadMaster, ip, port, devId, stopCondition);
// 3. 处理结果 // 3. 回调处理结果（判空避免NPE）
if (resultHandler != null) { if (resultHandler != null) {
resultHandler.accept(new DeviceStatusReaderDto(ip, port, devId, result)); resultHandler.accept(new DeviceStatusReaderDto(ip, port, devId, result));
} }
} catch (ModbusInitException e) { } catch (ModbusInitException e) {
log.error("设备{}: Modbus连接初始化失败", devId, e); log.error("设备{}: Modbus连接初始化失败", devId, e);
recordAlarm("03", "ip:" + ip + ",port:" + port + ",deviceId:" + devId, "连接初始化失败：" + e.getMessage()); recordAlarm("03", "ip:" + ip + ",port:" + port + ",deviceId:" + devId,
"连接初始化失败：" + e.getMessage());
} catch (Throwable e) { } catch (Throwable e) {
log.error("设备{}: 监控任务异常", devId, e); log.error("设备{}: 监控任务致命异常", devId, e);
String alarmMsg = "监控任务异常：" + e.getMessage(); String alarmMsg = e instanceof OutOfMemoryError ? "内存溢出：" + e.getMessage() : "监控任务异常：" + e.getMessage();
recordAlarm("03", "ip:" + ip + ",port:" + port + ",deviceId:" + devId, alarmMsg); recordAlarm("03", "ip:" + ip + ",port:" + port + ",deviceId:" + devId, alarmMsg);
} finally { } finally {
// 关键：优先销毁连接，确保资源释放（兼容所有版本） // 关键：无论成功/失败，都强制销毁连接（避免资源泄漏）
if (threadMaster != null) { destroyModbusMaster(threadMaster, devId);
try {
threadMaster.destroy();
log.debug("设备{}: Modbus连接已销毁", devId);
} catch (Exception e) {
log.error("设备{}: Modbus连接销毁失败", devId, e);
}
}
latch.countDown(); latch.countDown();
log.info("设备{}: 监控任务完成（剩余任务：{}）", devId, latch.getCount()); log.info("设备{}: 监控任务完成（剩余任务：{}）", devId, latch.getCount());
} }
}); });
} }
// 等待任务完成，超时后记录告警 // 等待所有任务完成，超时后强制中断（避免长期阻塞）
try { try {
if (!latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)) { if (!latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
log.warn("IP{}: 部分设备监控超时（未完成：{}）", ip, latch.getCount()); log.warn("IP{}: 部分设备监控超时（未完成任务数：{}），强制中断", ip, latch.getCount());
recordAlarm("03", "ip:" + ip + ",port:" + port, "部分设备监控超时（未完成：" + latch.getCount() + "）"); recordAlarm("03", "ip:" + ip + ",port:" + port,
// 超时后主动关闭所有线程池任务（避免长期阻塞） "部分设备监控超时（未完成：" + latch.getCount() + "）");
// 超时后关闭线程池（避免任务堆积）
FIXED_THREAD_POOL.shutdownNow(); FIXED_THREAD_POOL.shutdownNow();
} }
} catch (InterruptedException e) { } catch (InterruptedException e) {
log.error("IP{}: 监控任务被中断", ip, e); log.error("IP{}: 监控任务被中断", ip, e);
Thread.currentThread().interrupt(); Thread.currentThread().interrupt(); // 恢复中断状态
} }
} }
/** /**
* 统一告警记录（抽离避免重复代码） * 统一告警记录（抽离避免重复代码）
*/ */
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment