0214
/
ALOps_sys1
2
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

11.10采集规则改变

Browse Source
dev
xiaohuo 8 months ago
parent 4aedf1dc27
commit 9a4ba243d7
2 changed files with 60 additions and 29 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 2
      ALOps_sys_backend/alops-system/src/main/java/com/alops/system/job/GatherJob.java
  2. 87
      ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/ScheduleRulesServiceImpl.java

2
ALOps_sys_backend/alops-system/src/main/java/com/alops/system/job/GatherJob.java
Unescape View File

@ -124,7 +124,7 @@ public class GatherJob implements Job {
rule.setStatus(3);
scheduleRulesMapper.updateScheduleRules(rule);
try {
String Content=String.format("采集任务整体执行失败，规则ID=%s，规则名称=%s，原因：%s",
String Content=String.format("采集任务整体执行失败，规则ID=%s，规则名称=%s，原因：%s，已取消任务",
rule.getId(), rule.getRuleName(), e.getMessage());
iAlertSenderService.sendInternalMessage("采集任务整体执行失败" ,Content,"2" );

87
ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/ScheduleRulesServiceImpl.java
Unescape View File

@ -205,7 +205,7 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService {
oidMap.put(dev.getEquManuId(), equOidMapper.findByManuIdAndStatus(dev.getEquManuId()));
}
// 并发采集
ExecutorService executor = Executors.newFixedThreadPool(10);
ExecutorService executor = Executors.newFixedThreadPool(12);
List<Future<GatherEntities>> futures = new ArrayList<>();
for (EquBase device : devices) {
futures.add(executor.submit(() -> {
@ -271,12 +271,12 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService {
gatherValidator.validateBeforeDbOperation(successList, failed, oidMap);
List<String> failedEquIds = validationResult.getFailed().stream()
.map(f -> f.get("equId"))
.collect(Collectors.toList());
// List<String> failedEquIds = validationResult.getFailed().stream()
// .map(f -> f.get("equId"))
// .collect(Collectors.toList());
// 事务内：删除成功设备依附表 + 插入成功设备数据 + 执行预警
deleteAllEquipmentData(failedEquIds); // 删除依附表，不含 equ_gather
deleteAllEquipmentData(validationResult.getSuccessList()); // 删除依附表，不含 equ_gather
// 插入成功设备数据（内部失败会抛异常，触发事务回滚）
persistAndUpdate(validationResult.getSuccessList());
@ -287,33 +287,64 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService {
return Code.SUCCESS;
}
@Transactional(rollbackFor = Exception.class)
public void deleteAllEquipmentData(List<String> failedEquIds) {
// 1️⃣ 查询失败设备最新采集ID
List<String> excludeGatherIds = failedEquIds == null || failedEquIds.isEmpty() ?
Collections.emptyList() : equGatherMapper.selectLatestGatherIdsByEquIds(failedEquIds);
// 2️⃣ 依附表列表
String[] dependentTables = {"equ_interface", "equ_routing", "equ_veneer", "equ_vlan"};
for (String table : dependentTables) {
int deleted;
if (excludeGatherIds.isEmpty()) {
// 没有失败设备，直接清空表
deleted = jdbcTemplate.update("DELETE FROM " + table);
} else {
// 使用占位符方式安全删除
String inSql = excludeGatherIds.stream().map(id -> "?").collect(Collectors.joining(","));
String sql = String.format("DELETE FROM %s WHERE equ_gather_id NOT IN (%s)", table, inSql);
deleted = jdbcTemplate.update(sql, excludeGatherIds.toArray());
}
log.info("表 {} 删除 {} 条数据，保留失败设备采集ID: {}", table, deleted, excludeGatherIds);
}
// @Transactional(rollbackFor = Exception.class)
// public void deleteAllEquipmentData(List<String> failedEquIds) {
// // 1️⃣ 查询失败设备最新采集ID
// List<String> excludeGatherIds = failedEquIds == null || failedEquIds.isEmpty() ?
// Collections.emptyList() : equGatherMapper.selectLatestGatherIdsByEquIds(failedEquIds);
//
// // 2️⃣ 依附表列表
// String[] dependentTables = {"equ_interface", "equ_routing", "equ_veneer", "equ_vlan"};
//
// for (String table : dependentTables) {
// int deleted;
// if (excludeGatherIds.isEmpty()) {
// // 没有失败设备，直接清空表
// deleted = jdbcTemplate.update("DELETE FROM " + table);
// } else {
// // 使用占位符方式安全删除
// String inSql = excludeGatherIds.stream().map(id -> "?").collect(Collectors.joining(","));
// String sql = String.format("DELETE FROM %s WHERE equ_gather_id NOT IN (%s)", table, inSql);
// deleted = jdbcTemplate.update(sql, excludeGatherIds.toArray());
// }
// log.info("表 {} 删除 {} 条数据，保留失败设备采集ID: {}", table, deleted, excludeGatherIds);
// }
//
// log.info("依附表数据删除完成，共保留 {} 个失败设备最新采集ID", excludeGatherIds.size());
// }
@Transactional(rollbackFor = Exception.class)
public void deleteAllEquipmentData(List<GatherEntities> successList) {
// 1️⃣ 提取成功设备ID列表
List<String> successEquIds = successList.stream()
.map(g -> g.getEquGather().getEquId())
.filter(Objects::nonNull)
.distinct()
.collect(Collectors.toList());
// 2️⃣ 查询这些设备的最新采集ID（要保留的）
List<String> latestGatherIds = equGatherMapper.selectLatestGatherIdsByEquIds(successEquIds);
if (latestGatherIds == null || latestGatherIds.isEmpty()) {
log.warn("未查询到成功设备的最新采集ID，跳过删除操作");
return;
}
log.info("依附表数据删除完成，共保留 {} 个失败设备最新采集ID", excludeGatherIds.size());
// 3️⃣ 依附表列表
String[] dependentTables = {"equ_interface", "equ_routing", "equ_veneer", "equ_vlan"};
// 4️⃣ 拼 SQL: 删除所有 成功设备
String placeholders = latestGatherIds.stream().map(id -> "?").collect(Collectors.joining(","));
for (String table : dependentTables) {
String sql = String.format("DELETE FROM %s WHERE equ_gather_id IN (%s)", table, placeholders);
int deleted = jdbcTemplate.update(sql, latestGatherIds.toArray());
log.info("表 {} 删除旧数据 {} 条）", table, deleted);
}
log.info("✅ 成功清理依附表数据，");
}
/**
* 批量落库、更新采集次数、执行最新预警
* 事务保证，如果任何步骤失败，整个操作回滚

Write Preview
Loading…
Cancel
Save