package com.zksy.api.service; import com.zksy.api.domain.AlarmData; import com.zksy.api.mapper.AlarmDataMapper; import com.zksy.api.mapper.AlarmLogMapper; import com.zksy.api.enums.AlarmLogOperTypeEnum; import com.zksy.api.support.AlarmLifecycleSupport; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.support.TransactionTemplate; import java.time.LocalDateTime; import java.util.List; import java.util.Objects; /** * 采集告警及离线告警的统一入库规则;告警处理状态与设备实时报警状态互相独立。 */ @Slf4j @Service public class CollectionAlarmWriter { private final AlarmDataMapper mapper; private final AlarmLogMapper logMapper; private final TransactionTemplate transaction; public CollectionAlarmWriter(AlarmDataMapper mapper, AlarmLogMapper logMapper, PlatformTransactionManager transactionManager) { this.mapper = mapper; this.logMapper = logMapper; transaction = new TransactionTemplate(transactionManager); transaction.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); transaction.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED); transaction.setTimeout(10); } /** * 已有未处理告警时,同级跳过,等级升降则更新原记录及对应采集值。 * * @return true表示新增,false表示复用原记录(更新或跳过);真实入库失败抛出异常。 */ public boolean saveOrUpdateUnhandled(AlarmData alarm) { if (alarm == null) throw new IllegalArgumentException("告警不能为空"); validateKey(alarm.getDeviceCode(), alarm.getWarningCode()); String warningName = firstNonBlank(mapper.selectWarningName(alarm.getWarningCode()), alarm.getWarningCode(), "监测报警"); alarm.setAlarmContent(AlarmLifecycleSupport.alarmContent(warningName, alarm.getActualValue(), alarm.getMinValue(), alarm.getMaxValue(), mapper.selectWarningUnit(alarm.getWarningCode()))); if (alarm.getCreateTime() == null) alarm.setCreateTime(LocalDateTime.now()); if (alarm.getAlarmStatus() == null) alarm.setAlarmStatus(0); return Boolean.TRUE.equals(transaction.execute(status -> { lockKey(alarm.getDeviceCode(), alarm.getWarningCode()); AlarmData existing = mapper.selectUnhandledAlarmForUpdate(alarm.getDeviceCode(), alarm.getWarningCode()); if (existing != null) { // 缺失等级的报文不能清除已有有效等级;真实采集端提供1-4级。 if (alarm.getAlarmLevel() != null && !Objects.equals(existing.getAlarmLevel(), alarm.getAlarmLevel())) { if (mapper.updateUnhandledAlarmLevel(existing.getId(), alarm) != 1) { throw new IllegalStateException("未处理告警等级更新未成功"); } log.info("更新未处理告警等级: deviceCode={}, warningCode={}, alarmId={}, level={}->{}", alarm.getDeviceCode(), alarm.getWarningCode(), existing.getId(), existing.getAlarmLevel(), alarm.getAlarmLevel()); } else { log.debug("跳过同级未处理告警: deviceCode={}, warningCode={}, existingAlarmId={}", alarm.getDeviceCode(), alarm.getWarningCode(), existing.getId()); } return false; } // 新记录由程序赋号;复用/恢复已有记录不会生成或修改号码,不信任请求带入的号码。 if (!Integer.valueOf(1).equals(mapper.lockAlarmNumberGeneration())) { throw new IllegalStateException("未取得报警单号生成事务锁"); } alarm.setAlarmNo(AlarmLifecycleSupport.nextAlarmNo(mapper::existsAlarmNo)); if (mapper.insert(alarm) != 1) throw new IllegalStateException("告警入库未成功"); writeLifecycleLog(alarm.getId(), AlarmLogOperTypeEnum.CREATED, alarm.getAlarmContent(), "监测系统", alarm.getCreateTime(), 0L); return true; })); } private String firstNonBlank(String... values) { for (String value : values) { if (value != null && !value.trim().isEmpty()) return value.trim(); } return "监测报警"; } /** * 有效恢复数据才调用;0行表示无待处理告警,重复正常上报不会改写历史处理信息。 */ public int markRecovered(String deviceCode, String warningCode) { validateKey(deviceCode, warningCode); return transaction.execute(status -> { lockKey(deviceCode, warningCode); List alarmIds = mapper.selectRecoverableAlarmIdsForUpdate(deviceCode, warningCode); for (String alarmId : alarmIds) { long beforeLogId = logMapper.latestLogId(alarmId); if (mapper.markAlarmRecovered(alarmId) != 1) throw new IllegalStateException("告警恢复更新未成功"); writeLifecycleLog(alarmId, AlarmLogOperTypeEnum.HANDLED, "监测恢复,报警自动解除", "系统自动", LocalDateTime.now(), beforeLogId); } int count = alarmIds.size(); if (count > 0) { log.info("正常采集数据自动处理告警: deviceCode={}, warningCode={}, count={}", deviceCode, warningCode, count); } return count; }); } private void writeLifecycleLog(String alarmId, AlarmLogOperTypeEnum type, String description, String actor, LocalDateTime time, long afterLogId) { if (alarmId == null || alarmId.trim().isEmpty()) throw new IllegalStateException("报警日志缺少报警ID"); int count = logMapper.insertLifecycleIfAbsent(alarmId, type.getCode(), description, actor, time, afterLogId); if (count != 1 && !(count == 0 && logMapper.existsLifecycle(alarmId, type.getCode(), afterLogId))) { throw new IllegalStateException("报警日志写入失败"); } } private void lockKey(String deviceCode, String warningCode) { if (!Integer.valueOf(1).equals(mapper.lockAlarmKey(deviceCode, warningCode))) { throw new IllegalStateException("未取得告警入库事务锁"); } } private void validateKey(String deviceCode, String warningCode) { if (deviceCode == null || deviceCode.trim().isEmpty() || warningCode == null || warningCode.trim().isEmpty()) { throw new IllegalArgumentException("告警必须包含device_code和warning_code"); } } }