Skip to content

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

1 指令指令完成，但是 没有检测到机械臂完成。

parent 985bc9d8
Hide whitespace changes
Inline Side-by-side
Showing with 101 additions and 322 deletions
zhmes-agecal-system/src/main/java/com/zehong/system/task/DeviceCommunicationJob.java
View file @ d34ebdc1
......@@ -12,29 +12,25 @@ import com.serotonin.modbus4j.msg.ReadHoldingRegistersResponse;
import com.zehong.system.domain.TEquipmentAlarmData;
import com.zehong.system.domain.TStoreyInfo;
import com.zehong.system.mapper.TStoreyInfoMapper;
import com.zehong.system.modbus.business.DeviceStatusReaderAndTimeSetter;
import com.zehong.system.modbus.handler.ModbusResultHandler;
import com.zehong.system.modbus.handler.dto.DeviceStatusReaderDto;
import com.zehong.system.modbus.util.Modbus4jUtils;
import com.zehong.system.service.ITEquipmentAlarmDataService;
import org.apache.commons.lang3.StringUtils;
import org.quartz.*;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.Job;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.lang.reflect.Field;
import java.net.Socket;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Predicate;
/**
* @author lenovo
* @date 2025/6/25
......@@ -44,14 +40,6 @@ import java.util.function.Predicate;
@DisallowConcurrentExecution
public class DeviceCommunicationJob implements Job {
private static final Logger log = LoggerFactory.getLogger(DeviceCommunicationJob.class);
// 超时控制（确保在Cron周期内完成）
private static final int TOTAL_TIMEOUT_SECONDS = 120; // 2分钟总超时
private static final int PORT_TIMEOUT_SECONDS = 30; // 单个端口30秒超时
public static final int RETRY_DELAY = 200; // 重试间隔200ms（缩短等待）
// 简化线程池配置
private final ExecutorService deviceExecutor = Executors.newFixedThreadPool(4);
// 工厂（单例）
private static final ModbusFactory modbusFactory = new ModbusFactory();
public static final int MAX_RETRIES = 1; // 仅1次重试（减少总耗时）
......@@ -60,8 +48,8 @@ public class DeviceCommunicationJob implements Job {
public static final int REGISTER_COUNT = 10;
@Resource
private ITEquipmentAlarmDataService alarmDataService;
@Resource
private DeviceStatusReaderAndTimeSetter deviceStatusReaderAndTimeSetter;
// 移除静态线程池，使用单线程顺序执行
private final ExecutorService sequentialExecutor = Executors.newSingleThreadExecutor();
@Resource
private TStoreyInfoMapper tStoreyInfoMapper;
......@@ -70,335 +58,125 @@ public class DeviceCommunicationJob implements Job {
@Override
public void execute(JobExecutionContext context) {
// 关键：确保任何异常都不会传播到Quartz框架
log.info("=== DeviceCommunicationJob 开始执行 ===");
try {
executeSafely(context);
} catch (Throwable e) {
// 这里捕获所有异常，确保不会传播到Quartz
log.error("DeviceCommunicationJob执行过程中发生未预期异常，但已被捕获避免触发器ERROR状态", e);
String storeyIdFromContext = getStoreyIdFromContext(context);
if(StringUtils.isNotBlank(storeyIdFromContext)) {
Long fStoreyId = Long.parseLong(storeyIdFromContext);
TStoreyInfo tStoreyInfo = tStoreyInfoMapper.selectTStoreyInfoById(fStoreyId);
recordAlarm(tStoreyInfo,"任务执行异常（已捕获）: " + e.getMessage());
}
// 使用单线程顺序执行，避免并发问题
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
executeJobSafely(context);
}, sequentialExecutor);
// 设置总超时
future.get(180, TimeUnit.SECONDS); // 3分钟总超时
log.info("=== DeviceCommunicationJob 正常完成 ===");
} catch (TimeoutException e) {
log.error("任务执行超时", e);
recordAlarm(context, "任务执行超时（3分钟）");
} catch (Exception e) {
log.error("任务执行异常（已捕获，不会影响触发器）", e);
recordAlarm(context, "任务执行异常: " + e.getMessage());
}
}
private void executeSafely(JobExecutionContext context) {
long startTime = System.currentTimeMillis();
String fStoreyIdStr = null;
TStoreyInfo tStoreyInfo = null;
private void executeJobSafely(JobExecutionContext context) {
try {
// 参数提取和验证
// 1. 提取参数
JobDataMap data = context.getJobDetail().getJobDataMap();
fStoreyIdStr = data.getString("fStoreyId");
String fStoreyIdStr = data.getString("fStoreyId");
if (StringUtils.isBlank(fStoreyIdStr)) {
log.warn("fStoreyId参数为空，跳过执行");
log.warn("fStoreyId参数为空");
return;
}
Long fStoreyId = Long.parseLong(fStoreyIdStr);
tStoreyInfo = tStoreyInfoMapper.selectTStoreyInfoById(fStoreyId);
if (tStoreyInfo == null) {
log.warn("设备信息不存在: fStoreyId={}", fStoreyId);
// 清理无效任务
cleanupInvalidJob(context, fStoreyId);
return;
}
TStoreyInfo tStoreyInfo = tStoreyInfoMapper.selectTStoreyInfoById(fStoreyId);
if (StringUtils.isBlank(tStoreyInfo.getfIp())) {
log.warn("设备IP为空: fStoreyId={}", fStoreyId);
recordAlarm(tStoreyInfo,"设备IP为空");
if (tStoreyInfo == null || StringUtils.isBlank(tStoreyInfo.getfIp())) {
log.warn("设备信息无效: fStoreyId={}", fStoreyId);
return;
}
// 执行设备通信（带超时控制）
executeDeviceCommunicationWithTimeout(tStoreyInfo, fStoreyId);
long costTime = System.currentTimeMillis() - startTime;
log.info("设备通信任务成功完成: fStoreyId={}, 耗时={}ms", fStoreyId, costTime);
// 2. 顺序执行三个端口（不要并行）
executePortSequentially(tStoreyInfo, fStoreyId);
} catch (NumberFormatException e) {
log.warn("fStoreyId格式错误: {}", fStoreyIdStr, e);
recordAlarm(tStoreyInfo, "fStoreyId格式错误: " + e.getMessage());
} catch (Exception e) {
log.error("设备通信任务执行异常: fStoreyIdStr={}", fStoreyIdStr, e);
recordAlarm(tStoreyInfo, "任务执行异常: " + e.getMessage());
// 注意：这里不再抛出异常！
log.error("executeJobSafely内部异常", e);
// 这里不抛出异常，只是记录日志
}
}
private void executeDeviceCommunicationWithTimeout(TStoreyInfo tStoreyInfo, Long fStoreyId) {
private void executePortSequentially(TStoreyInfo tStoreyInfo, Long fStoreyId) {
String ip = tStoreyInfo.getfIp();
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
executeDeviceCommunication(tStoreyInfo, fStoreyId);
});
try {
future.get(120, TimeUnit.SECONDS); // 2分钟超时
} catch (TimeoutException e) {
log.warn("设备通信超时: fStoreyId={}", fStoreyId);
future.cancel(true);
recordAlarm(tStoreyInfo,"设备通信超时（2分钟）");
} catch (Exception e) {
log.error("设备通信异常: fStoreyId={}", fStoreyId, e);
// 不抛出异常，只记录告警
recordAlarm(tStoreyInfo,"设备通信异常: " + e.getMessage());
}
}
private void cleanupInvalidJob(JobExecutionContext context, Long fStoreyId) {
try {
JobKey jobKey = context.getJobDetail().getKey();
context.getScheduler().deleteJob(jobKey);
log.info("清理无效任务: fStoreyId={}, jobKey={}", fStoreyId, jobKey);
} catch (SchedulerException e) {
log.error("清理无效任务失败: fStoreyId={}", fStoreyId, e);
}
}
private String getStoreyIdFromContext(JobExecutionContext context) {
try {
JobDataMap data = context.getJobDetail().getJobDataMap();
return data != null ? data.getString("fStoreyId") : "unknown";
} catch (Exception e) {
return "unknown";
}
}
private void executeDeviceCommunication(TStoreyInfo tStoreyInfo, Long fStoreyId) {
String ip = tStoreyInfo.getfIp();
// 使用并行流简化设备通信（按端口分组并行）
List<CompletableFuture<Void>> portFutures = Arrays.asList(
CompletableFuture.runAsync(() ->
executeSinglePort(ip, 501, Arrays.asList(1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27), fStoreyId, tStoreyInfo)
),
CompletableFuture.runAsync(() ->
executeSinglePort(ip, 502, Arrays.asList(28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54), fStoreyId, tStoreyInfo)
),
CompletableFuture.runAsync(() ->
executeSinglePort(ip, 503, Arrays.asList(55,56,57,58,59,60,61,62,63,64,65,66,67,68,69,70,71,72), fStoreyId, tStoreyInfo)
)
);
// 等待所有端口完成（带超时）
try {
CompletableFuture.allOf(portFutures.toArray(new CompletableFuture[0]))
.get(PORT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
} catch (TimeoutException e) {
log.warn("端口通信超时，取消未完成的任务");
portFutures.forEach(future -> future.cancel(true));
} catch (Exception e) {
log.error("端口通信异常", e);
}
}
private void executeSinglePort(String ip, int port, List<Integer> deviceIds, Long fStoreyId, TStoreyInfo tStoreyInfo) {
try {
Predicate<int[]> stopCondition = ModbusResultHandler.createDefaultStopCondition();
startMultiDeviceMonitoring(ip, port, deviceIds, resultHandler, stopCondition, PORT_TIMEOUT_SECONDS);
log.debug("端口{}通信完成: fStoreyId={}", port, fStoreyId);
} catch (Exception e) {
log.error("端口{}通信异常: fStoreyId={}", port, fStoreyId, e);
recordAlarm(tStoreyInfo,"端口" + port + "通信异常: " + e.getMessage());
}
// 顺序执行，避免并发问题
executeSinglePortSafely(ip, 501, getOffsets501(), fStoreyId, tStoreyInfo);
executeSinglePortSafely(ip, 502, getOffsets502(), fStoreyId, tStoreyInfo);
executeSinglePortSafely(ip, 503, getOffsets503(), fStoreyId, tStoreyInfo);
}
/**
* 启动多设备监控（核心方法：修正threadMaster调用）
* @param ip 设备IP
* @param port 端口（501/502/503）
* @param deviceIds 设备ID列表
* @param resultHandler 结果处理器
* @param stopCondition 停止条件
* @param portTimeout 端口总超时（秒）
*/
public void startMultiDeviceMonitoring(
String ip, int port, List<Integer> deviceIds,
Consumer<DeviceStatusReaderDto> resultHandler,
Predicate<int[]> stopCondition,
int portTimeout) {
final CountDownLatch latch = new CountDownLatch(deviceIds.size());
final AtomicInteger errorCount = new AtomicInteger(0);
private void executeSinglePortSafely(String ip, int port, List<Integer> deviceIds,
Long fStoreyId, TStoreyInfo tStoreyInfo) {
log.info("开始执行端口{}: ip={}, 设备数={}", port, ip, deviceIds.size());
for (int deviceId : deviceIds) {
final int devId = deviceId;
deviceExecutor.submit(() -> {
boolean success = false;
try {
// 执行设备通信
executeSingleDevice(ip, port, devId, resultHandler, stopCondition);
success = true;
} catch (Exception e) {
errorCount.incrementAndGet();
log.error("设备通信失败: ip={}, port={}, deviceId={}", ip, port, devId, e);
} finally {
latch.countDown();
log.debug("设备{}处理完成: 成功={}", devId, success);
}
});
}
try {
boolean completed = latch.await(portTimeout, TimeUnit.SECONDS);
if (!completed) {
log.warn("端口通信超时: ip={}, port={}, 完成数={}/{}",
ip, port, deviceIds.size() - latch.getCount(), deviceIds.size());
}
// 如果有设备通信失败，抛出异常让上层感知
if (errorCount.get() > 0) {
throw new RuntimeException(String.format("端口%d有%d个设备通信失败", port, errorCount.get()));
try {
executeSingleDeviceSafely(ip, port, deviceId, fStoreyId, tStoreyInfo);
// 每个设备执行后短暂休息，避免资源竞争
Thread.sleep(50);
} catch (Exception e) {
log.error("设备{}执行异常: ip={}, port={}", deviceId, ip, port, e);
// 继续执行下一个设备，不中断整个任务
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("端口通信被中断", e);
}
log.info("端口{}执行完成: ip={}", port, ip);
}
private void executeSingleDevice(String ip, int port, int deviceId,
Consumer<DeviceStatusReaderDto> resultHandler,
Predicate<int[]> stopCondition) {
private void executeSingleDeviceSafely(String ip, int port, int deviceId,
Long fStoreyId, TStoreyInfo tStoreyInfo) {
ModbusMaster master = null;
try {
// 创建连接
master = createModbusMaster(ip, port);
int[] result = readWithConditionalRetry(master, ip, port, deviceId, stopCondition);
// 读取数据（简化重试逻辑）
int[] result = readDeviceRegistersOnce(master, deviceId);
// 处理结果
if (resultHandler != null) {
resultHandler.accept(new DeviceStatusReaderDto(ip, port, deviceId, result));
}
log.debug("设备{}通信完成: ip={}, port={}", deviceId, ip, port);
} catch (Exception e) {
throw new RuntimeException("设备通信异常", e);
} finally {
destroyModbusMaster(master, deviceId);
}
}
/**
* 销毁Modbus连接（先关Socket，再销毁Master，彻底释放）
*/
private void destroyModbusMaster(ModbusMaster master, int deviceId) {
if (master == null) return;
log.debug("设备{}通信成功: ip={}, port={}", deviceId, ip, port);
try {
// 1. 直接获取并关闭TcpMaster的socket字段
Socket socket = getUnderlyingSocket(master);
if (socket != null && !socket.isClosed()) {
socket.close();
log.debug("设备{}: 底层Socket已强制关闭", deviceId);
}
} catch (Exception e) {
log.error("设备{}: 关闭Socket异常", deviceId, e);
log.warn("设备{}通信失败: ip={}, port={}, 错误: {}",
deviceId, ip, port, e.getMessage());
// 不抛出异常，只记录警告
} finally {
// 2. 调用自带的destroy()方法，双重保障
try {
master.destroy();
log.debug("设备{}: ModbusMaster已销毁", deviceId);
} catch (Exception e) {
log.error("设备{}: 销毁ModbusMaster异常", deviceId, e);
}
}
}
/**
* 修正：直接从TcpMaster获取socket字段（适配Modbus4j 3.0.3版本）
*/
private Socket getUnderlyingSocket(ModbusMaster master) {
if (!(master instanceof TcpMaster)) {
log.error("ModbusMaster不是TcpMaster类型，无法获取Socket");
return null;
}
TcpMaster tcpMaster = (TcpMaster) master;
try {
// 直接获取TcpMaster的private字段 "socket"（你的版本中直接存在该字段）
Field socketField = TcpMaster.class.getDeclaredField("socket");
socketField.setAccessible(true); // 取消私有字段访问限制
Socket socket = (Socket) socketField.get(tcpMaster);
if (socket != null) {
log.debug("成功获取Socket：IP={}, 端口={}",
socket.getInetAddress(), socket.getPort());
} else {
log.debug("TcpMaster的socket字段为null（未建立连接）");
}
return socket;
} catch (NoSuchFieldException e) {
log.error("=== 字段未找到！===\n" +
"原因：TcpMaster中不存在socket字段（版本不匹配）\n" +
"解决方案：检查你的TcpMaster源码，确认Socket字段名（当前版本应为'socket'）", e);
} catch (IllegalAccessException e) {
log.error("反射访问权限失败（可能被安全管理器拦截）", e);
} catch (Exception e) {
log.error("获取Socket异常", e);
}
return null;
}
/**
* 带自定义条件的重试读取（线程安全版）
*/
private int[] readWithConditionalRetry(ModbusMaster threadMaster,String ip, int port, int deviceId,
Predicate<int[]> conditionChecker)
throws InterruptedException {
TEquipmentAlarmData alarmData;
int[] result = null;
int attempt = 0;
boolean conditionMet = false;
if(threadMaster != null) {
while (attempt <= MAX_RETRIES && !conditionMet) {
attempt++;
// 确保资源释放
if (master != null) {
try {
log.info("当前 - 尝试第:{}次 读取设备ip:{} port:{} device:{}的数据", attempt, ip,port,deviceId);
// 读取数据
result = readDeviceRegisters(threadMaster, deviceId);
// 使用自定义条件检查
if (conditionChecker.test(result)) {
log.info("当前 设备ip:{} port:{} device:{}的数据条件满足", ip,port,deviceId);
conditionMet = true;
} else if (attempt <= MAX_RETRIES) {
log.info("当前 设备ip:{} port:{} device:{}的数据条件未满足，等待重试...", ip,port,deviceId);
Thread.sleep(RETRY_DELAY);
}
} catch (Throwable e) {
log.error("设备{}: 第{}次读取异常", deviceId, attempt, e);
if (attempt <= MAX_RETRIES) {
Thread.sleep(RETRY_DELAY);
}
master.destroy();
} catch (Exception e) {
log.debug("释放Modbus连接异常: deviceId={}", deviceId, e);
}
}
// 如果达到最大重试次数仍未满足条件
if (!conditionMet) {
log.info("当前 设备ip:{} port:{} device:{}的尝试次数达到最大:{}", ip,port,deviceId,MAX_RETRIES);
}
}
return result != null ? result : new int[REGISTER_COUNT];
}
/**
* 读取设备寄存器（线程安全版）
* 简化版：只读取一次，不重试
*/
private int[] readDeviceRegisters(ModbusMaster master, int deviceId )
private int[] readDeviceRegistersOnce(ModbusMaster master, int deviceId)
throws ModbusTransportException {
ReadHoldingRegistersRequest request =
Modbus4jUtils.getReadHoldingRegistersRequest(deviceId, START_ADDRESS, REGISTER_COUNT);
// 创建读取请求
ReadHoldingRegistersRequest request = Modbus4jUtils.getReadHoldingRegistersRequest(deviceId,START_ADDRESS, REGISTER_COUNT);
// 发送请求并获取响应
ModbusResponse response = master.send(request);
// 检查响应类型
if (!(response instanceof ReadHoldingRegistersResponse)) {
throw new IllegalArgumentException("Invalid response type: " + response.getClass().getName());
throw new IllegalArgumentException("Invalid response type");
}
ReadHoldingRegistersResponse regResponse = (ReadHoldingRegistersResponse) response;
......@@ -412,6 +190,35 @@ public class DeviceCommunicationJob implements Job {
return unsignedValues;
}
private void recordAlarm(JobExecutionContext context, String alarmData) {
try {
JobDataMap data = context.getJobDetail().getJobDataMap();
String fStoreyIdStr = data.getString("fStoreyId");
TEquipmentAlarmData alarm = new TEquipmentAlarmData();
alarm.setfAlarmType("03");
alarm.setfEquipmentCode(fStoreyIdStr != null ? fStoreyIdStr : "unknown");
alarm.setfAlarmData(alarmData);
alarm.setfCreateTime(new Date());
alarmDataService.insertTEquipmentAlarmData(alarm);
} catch (Exception e) {
log.error("记录告警失败", e);
}
}
// 设备列表
private List<Integer> getOffsets501() {
return Arrays.asList(1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27);
}
private List<Integer> getOffsets502() {
return Arrays.asList(28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54);
}
private List<Integer> getOffsets503() {
return Arrays.asList(55,56,57,58,59,60,61,62,63,64,65,66,67,68,69,70,71,72);
}
/**
* 创建新的Modbus连接
*/
......@@ -428,32 +235,4 @@ public class DeviceCommunicationJob implements Job {
log.debug("Modbus连接创建成功：IP={}, 端口={}, Master={}", ip, port, master);
return master;
}
private void recordTimeoutAlarm(JobExecutionContext context) {
try {
JobDataMap data = context.getJobDetail().getJobDataMap();
String fStoreyIdStr = data.getString("fStoreyId");
TEquipmentAlarmData alarm = new TEquipmentAlarmData();
alarm.setfAlarmType("03");
alarm.setfEquipmentCode(fStoreyIdStr != null ? fStoreyIdStr : "unknown");
alarm.setfAlarmData("设备通信任务执行超时（>" + TOTAL_TIMEOUT_SECONDS + "秒）");
alarm.setfCreateTime(new Date());
alarmDataService.insertTEquipmentAlarmData(alarm);
} catch (Exception e) {
log.error("记录超时告警失败", e);
}
}
private void recordAlarm(TStoreyInfo tStoreyInfo, String alarmData) {
try {
TEquipmentAlarmData alarm = new TEquipmentAlarmData();
alarm.setfAlarmType("03");
alarm.setfEquipmentCode(tStoreyInfo != null ? tStoreyInfo.getfStoreyCode() : "unknown");
alarm.setfAlarmData(alarmData);
alarm.setfCreateTime(new Date());
alarmDataService.insertTEquipmentAlarmData(alarm);
} catch (Exception e) {
log.error("记录告警失败", e);
}
}
}
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