|
|
|
|
@ -21,6 +21,7 @@ import org.quartz.SchedulerException; |
|
|
|
|
import org.slf4j.Logger; |
|
|
|
|
import org.slf4j.LoggerFactory; |
|
|
|
|
import org.springframework.beans.factory.annotation.Autowired; |
|
|
|
|
import org.springframework.jdbc.core.JdbcTemplate; |
|
|
|
|
import org.springframework.stereotype.Service; |
|
|
|
|
import org.springframework.transaction.annotation.Propagation; |
|
|
|
|
import org.springframework.transaction.annotation.Transactional; |
|
|
|
|
@ -50,6 +51,15 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService { |
|
|
|
|
@Autowired |
|
|
|
|
private IAlertLaunchService iAlertLaunchService; |
|
|
|
|
|
|
|
|
|
@Autowired |
|
|
|
|
private EquInterfaceMapper equInterfaceMapper; |
|
|
|
|
@Autowired |
|
|
|
|
private EquRoutingMapper equRoutingMapper; |
|
|
|
|
@Autowired |
|
|
|
|
private EquVeneerMapper equVeneerMapper; |
|
|
|
|
@Autowired |
|
|
|
|
private EquVlanMapper equVlanMapper; |
|
|
|
|
|
|
|
|
|
@Autowired |
|
|
|
|
private IQuartzService quartzService; |
|
|
|
|
|
|
|
|
|
@ -57,9 +67,16 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService { |
|
|
|
|
private MemoryToolService memoryToolService; |
|
|
|
|
@Autowired |
|
|
|
|
private IAlertSenderService iAlertSenderService; |
|
|
|
|
@Autowired |
|
|
|
|
public GatherValidator gatherValidator; |
|
|
|
|
|
|
|
|
|
private static final Logger log = LoggerFactory.getLogger(ScheduleRulesServiceImpl.class); |
|
|
|
|
private final JdbcTemplate jdbcTemplate; |
|
|
|
|
|
|
|
|
|
// 构造器注入
|
|
|
|
|
public ScheduleRulesServiceImpl(JdbcTemplate jdbcTemplate) { |
|
|
|
|
this.jdbcTemplate = jdbcTemplate; |
|
|
|
|
} |
|
|
|
|
private static final Logger log = LoggerFactory.getLogger(ScheduleRulesServiceImpl.class); |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@Override |
|
|
|
|
@ -99,34 +116,7 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService { |
|
|
|
|
public Boolean executeRule(ScheduleRules newRule) throws SchedulerException, ExecutionException, InterruptedException { |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// // 1. 获取所有正在执行(1)和待执行(0)的规则
|
|
|
|
|
// List<ScheduleRules> runningOrWaiting = scheduleRulesMapper.selectList();
|
|
|
|
|
// if (runningOrWaiting!=null) {
|
|
|
|
|
// for (ScheduleRules rule : runningOrWaiting) {
|
|
|
|
|
// switch (rule.getStatus()) {
|
|
|
|
|
// case 1: // 正在执行
|
|
|
|
|
// log.info("规则[{}]：正在执行上一个任务，等待上一个任务完成……", rule.getId());
|
|
|
|
|
// waitTaskFinish(rule.getId());
|
|
|
|
|
// log.info("规则[{}]：完成上一个任务任务，开始新的任务…", rule.getId());
|
|
|
|
|
// if (rule.getRuleType() == 2 || rule.getRuleType() == 1) {
|
|
|
|
|
// rule.setStatus(2);
|
|
|
|
|
// } else if (rule.getRuleType() == 0) {
|
|
|
|
|
// rule.setStatus(4);
|
|
|
|
|
// }
|
|
|
|
|
// scheduleRulesMapper.updateScheduleRules(rule);
|
|
|
|
|
// break;
|
|
|
|
|
// case 0: // 待执行
|
|
|
|
|
// if (rule.getRuleType() == 0 || rule.getRuleType() == 1) {
|
|
|
|
|
// quartzService.deleteJob(rule.getId());
|
|
|
|
|
// }
|
|
|
|
|
// rule.setStatus(4);
|
|
|
|
|
// scheduleRulesMapper.updateScheduleRules(rule);
|
|
|
|
|
// break;
|
|
|
|
|
// default:
|
|
|
|
|
// break;
|
|
|
|
|
// }
|
|
|
|
|
// }
|
|
|
|
|
// }
|
|
|
|
|
|
|
|
|
|
// 2. 执行新规则
|
|
|
|
|
try { |
|
|
|
|
Date now = new Date(); |
|
|
|
|
@ -151,7 +141,7 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService { |
|
|
|
|
scheduleRulesMapper.updateScheduleRules(newRule); |
|
|
|
|
|
|
|
|
|
String content = String.format("立即任务执行完成", newRule.getRuleName()); |
|
|
|
|
iAlertSenderService.sendInternalMessage("采集任务成功",content, "2"); |
|
|
|
|
iAlertSenderService.sendInternalMessage("采集任务整体执行成功",content, "2"); |
|
|
|
|
|
|
|
|
|
} catch (Exception e) { |
|
|
|
|
newRule.setStatus(3); // 执行失败
|
|
|
|
|
@ -159,7 +149,7 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService { |
|
|
|
|
|
|
|
|
|
String content = String.format("立即任务执行失败：%s", e.getMessage()); |
|
|
|
|
try { |
|
|
|
|
iAlertSenderService.sendInternalMessage("采集任务失败",content, "2"); |
|
|
|
|
iAlertSenderService.sendInternalMessage("采集任务整体执行失败",content, "2"); |
|
|
|
|
} catch (Exception ne) { |
|
|
|
|
log.error("发送站内信失败: {}", ne.getMessage(), ne); |
|
|
|
|
} |
|
|
|
|
@ -192,96 +182,8 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 删除所有设备相关表数据，保证事务一致性 |
|
|
|
|
*/ |
|
|
|
|
@Transactional(rollbackFor = Exception.class) |
|
|
|
|
public void deleteAllEquipmentData() { |
|
|
|
|
equBaseMapper.deleteEquInterface(); |
|
|
|
|
equBaseMapper.deleteEquRouting(); |
|
|
|
|
equBaseMapper.deleteEquVeneer(); |
|
|
|
|
equBaseMapper.deleteEquVlan(); |
|
|
|
|
} |
|
|
|
|
//往上抛异常
|
|
|
|
|
// @Override
|
|
|
|
|
// public int executeImmediately() throws ExecutionException, InterruptedException {
|
|
|
|
|
// // 删除之前的信息
|
|
|
|
|
// deleteAllEquipmentData();
|
|
|
|
|
//
|
|
|
|
|
// List<EquBase> devices = equBaseMapper.findEquBaseList();
|
|
|
|
|
// if (devices.isEmpty()) return Code.FAIL;
|
|
|
|
|
//
|
|
|
|
|
// Map<String, List<EquOid>> oidMap = new HashMap<>();
|
|
|
|
|
// for (EquBase dev : devices) {
|
|
|
|
|
// oidMap.put(dev.getEquManuId(), equOidMapper.findByManuIdAndStatus(dev.getEquManuId()));
|
|
|
|
|
// }
|
|
|
|
|
//
|
|
|
|
|
// ExecutorService executor = Executors.newFixedThreadPool(10);
|
|
|
|
|
// List<Future<GatherEntities>> futures = new ArrayList<>();
|
|
|
|
|
//
|
|
|
|
|
// for (EquBase device : devices) {
|
|
|
|
|
// futures.add(executor.submit(() -> {
|
|
|
|
|
// long start = System.currentTimeMillis();
|
|
|
|
|
//
|
|
|
|
|
// //获取对应oid
|
|
|
|
|
// List<EquOid> oids = oidMap.getOrDefault(device.getEquManuId(), Collections.emptyList());
|
|
|
|
|
// if (oids.isEmpty()) {
|
|
|
|
|
// System.err.println("⚠️ 设备[" + device.getName() + "]未配置OID，跳过");
|
|
|
|
|
// return null;
|
|
|
|
|
// }
|
|
|
|
|
// //采集对应结果
|
|
|
|
|
// Map<String, Object> snmpResults = snmpService.query(device.getIp(), oids);
|
|
|
|
|
// if (snmpResults == null || snmpResults.isEmpty()) {
|
|
|
|
|
// System.err.println("⚠️ 设备[" + device.getName() + "] SNMP采集为空，跳过");
|
|
|
|
|
// return null;
|
|
|
|
|
// }
|
|
|
|
|
//
|
|
|
|
|
// long afterSnmp = System.currentTimeMillis();
|
|
|
|
|
// System.out.println("SNMP查询耗时: " + (afterSnmp - start) + " ms");
|
|
|
|
|
// //生成对应实体
|
|
|
|
|
// GatherEntities entities = memoryToolService.buildAndInsertEntity(device, snmpResults);
|
|
|
|
|
//
|
|
|
|
|
// if (entities == null || entities.getEquGather() == null) {
|
|
|
|
|
// System.err.println("⚠️ 设备[" + device.getName() + "] 构建实体失败，跳过");
|
|
|
|
|
// return null;
|
|
|
|
|
// }
|
|
|
|
|
//
|
|
|
|
|
// long afterBuild = System.currentTimeMillis();
|
|
|
|
|
// System.out.println("buildAndInsertEntity耗时: " + (afterBuild - afterSnmp) + " ms");
|
|
|
|
|
//
|
|
|
|
|
// //构建健康度
|
|
|
|
|
// if (entities != null && entities.getEquGather() != null) {
|
|
|
|
|
// Double temperature = parseDouble(snmpResults.get("temperature"));
|
|
|
|
|
// Double cpuUsage = parseDouble(snmpResults.get("cpu_utilization"));
|
|
|
|
|
// Double memUsage = parseDouble(snmpResults.get("memory_utilization"));
|
|
|
|
|
// double tempHealth = calculateTemperatureHealth(temperature);
|
|
|
|
|
// double cpuHealth = calculateCpuHealth(cpuUsage);
|
|
|
|
|
// double memHealth = calculateMemoryHealth(memUsage);
|
|
|
|
|
// double totalHealth = Math.max(Math.max(tempHealth, 1), Math.max(cpuHealth, memHealth));
|
|
|
|
|
// entities.getEquGather().setHealth(BigDecimal.valueOf(totalHealth));
|
|
|
|
|
// return entities;
|
|
|
|
|
// }
|
|
|
|
|
// return null;
|
|
|
|
|
// }));
|
|
|
|
|
// }
|
|
|
|
|
// //收集采集结果，批量存入数据库
|
|
|
|
|
// List<GatherEntities> allEntities = new ArrayList<>();
|
|
|
|
|
// try {
|
|
|
|
|
// for (Future<GatherEntities> f : futures) {
|
|
|
|
|
// GatherEntities g = f.get(); // 阻塞等待线程完成，如果线程异常会抛ExecutionException
|
|
|
|
|
// if (g == null) throw new RuntimeException("设备采集返回 null，可能采集失败");
|
|
|
|
|
// allEntities.add(g);
|
|
|
|
|
// }
|
|
|
|
|
// } catch (ExecutionException | InterruptedException e) {
|
|
|
|
|
// e.printStackTrace();
|
|
|
|
|
// executor.shutdownNow(); // 出异常直接中断剩余线程
|
|
|
|
|
// throw e; // 抛给调用方 Job 捕获
|
|
|
|
|
// } finally {
|
|
|
|
|
// executor.shutdown();
|
|
|
|
|
// }
|
|
|
|
|
// return persistAndUpdate(allEntities);
|
|
|
|
|
//
|
|
|
|
|
// }
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@Transactional(rollbackFor = Exception.class) |
|
|
|
|
@Override |
|
|
|
|
@ -293,12 +195,11 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService { |
|
|
|
|
return Code.FAIL; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 构建设备 OID 映射
|
|
|
|
|
// 构建设备 OID 映射（插入）
|
|
|
|
|
Map<String, List<EquOid>> oidMap = new HashMap<>(); |
|
|
|
|
for (EquBase dev : devices) { |
|
|
|
|
oidMap.put(dev.getEquManuId(), equOidMapper.findByManuIdAndStatus(dev.getEquManuId())); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 并发采集
|
|
|
|
|
ExecutorService executor = Executors.newFixedThreadPool(10); |
|
|
|
|
List<Future<GatherEntities>> futures = new ArrayList<>(); |
|
|
|
|
@ -345,130 +246,189 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService { |
|
|
|
|
|
|
|
|
|
// 收集采集结果
|
|
|
|
|
List<GatherEntities> successList = new ArrayList<>(); |
|
|
|
|
List<String> failed = new ArrayList<>(); |
|
|
|
|
List<Map<String, String>> failed = new ArrayList<>(); // 包含设备ID和设备名称
|
|
|
|
|
|
|
|
|
|
for (int i = 0; i < futures.size(); i++) { |
|
|
|
|
EquBase dev = devices.get(i); |
|
|
|
|
try { |
|
|
|
|
GatherEntities g = futures.get(i).get(); |
|
|
|
|
if (g != null) successList.add(g); |
|
|
|
|
else failed.add(dev.getName()); |
|
|
|
|
else failed.add(Map.of("equId", dev.getId(), "equName", dev.getName())); |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
failed.add(dev.getName()); |
|
|
|
|
failed.add(Map.of("equId", dev.getId(), "equName", dev.getName())); |
|
|
|
|
log.error("收集设备 [{}] 结果失败", dev.getName(), e); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
//关闭线程池
|
|
|
|
|
executor.shutdown(); |
|
|
|
|
|
|
|
|
|
// 过滤 successList 中不完整的设备，记录失败设备，equ_gather和equ_base只要有数据就行
|
|
|
|
|
GatherValidator.ValidationResult validationResult = |
|
|
|
|
gatherValidator.validateBeforeDbOperation(successList, failed, oidMap); |
|
|
|
|
|
|
|
|
|
// // ✅ 完整性校验：找出所有子表为空的设备
|
|
|
|
|
// List<String> invalidDevices = successList.stream()
|
|
|
|
|
// .filter(g -> g.getSubEntities() == null
|
|
|
|
|
// || g.getSubEntities().isEmpty()
|
|
|
|
|
// || g.getSubEntities().values().stream().anyMatch(list -> list == null || list.isEmpty()))
|
|
|
|
|
// .map(g -> g.getEquGather().getEquId())
|
|
|
|
|
// .collect(Collectors.toList());
|
|
|
|
|
//
|
|
|
|
|
// // 如果全部失败或者发现不完整设备，抛异常
|
|
|
|
|
// if (successList.isEmpty() || !invalidDevices.isEmpty()) {
|
|
|
|
|
// String msg = "本次采集失败，保持旧数据完好。失败设备：" +
|
|
|
|
|
// String.join(", ", failed) +
|
|
|
|
|
// (invalidDevices.isEmpty() ? "" : "，采集数据不完整设备：" + String.join(", ", invalidDevices));
|
|
|
|
|
// log.error(msg);
|
|
|
|
|
// throw new RuntimeException(msg); // 异常抛出去，外层可捕获发送邮件
|
|
|
|
|
// }
|
|
|
|
|
// ✅ 完整性校验：找出所有子表为空的设备及对应子表
|
|
|
|
|
Map<String, List<String>> invalidDeviceMap = successList.stream() |
|
|
|
|
.map(g -> { |
|
|
|
|
String equId = g.getEquGather().getEquId(); |
|
|
|
|
|
|
|
|
|
// 找出这个设备下不完整的子表
|
|
|
|
|
List<String> emptyTables = new ArrayList<>(); |
|
|
|
|
|
|
|
|
|
// subEntities 为空或没有数据
|
|
|
|
|
if (g.getSubEntities() == null || g.getSubEntities().isEmpty()) { |
|
|
|
|
emptyTables.add("所有子表为空"); |
|
|
|
|
} else { |
|
|
|
|
// 检查每个子表的值
|
|
|
|
|
g.getSubEntities().forEach((tableName, list) -> { |
|
|
|
|
if (list == null || list.isEmpty()) { |
|
|
|
|
emptyTables.add(tableName); |
|
|
|
|
} |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
List<String> failedEquIds = validationResult.getFailed().stream() |
|
|
|
|
.map(f -> f.get("equId")) |
|
|
|
|
.collect(Collectors.toList()); |
|
|
|
|
|
|
|
|
|
// 如果有任何空子表，返回 (设备ID, 空表列表)
|
|
|
|
|
return emptyTables.isEmpty() ? null : Map.entry(equId, emptyTables); |
|
|
|
|
}) |
|
|
|
|
.filter(Objects::nonNull) |
|
|
|
|
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); |
|
|
|
|
// 事务内：删除成功设备依附表 + 插入成功设备数据 + 执行预警
|
|
|
|
|
deleteAllEquipmentData(failedEquIds); // 删除依附表，不含 equ_gather
|
|
|
|
|
|
|
|
|
|
// ✅ 提取不完整设备ID列表
|
|
|
|
|
List<String> invalidDevices = new ArrayList<>(invalidDeviceMap.keySet()); |
|
|
|
|
// 插入成功设备数据（内部失败会抛异常，触发事务回滚）
|
|
|
|
|
persistAndUpdate(successList); |
|
|
|
|
|
|
|
|
|
// ✅ 如果全部失败或有不完整设备，抛异常
|
|
|
|
|
if (successList.isEmpty() || !invalidDevices.isEmpty()) { |
|
|
|
|
StringBuilder msg = new StringBuilder("本次采集失败，保持旧数据完好。"); |
|
|
|
|
// 发送成功日志
|
|
|
|
|
log.info("本次采集成功 {} 台，删除旧表并写入最新数据，失败设备：{}", successList.size(), failed); |
|
|
|
|
|
|
|
|
|
if (!failed.isEmpty()) { |
|
|
|
|
msg.append("失败设备：").append(String.join(", ", failed)); |
|
|
|
|
return Code.SUCCESS; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if (!invalidDeviceMap.isEmpty()) { |
|
|
|
|
msg.append("；采集数据不完整设备如下："); |
|
|
|
|
invalidDeviceMap.forEach((equId, tables) -> |
|
|
|
|
msg.append("\n设备[").append(equId).append("] 缺少表：") |
|
|
|
|
.append(String.join(", ", tables))); |
|
|
|
|
@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.error(msg.toString()); |
|
|
|
|
throw new RuntimeException(msg.toString()); |
|
|
|
|
log.info("表 {} 删除 {} 条数据，保留失败设备采集ID: {}", table, deleted, excludeGatherIds); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 事务内：删除旧表 + 插入成功设备数据+执行预警
|
|
|
|
|
deleteAllEquipmentData(); // 删除依附表，不含 equ_gather
|
|
|
|
|
|
|
|
|
|
int inserted = persistAndUpdate(successList); |
|
|
|
|
|
|
|
|
|
if (inserted <= 0) { |
|
|
|
|
throw new RuntimeException("成功采集设备写入失败，回滚事务"); |
|
|
|
|
} |
|
|
|
|
//发送一个采集成功的信
|
|
|
|
|
log.info("本次采集成功 {} 台，删除旧表并写入最新数据，失败设备：{}", successList.size(), failed); |
|
|
|
|
return Code.SUCCESS; |
|
|
|
|
log.info("依附表数据删除完成，共保留 {} 个失败设备最新采集ID", excludeGatherIds.size()); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 批量落库、更新采集次数、执行最新预警 |
|
|
|
|
* 事务保证，如果任何步骤失败，整个操作回滚 |
|
|
|
|
*/ |
|
|
|
|
@Transactional(rollbackFor = Exception.class) |
|
|
|
|
public int persistAndUpdate(List<GatherEntities> allEntities) throws ExecutionException, InterruptedException { |
|
|
|
|
public int persistAndUpdate(List<GatherEntities> allEntities) { |
|
|
|
|
if (allEntities == null || allEntities.isEmpty()) { |
|
|
|
|
return 0; |
|
|
|
|
throw new RuntimeException("待插入实体为空，无法落库"); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
try { |
|
|
|
|
// ===== 1. 主表批量插入 =====
|
|
|
|
|
List<EquGather> gatherList = allEntities.stream() |
|
|
|
|
.map(GatherEntities::getEquGather) |
|
|
|
|
.filter(Objects::nonNull) |
|
|
|
|
.collect(Collectors.toList()); |
|
|
|
|
|
|
|
|
|
if (!gatherList.isEmpty()) { |
|
|
|
|
equGatherMapper.batchInsertEquGather(gatherList); |
|
|
|
|
log.info("主表批量插入完成，数量：{}", gatherList.size()); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 批量落库
|
|
|
|
|
long persistStart = System.currentTimeMillis(); |
|
|
|
|
int result = memoryToolService.persistAll(allEntities); |
|
|
|
|
long persistEnd = System.currentTimeMillis(); |
|
|
|
|
System.out.println("批量落库耗时: " + (persistEnd - persistStart) + " ms"); |
|
|
|
|
// ===== 2. 子表按表名分组批量插入 =====
|
|
|
|
|
Map<String, List<Object>> groupedSubEntities = new HashMap<>(); |
|
|
|
|
for (GatherEntities g : allEntities) { |
|
|
|
|
g.getSubEntities().forEach((table, list) -> { |
|
|
|
|
if (list != null && !list.isEmpty()) { |
|
|
|
|
groupedSubEntities.computeIfAbsent(table, k -> new ArrayList<>()).addAll(list); |
|
|
|
|
} |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 更新采集次数
|
|
|
|
|
Set<String> idSet = allEntities.stream() |
|
|
|
|
.filter(g -> g.getEquGather() != null) |
|
|
|
|
.map(g -> g.getEquGather().getEquId()) |
|
|
|
|
for (Map.Entry<String, List<Object>> entry : groupedSubEntities.entrySet()) { |
|
|
|
|
String tableName = entry.getKey(); |
|
|
|
|
List<Object> list = entry.getValue(); |
|
|
|
|
insertEntityList(tableName, list); // 内部异常直接抛出
|
|
|
|
|
log.info("子表 {} 批量插入完成，数量：{}", tableName, list.size()); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// ===== 3. 更新 equ_base =====
|
|
|
|
|
List<EquBase> updateList = allEntities.stream() |
|
|
|
|
.map(GatherEntities::getUpdateEntities) |
|
|
|
|
.filter(map -> map != null && map.get("equ_base") != null) |
|
|
|
|
.map(map -> (EquBase) map.get("equ_base")) |
|
|
|
|
.filter(Objects::nonNull) |
|
|
|
|
.collect(Collectors.toList()); |
|
|
|
|
|
|
|
|
|
if (!updateList.isEmpty()) { |
|
|
|
|
equBaseMapper.updateEquBaseVersionBatch(updateList); |
|
|
|
|
log.info("equ_base 批量更新完成，数量：{}", updateList.size()); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// ===== 4. 更新采集次数 =====
|
|
|
|
|
Set<String> equIds = allEntities.stream() |
|
|
|
|
.map(GatherEntities::getEquGather) |
|
|
|
|
.filter(Objects::nonNull) |
|
|
|
|
.map(EquGather::getEquId) |
|
|
|
|
.filter(Objects::nonNull) |
|
|
|
|
.collect(Collectors.toSet()); |
|
|
|
|
if (!idSet.isEmpty()) { |
|
|
|
|
equBaseMapper.incrementCollectCount(new ArrayList<>(idSet)); |
|
|
|
|
|
|
|
|
|
if (!equIds.isEmpty()) { |
|
|
|
|
equBaseMapper.incrementCollectCount(new ArrayList<>(equIds)); |
|
|
|
|
log.info("采集次数更新完成，设备数量：{}", equIds.size()); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 执行最新预警
|
|
|
|
|
// ===== 5. 执行预警（容错，不影响主事务） =====
|
|
|
|
|
try { |
|
|
|
|
iAlertLaunchService.executeImmediateAlerts(); |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("预警执行异常，不影响采集事务", e); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
log.info("采集完一次！！！"); |
|
|
|
|
return result; |
|
|
|
|
log.info("批量落库、更新完成，共处理设备：{}", allEntities.size()); |
|
|
|
|
return allEntities.size(); |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("批量落库失败，事务回滚", e); |
|
|
|
|
throw new RuntimeException("批量落库失败，事务回滚", e); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
// @Transactional(rollbackFor = Exception.class)
|
|
|
|
|
// public int persistAndUpdate(List<GatherEntities> allEntities) throws ExecutionException, InterruptedException {
|
|
|
|
|
// if (allEntities == null || allEntities.isEmpty()) {
|
|
|
|
|
// return 0;
|
|
|
|
|
// }
|
|
|
|
|
//
|
|
|
|
|
// // 批量落库
|
|
|
|
|
//
|
|
|
|
|
// int result = memoryToolService.persistAll(allEntities);
|
|
|
|
|
//
|
|
|
|
|
// //批量跟新
|
|
|
|
|
// List<EquBase> updateList = allEntities.stream()
|
|
|
|
|
// .map(GatherEntities::getUpdateEntities)
|
|
|
|
|
// .filter(map -> map != null && map.get("equ_base") != null)
|
|
|
|
|
// .map(map -> (EquBase) map.get("equ_base"))
|
|
|
|
|
// .filter(Objects::nonNull)
|
|
|
|
|
// .collect(Collectors.toList());
|
|
|
|
|
//
|
|
|
|
|
// if (!updateList.isEmpty()) {
|
|
|
|
|
// equBaseMapper.updateEquBaseVersionBatch(updateList);
|
|
|
|
|
// }
|
|
|
|
|
//
|
|
|
|
|
// // 更新采集次数
|
|
|
|
|
// Set<String> idSet = allEntities.stream()
|
|
|
|
|
// .filter(g -> g.getEquGather() != null)
|
|
|
|
|
// .map(g -> g.getEquGather().getEquId())
|
|
|
|
|
// .filter(Objects::nonNull)
|
|
|
|
|
// .collect(Collectors.toSet());
|
|
|
|
|
// if (!idSet.isEmpty()) {
|
|
|
|
|
// equBaseMapper.incrementCollectCount(new ArrayList<>(idSet));
|
|
|
|
|
// }
|
|
|
|
|
//
|
|
|
|
|
// try {
|
|
|
|
|
// iAlertLaunchService.executeImmediateAlerts(); // 内部规则级容错
|
|
|
|
|
// } catch (Exception e) {
|
|
|
|
|
// log.error("预警执行异常，统计预警次数出现错误，不影响采集", e);
|
|
|
|
|
// }
|
|
|
|
|
// log.info("采集完一次！！！");
|
|
|
|
|
// return result;
|
|
|
|
|
// }
|
|
|
|
|
|
|
|
|
|
// @Override
|
|
|
|
|
// public int executeImmediately() throws ExecutionException, InterruptedException {
|
|
|
|
|
@ -564,6 +524,37 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService { |
|
|
|
|
//
|
|
|
|
|
// }
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
public void insertEntityList(String tableName, List<Object> entities){ |
|
|
|
|
if (entities == null || entities.isEmpty()) { |
|
|
|
|
throw new IllegalArgumentException("批量插入实体列表为空，表：" + tableName); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
try { |
|
|
|
|
switch (tableName) { |
|
|
|
|
case "equ_interface": |
|
|
|
|
equInterfaceMapper.insertBatch((List<EquInterface>)(List<?>) entities); |
|
|
|
|
break; |
|
|
|
|
case "equ_routing": |
|
|
|
|
equRoutingMapper.insertBatchRouting((List<EquRouting>)(List<?>) entities); |
|
|
|
|
break; |
|
|
|
|
case "equ_veneer": |
|
|
|
|
equVeneerMapper.insertBatchVeneer((List<EquVeneer>)(List<?>) entities); |
|
|
|
|
break; |
|
|
|
|
case "equ_vlan": |
|
|
|
|
equVlanMapper.insertBatchVlan((List<EquVlan>)(List<?>) entities); |
|
|
|
|
break; |
|
|
|
|
default: |
|
|
|
|
throw new UnsupportedOperationException("未实现的表批量插入：" + tableName); |
|
|
|
|
} |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
// 打印日志并抛异常，保证事务回滚
|
|
|
|
|
log.error("批量插入表 {} 失败, 数据量: {}", tableName, entities.size(), e); |
|
|
|
|
throw new RuntimeException("批量插入表 " + tableName + " 失败", e); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// 健康度计算函数示例
|
|
|
|
|
private int calculateTemperatureHealth(Double temp) { |
|
|
|
|
if (temp == null) return 4; // 无数据视为需维护
|
|
|
|
|
|
xiaohuo
8 months ago