Skip to content

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

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

parent 59696230
Show whitespace changes
Inline Side-by-side
Showing with 111 additions and 28 deletions
zhmes-agecal-system/src/main/java/com/zehong/system/modbus/business/DeviceStatusReaderAndTimeSetter.java
View file @ 34db516e
......@@ -217,33 +217,41 @@ public class DeviceStatusReaderAndTimeSetter {
Consumer<DeviceStatusReaderDto> resultHandler,
Predicate<int[]> stopCondition,
int portTimeout) {
if (deviceIds == null || deviceIds.isEmpty()) return;
CountDownLatch latch = new CountDownLatch(deviceIds.size());
AtomicInteger successCount = new AtomicInteger(0);
final CountDownLatch latch = new CountDownLatch(deviceIds.size());
final AtomicInteger errorCount = new AtomicInteger(0);
for (int deviceId : deviceIds) {
final int devId = deviceId;
deviceExecutor.submit(() -> {
boolean success = false;
try {
// 简化设备通信逻辑
executeSingleDevice(ip, port, deviceId, resultHandler, stopCondition);
successCount.incrementAndGet();
// 执行设备通信
executeSingleDevice(ip, port, devId, resultHandler, stopCondition);
success = true;
} catch (Exception e) {
log.error("设备通信失败: ip={}, port={}, deviceId={}", ip, port, deviceId, e);
errorCount.incrementAndGet();
log.error("设备通信失败: ip={}, port={}, deviceId={}", ip, port, devId, e);
} finally {
latch.countDown();
log.debug("设备{}处理完成: 成功={}", devId, success);
}
});
}
try {
// 等待所有设备完成或超时
if (!latch.await(portTimeout, TimeUnit.SECONDS)) {
boolean completed = latch.await(portTimeout, TimeUnit.SECONDS);
if (!completed) {
log.warn("端口通信超时: ip={}, port={}, 完成数={}/{}",
ip, port, successCount.get(), deviceIds.size());
ip, port, deviceIds.size() - latch.getCount(), deviceIds.size());
}
// 如果有设备通信失败，抛出异常让上层感知
if (errorCount.get() > 0) {
throw new RuntimeException(String.format("端口%d有%d个设备通信失败", port, errorCount.get()));
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("端口通信被中断", e);
}
}
private void executeSingleDevice(String ip, int port, int deviceId,
......
zhmes-agecal-system/src/main/java/com/zehong/system/task/DeviceCommunicationJob.java
View file @ 34db516e
......@@ -17,7 +17,9 @@ import javax.annotation.Resource;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Predicate;
/**
......@@ -46,29 +48,102 @@ public class DeviceCommunicationJob implements Job {
@Override
public void execute(JobExecutionContext context) {
CompletableFuture<Void> taskFuture = null;
// 关键：确保任何异常都不会传播到Quartz框架
try {
// 1. 包装整个任务在CompletableFuture中，便于超时控制
taskFuture = CompletableFuture.runAsync(() -> {
executeInternal(context);
});
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());
}
}
}
private void executeSafely(JobExecutionContext context) {
long startTime = System.currentTimeMillis();
String fStoreyIdStr = null;
TStoreyInfo tStoreyInfo = null;
try {
// 参数提取和验证
JobDataMap data = context.getJobDetail().getJobDataMap();
fStoreyIdStr = data.getString("fStoreyId");
// 2. 设置总超时，避免任务无限期执行
taskFuture.get(TOTAL_TIMEOUT_SECONDS, TimeUnit.SECONDS);
if (StringUtils.isBlank(fStoreyIdStr)) {
log.warn("fStoreyId参数为空，跳过执行");
return;
}
} catch (TimeoutException e) {
log.error("任务执行超时，强制取消");
if (taskFuture != null) {
taskFuture.cancel(true);
Long fStoreyId = Long.parseLong(fStoreyIdStr);
tStoreyInfo = tStoreyInfoMapper.selectTStoreyInfoById(fStoreyId);
if (tStoreyInfo == null) {
log.warn("设备信息不存在: fStoreyId={}", fStoreyId);
// 清理无效任务
cleanupInvalidJob(context, fStoreyId);
return;
}
// 记录超时告警，但不抛出异常到Quartz
recordTimeoutAlarm(context);
if (StringUtils.isBlank(tStoreyInfo.getfIp())) {
log.warn("设备IP为空: fStoreyId={}", fStoreyId);
recordAlarm(tStoreyInfo,"设备IP为空");
return;
}
// 执行设备通信（带超时控制）
executeDeviceCommunicationWithTimeout(tStoreyInfo, fStoreyId);
long costTime = System.currentTimeMillis() - startTime;
log.info("设备通信任务成功完成: fStoreyId={}, 耗时={}ms", fStoreyId, costTime);
} 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());
// 注意：这里不再抛出异常！
}
}
private void executeDeviceCommunicationWithTimeout(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("任务执行异常", e);
// 记录异常，但不传播到Quartz
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 executeInternal(JobExecutionContext context) {
long startTime = System.currentTimeMillis();
String fStoreyIdStr = null;
......@@ -140,7 +215,7 @@ public class DeviceCommunicationJob implements Job {
log.debug("端口{}通信完成: fStoreyId={}", port, fStoreyId);
} catch (Exception e) {
log.error("端口{}通信异常: fStoreyId={}", port, fStoreyId, e);
recordAlarm(tStoreyInfo, "端口" + port + "通信异常: " + e.getMessage());
recordAlarm(tStoreyInfo,"端口" + port + "通信异常: " + e.getMessage());
}
}
......
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