wangwei_123
2 days ago
39 changed files with 1607 additions and 808 deletions
@ -1,119 +1,169 @@ |
|||||
package cn.iocoder.yudao.module.mqtt.config; |
package cn.iocoder.yudao.module.mqtt.config; |
||||
|
|
||||
|
import jakarta.annotation.PostConstruct; |
||||
import jakarta.annotation.PreDestroy; |
import jakarta.annotation.PreDestroy; |
||||
import lombok.extern.slf4j.Slf4j; |
import lombok.extern.slf4j.Slf4j; |
||||
|
|
||||
import java.util.ArrayList; |
import java.util.ArrayList; |
||||
|
import java.util.Collection; |
||||
import java.util.List; |
import java.util.List; |
||||
import java.util.concurrent.BlockingQueue; |
|
||||
import java.util.concurrent.LinkedBlockingQueue; |
|
||||
import java.util.concurrent.TimeUnit; |
|
||||
|
import java.util.concurrent.*; |
||||
import java.util.concurrent.atomic.AtomicBoolean; |
import java.util.concurrent.atomic.AtomicBoolean; |
||||
import java.util.function.Consumer; |
import java.util.function.Consumer; |
||||
|
|
||||
@Slf4j |
@Slf4j |
||||
public class TdengineBatchConfig<T> { |
public class TdengineBatchConfig<T> { |
||||
|
|
||||
// --- 这些参数可以在构造时传入，使其更灵活 ---
|
|
||||
private final int queueCapacity; |
private final int queueCapacity; |
||||
private final int batchSize; |
private final int batchSize; |
||||
private final long offerTimeoutMs; |
|
||||
|
private final long maxWaitTimeMs; |
||||
|
private final String processorName; |
||||
|
|
||||
private final String processorName; // 用于日志，区分不同的处理器实例
|
|
||||
private final BlockingQueue<T> dataQueue; |
private final BlockingQueue<T> dataQueue; |
||||
private final Consumer<List<T>> batchAction; // 【核心】用于处理一个批次的具体业务逻辑
|
|
||||
|
private final Consumer<List<T>> batchAction; |
||||
|
private final AtomicBoolean running = new AtomicBoolean(true); |
||||
|
private Thread workerThread; |
||||
|
|
||||
private final AtomicBoolean shuttingDown = new AtomicBoolean(false); |
|
||||
|
|
||||
/** |
|
||||
* 构造函数 |
|
||||
* @param processorName 处理器名称，用于日志区分 |
|
||||
* @param batchAction 一个函数，接收一个 List<T> 并执行相应的批量操作（例如，写入数据库） |
|
||||
* @param queueCapacity 队列容量 |
|
||||
* @param batchSize 批次大小 |
|
||||
* @param fixedRateMs 执行频率 |
|
||||
*/ |
|
||||
public TdengineBatchConfig(String processorName, Consumer<List<T>> batchAction, |
public TdengineBatchConfig(String processorName, Consumer<List<T>> batchAction, |
||||
int queueCapacity, int batchSize, long fixedRateMs) { |
|
||||
|
int queueCapacity, int batchSize, long maxWaitTimeMs) { |
||||
this.processorName = processorName; |
this.processorName = processorName; |
||||
this.batchAction = batchAction; |
this.batchAction = batchAction; |
||||
this.queueCapacity = queueCapacity; |
this.queueCapacity = queueCapacity; |
||||
this.batchSize = batchSize; |
this.batchSize = batchSize; |
||||
this.offerTimeoutMs = 100L; |
|
||||
|
this.maxWaitTimeMs = maxWaitTimeMs; |
||||
this.dataQueue = new LinkedBlockingQueue<>(this.queueCapacity); |
this.dataQueue = new LinkedBlockingQueue<>(this.queueCapacity); |
||||
|
startWorker(); |
||||
} |
} |
||||
|
|
||||
public void addToBatch(T data) { |
|
||||
if (data == null) { |
|
||||
return; |
|
||||
|
private void startWorker() { |
||||
|
this.workerThread = new Thread(this::processLoop, "TD-Worker-" + processorName); |
||||
|
this.workerThread.setDaemon(true); |
||||
|
this.workerThread.start(); |
||||
|
log.info("[{}] 批处理线程已启动，批次大小: {}, 最大等待: {}ms", |
||||
|
processorName, batchSize, maxWaitTimeMs); |
||||
|
} |
||||
|
|
||||
|
private void processLoop() { |
||||
|
List<T> buffer = new ArrayList<>(batchSize); |
||||
|
long lastFlushTime = System.currentTimeMillis(); |
||||
|
|
||||
|
while (running.get()) { |
||||
|
try { |
||||
|
long now = System.currentTimeMillis(); |
||||
|
long waitTime = maxWaitTimeMs - (now - lastFlushTime); |
||||
|
if (waitTime <= 0) waitTime = 1; |
||||
|
|
||||
|
T firstItem = dataQueue.poll(waitTime, TimeUnit.MILLISECONDS); |
||||
|
|
||||
|
if (firstItem != null) { |
||||
|
buffer.add(firstItem); |
||||
|
dataQueue.drainTo(buffer, batchSize - buffer.size()); |
||||
|
} |
||||
|
|
||||
|
boolean sizeReached = buffer.size() >= batchSize; |
||||
|
boolean timeReached = (System.currentTimeMillis() - lastFlushTime) >= maxWaitTimeMs; |
||||
|
|
||||
|
if ((sizeReached || timeReached) && !buffer.isEmpty()) { |
||||
|
doFlush(buffer); |
||||
|
buffer.clear(); |
||||
|
lastFlushTime = System.currentTimeMillis(); |
||||
|
} |
||||
|
|
||||
|
} catch (InterruptedException e) { |
||||
|
log.warn("[{}] 工作线程被中断", processorName); |
||||
|
Thread.currentThread().interrupt(); |
||||
|
break; |
||||
|
} catch (Exception e) { |
||||
|
log.error("[{}] Loop异常", processorName, e); |
||||
|
} |
||||
} |
} |
||||
// 如果正在关闭，则不再接受新数据
|
|
||||
if (shuttingDown.get()) { |
|
||||
log.warn("[{}] 正在关闭，已拒绝添加新数据。", this.processorName); |
|
||||
return; |
|
||||
|
if (!buffer.isEmpty()) doFlush(buffer); |
||||
|
} |
||||
|
|
||||
|
private void doFlush(List<T> batch) { |
||||
|
try { |
||||
|
if (log.isDebugEnabled()) { |
||||
|
log.debug("[{}] 触发批量入库，数量: {}", processorName, batch.size()); |
||||
|
} |
||||
|
batchAction.accept(new ArrayList<>(batch)); |
||||
|
} catch (Exception e) { |
||||
|
log.error("[{}] 批量入库失败！数量: {}", processorName, batch.size(), e); |
||||
} |
} |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 单条添加（保留兼容性） |
||||
|
*/ |
||||
|
public void addToBatch(T data) { |
||||
|
if (data == null || !running.get()) return; |
||||
|
|
||||
try { |
try { |
||||
if (!dataQueue.offer(data, offerTimeoutMs, TimeUnit.MILLISECONDS)) { |
|
||||
log.warn("[{}] 缓冲区已满且在 {} 毫秒内无法添加，数据可能被丢弃！当前队列大小: {}", |
|
||||
this.processorName, this.offerTimeoutMs, dataQueue.size()); |
|
||||
|
if (!dataQueue.offer(data, 100, TimeUnit.MILLISECONDS)) { |
||||
|
log.warn("[{}] 队列已满({})，丢弃数据", processorName, queueCapacity); |
||||
} |
} |
||||
} catch (InterruptedException e) { |
} catch (InterruptedException e) { |
||||
log.error("[{}] 添加数据到缓冲区时线程被中断", this.processorName, e); |
|
||||
Thread.currentThread().interrupt(); |
Thread.currentThread().interrupt(); |
||||
} |
} |
||||
} |
} |
||||
|
|
||||
// 注意：@Scheduled 注解不能用在非 Spring Bean 的普通类方法上。
|
|
||||
// 我们将在下一步的配置类中解决这个问题。
|
|
||||
public void flush() { |
|
||||
if (dataQueue.isEmpty()) { |
|
||||
|
/** |
||||
|
* 批量添加（新增优化方法）- 关键优化点！ |
||||
|
*/ |
||||
|
public void addToBatch(Collection<T> dataList) { |
||||
|
if (dataList == null || dataList.isEmpty() || !running.get()) { |
||||
return; |
return; |
||||
} |
} |
||||
List<T> batchList = new ArrayList<>(batchSize); |
|
||||
try { |
|
||||
int drainedCount = dataQueue.drainTo(batchList, batchSize); |
|
||||
if (drainedCount > 0) { |
|
||||
log.debug("[{}] 定时任务触发，准备将 {} 条数据进行批量处理...", this.processorName, drainedCount); |
|
||||
// 调用构造时传入的业务逻辑
|
|
||||
this.batchAction.accept(batchList); |
|
||||
log.debug("[{}] 成功批量处理 {} 条数据。", this.processorName, drainedCount); |
|
||||
|
|
||||
|
int added = 0; |
||||
|
int dropped = 0; |
||||
|
|
||||
|
for (T data : dataList) { |
||||
|
if (data == null) continue; |
||||
|
|
||||
|
try { |
||||
|
// 批量添加时使用较短的超时，避免阻塞太久
|
||||
|
if (dataQueue.offer(data, 10, TimeUnit.MILLISECONDS)) { |
||||
|
added++; |
||||
|
} else { |
||||
|
dropped++; |
||||
|
} |
||||
|
} catch (InterruptedException e) { |
||||
|
Thread.currentThread().interrupt(); |
||||
|
log.warn("[{}] 批量添加被中断，已添加: {}, 剩余: {}", |
||||
|
processorName, added, dataList.size() - added); |
||||
|
break; |
||||
} |
} |
||||
} catch (Exception e) { |
|
||||
log.error("[{}] 批量处理数据时发生严重错误！数据量: {}", this.processorName, batchList.size(), e); |
|
||||
|
} |
||||
|
|
||||
|
if (dropped > 0) { |
||||
|
log.warn("[{}] 批量添加完成，成功: {}, 丢弃: {} (队列已满)", |
||||
|
processorName, added, dropped); |
||||
|
} else if (log.isDebugEnabled()) { |
||||
|
log.debug("[{}] 批量添加完成，数量: {}", processorName, added); |
||||
} |
} |
||||
} |
} |
||||
|
|
||||
|
/** |
||||
|
* 获取当前队列大小（用于监控） |
||||
|
*/ |
||||
|
public int getQueueSize() { |
||||
|
return dataQueue.size(); |
||||
|
} |
||||
|
|
||||
@PreDestroy |
@PreDestroy |
||||
public void onShutdown() { |
public void onShutdown() { |
||||
log.info("[{}] 应用即将关闭，开始执行最后的缓冲区数据刷新...", this.processorName); |
|
||||
// 1. 设置关闭标志，阻止新数据进入
|
|
||||
shuttingDown.set(true); |
|
||||
|
|
||||
// 2. 将队列中剩余的所有数据一次性取出到一个临时列表
|
|
||||
List<T> remainingData = new ArrayList<>(); |
|
||||
dataQueue.drainTo(remainingData); |
|
||||
|
|
||||
if (remainingData.isEmpty()) { |
|
||||
log.info("[{}] 缓冲区为空，无需处理。", this.processorName); |
|
||||
return; |
|
||||
} |
|
||||
log.info("[{}] 停机处理中，剩余 {} 条数据待处理...", this.processorName, remainingData.size()); |
|
||||
|
|
||||
// 3. 对取出的数据进行分批处理
|
|
||||
for (int i = 0; i < remainingData.size(); i += batchSize) { |
|
||||
// 计算当前批次的结束索引
|
|
||||
int end = Math.min(i + batchSize, remainingData.size()); |
|
||||
// 获取子列表作为当前批次
|
|
||||
List<T> batch = remainingData.subList(i, end); |
|
||||
|
log.info("[{}] 正在停止...", processorName); |
||||
|
running.set(false); |
||||
|
|
||||
|
if (workerThread != null) { |
||||
|
workerThread.interrupt(); |
||||
try { |
try { |
||||
log.debug("[{}] 正在处理最后批次的数据，数量: {}", this.processorName, batch.size()); |
|
||||
this.batchAction.accept(batch); |
|
||||
} catch (Exception e) { |
|
||||
log.error("[{}] 关闭过程中批量处理数据时发生严重错误！数据量: {}", this.processorName, batch.size(), e); |
|
||||
|
workerThread.join(5000); |
||||
|
} catch (InterruptedException e) { |
||||
|
Thread.currentThread().interrupt(); |
||||
} |
} |
||||
} |
} |
||||
|
|
||||
log.info("[{}] 缓冲区数据已全部处理完毕。", this.processorName); |
|
||||
|
log.info("[{}] 已停止。剩余队列: {}", processorName, dataQueue.size()); |
||||
} |
} |
||||
|
|
||||
} |
} |
||||
@ -0,0 +1,917 @@ |
|||||
|
package cn.iocoder.yudao.module.mqtt.processor; |
||||
|
|
||||
|
import cn.hutool.json.JSONObject; |
||||
|
import cn.hutool.json.JSONUtil; |
||||
|
import cn.iocoder.yudao.framework.common.util.object.BeanUtils; |
||||
|
import cn.iocoder.yudao.module.hand.dal.*; |
||||
|
import cn.iocoder.yudao.module.hand.enums.*; |
||||
|
import cn.iocoder.yudao.module.hand.service.*; |
||||
|
import cn.iocoder.yudao.module.hand.util.*; |
||||
|
import cn.iocoder.yudao.module.hand.vo.*; |
||||
|
import cn.iocoder.yudao.module.mqtt.config.TdengineBatchConfig; |
||||
|
import cn.iocoder.yudao.module.mqtt.mqtt.Client; |
||||
|
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; |
||||
|
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils; |
||||
|
import com.fasterxml.jackson.core.JsonProcessingException; |
||||
|
import com.fasterxml.jackson.databind.ObjectMapper; |
||||
|
import jakarta.annotation.Resource; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.apache.commons.lang3.StringUtils; |
||||
|
import org.apache.kafka.clients.consumer.ConsumerRecord; |
||||
|
import org.springframework.core.task.TaskExecutor; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
|
||||
|
import java.sql.Timestamp; |
||||
|
import java.time.LocalDateTime; |
||||
|
import java.util.*; |
||||
|
import java.util.function.Function; |
||||
|
import java.util.stream.Collectors; |
||||
|
|
||||
|
/** |
||||
|
* 批量设备消息处理器 |
||||
|
* <p> |
||||
|
* 核心优化： |
||||
|
* 1. 批量获取基础数据（租户信息、设备信息、报警规则） |
||||
|
* 2. 内存中完成所有业务逻辑计算 |
||||
|
* 3. 批量执行所有数据库写操作 |
||||
|
*/ |
||||
|
@Slf4j |
||||
|
@Component |
||||
|
public class BatchDeviceMessageProcessor { |
||||
|
|
||||
|
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); |
||||
|
|
||||
|
@Resource |
||||
|
private RedisUtil redisUtil; |
||||
|
@Resource |
||||
|
private TdengineBatchConfig<HandOriginalLog> tdengineBatchProcessor; |
||||
|
@Resource |
||||
|
private TdengineBatchConfig<TdengineDataVo> tdengineBatchConfig; |
||||
|
@Resource |
||||
|
private TdengineBatchConfig<AlarmMessageLog> alarmProcessor; |
||||
|
@Resource |
||||
|
private HandDetectorService handDetectorService; |
||||
|
@Resource |
||||
|
private HandAlarmService handAlarmService; |
||||
|
@Resource |
||||
|
private AlarmRuleService alarmRuleService; |
||||
|
@Resource |
||||
|
private FenceService fenceService; |
||||
|
@Resource |
||||
|
private FenceAlarmService fenceAlarmService; |
||||
|
@Resource |
||||
|
private Client client; |
||||
|
@Resource(name = "mqttAlarmExecutor") |
||||
|
private TaskExecutor alarmExecutor; |
||||
|
|
||||
|
/** |
||||
|
* 批量处理 Kafka 消息的主入口 |
||||
|
*/ |
||||
|
public void processBatch(List<ConsumerRecord<String, String>> records) { |
||||
|
long startTime = System.currentTimeMillis(); |
||||
|
|
||||
|
if (CollectionUtils.isEmpty(records)) { |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
log.info("[批量处理] 开始处理，消息数量: {}", records.size()); |
||||
|
|
||||
|
try { |
||||
|
// === 阶段1: 数据准备 ===
|
||||
|
BatchContext context = prepareBatchContext(records); |
||||
|
|
||||
|
if (context.isEmpty()) { |
||||
|
log.warn("[批量处理] 无有效数据，处理结束"); |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
// === 阶段2: 业务逻辑处理 ===
|
||||
|
processBatchLogic(records, context); |
||||
|
|
||||
|
// === 阶段3: 批量持久化 ===
|
||||
|
persistBatchData(context); |
||||
|
|
||||
|
log.info("[批量处理] 完成，消息数量: {}，耗时: {} ms", |
||||
|
records.size(), System.currentTimeMillis() - startTime); |
||||
|
|
||||
|
} catch (Exception e) { |
||||
|
log.error("[批量处理] 发生异常", e); |
||||
|
// 根据业务需求决定是否需要回滚或重试
|
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private BatchContext prepareBatchContext(List<ConsumerRecord<String, String>> records) { |
||||
|
BatchContext context = new BatchContext(); |
||||
|
|
||||
|
// 1. 提取所有有效的 SNs
|
||||
|
List<String> sns = records.stream() |
||||
|
.map(ConsumerRecord::key) |
||||
|
.filter(StringUtils::isNotBlank) |
||||
|
.distinct() |
||||
|
.toList(); |
||||
|
|
||||
|
if (sns.isEmpty()) { |
||||
|
return context; |
||||
|
} |
||||
|
|
||||
|
// 2. 批量获取租户信息
|
||||
|
context.snToTenantMap = getTenantIdsInBatch(sns); |
||||
|
|
||||
|
// 3. 按租户分组 SN
|
||||
|
Map<Long, List<String>> tenantToSnsMap = sns.stream() |
||||
|
.filter(context.snToTenantMap::containsKey) |
||||
|
.collect(Collectors.groupingBy(context.snToTenantMap::get)); |
||||
|
|
||||
|
// 4. 批量获取设备信息
|
||||
|
context.snToDeviceMap = getDeviceVosInBatch(tenantToSnsMap); |
||||
|
|
||||
|
// 5. 批量获取报警规则
|
||||
|
Set<Long> tenantIds = tenantToSnsMap.keySet(); |
||||
|
context.tenantAlarmRules = getAlarmRulesForTenants(tenantIds); |
||||
|
|
||||
|
// 6. 批量获取围栏信息
|
||||
|
context.fenceCache = getFenceInfoBatch(context.snToDeviceMap.values()); |
||||
|
|
||||
|
return context; |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 阶段2: 处理业务逻辑（内存操作） |
||||
|
*/ |
||||
|
private void processBatchLogic(List<ConsumerRecord<String, String>> records, BatchContext context) { |
||||
|
for (ConsumerRecord<String, String> record : records) { |
||||
|
try { |
||||
|
processSingleMessage(record, context); |
||||
|
} catch (Exception e) { |
||||
|
log.error("[批量处理] 处理单条消息失败，SN: {}", record.key(), e); |
||||
|
// 继续处理下一条，不中断整个批次
|
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 处理单条消息 |
||||
|
*/ |
||||
|
private void processSingleMessage(ConsumerRecord<String, String> record, BatchContext context) { |
||||
|
String sn = record.key(); |
||||
|
String payload = record.value(); |
||||
|
|
||||
|
// 1. 基础校验
|
||||
|
Long tenantId = context.snToTenantMap.get(sn); |
||||
|
HandDataVo device = context.snToDeviceMap.get(sn); |
||||
|
|
||||
|
if (tenantId == null || device == null) { |
||||
|
log.warn("[批量处理] SN {} 无租户或设备信息，跳过", sn); |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
// 2. 保存原始日志
|
||||
|
context.originalLogs.add(createOriginalLog(sn, payload, tenantId)); |
||||
|
|
||||
|
// 3. 检查设备是否启用
|
||||
|
if (EnableStatus.DISABLED.value().equals(device.getEnableStatus())) { |
||||
|
log.debug("[批量处理] 设备未启用，SN: {}", sn); |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
// 4. 数据解析与转换
|
||||
|
HandDataVo handVo = parseAndConvertData(sn, payload, device); |
||||
|
|
||||
|
// 5. 获取报警规则
|
||||
|
Map<Long, List<AlarmRuleDO>> ruleMap = context.tenantAlarmRules.get(tenantId); |
||||
|
AlarmRuleDO alarmRule = getAlarmRule(handVo, ruleMap); |
||||
|
|
||||
|
// 6. 气体报警处理
|
||||
|
processGasAlarm(handVo, alarmRule, context); |
||||
|
|
||||
|
// 7. 电池报警处理
|
||||
|
processBatteryAlarm(handVo, context); |
||||
|
|
||||
|
// 8. 围栏报警处理
|
||||
|
processFenceAlarm(handVo, context); |
||||
|
|
||||
|
// 9. 保存处理后的数据日志
|
||||
|
context.processedLogs.add(createTdengineDataVo(handVo)); |
||||
|
|
||||
|
// 10. 记录需要更新到 Redis 的数据
|
||||
|
context.redisUpdates.put(sn, handVo); |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 阶段3: 批量持久化数据 |
||||
|
*/ |
||||
|
private void persistBatchData(BatchContext context) { |
||||
|
// 1. 批量保存原始日志
|
||||
|
if (!context.originalLogs.isEmpty()) { |
||||
|
tdengineBatchProcessor.addToBatch(context.originalLogs); |
||||
|
log.debug("[批量持久化] 原始日志: {} 条", context.originalLogs.size()); |
||||
|
} |
||||
|
|
||||
|
// 2. 批量保存处理后日志
|
||||
|
if (!context.processedLogs.isEmpty()) { |
||||
|
tdengineBatchConfig.addToBatch(context.processedLogs); |
||||
|
log.debug("[批量持久化] 处理日志: {} 条", context.processedLogs.size()); |
||||
|
} |
||||
|
|
||||
|
// 3. 批量保存报警消息日志
|
||||
|
if (!context.alarmMessageLogs.isEmpty()) { |
||||
|
alarmProcessor.addToBatch(context.alarmMessageLogs); |
||||
|
log.debug("[批量持久化] 报警消息: {} 条", context.alarmMessageLogs.size()); |
||||
|
} |
||||
|
|
||||
|
// 4. 批量创建气体报警
|
||||
|
if (!context.gasAlarmsToCreate.isEmpty()) { |
||||
|
handAlarmService.batchCreateHandAlarm(context.gasAlarmsToCreate); |
||||
|
log.debug("[批量持久化] 新增气体报警: {} 条", context.gasAlarmsToCreate.size()); |
||||
|
} |
||||
|
|
||||
|
// 5. 批量更新气体报警
|
||||
|
if (!context.gasAlarmsToUpdate.isEmpty()) { |
||||
|
handAlarmService.batchUpdateById(context.gasAlarmsToUpdate); |
||||
|
log.debug("[批量持久化] 更新气体报警: {} 条", context.gasAlarmsToUpdate.size()); |
||||
|
} |
||||
|
|
||||
|
// 6. 批量创建围栏报警
|
||||
|
if (!context.fenceAlarmsToCreate.isEmpty()) { |
||||
|
fenceAlarmService.batchCreateFenceAlarm(context.fenceAlarmsToCreate); |
||||
|
log.debug("[批量持久化] 新增围栏报警: {} 条", context.fenceAlarmsToCreate.size()); |
||||
|
} |
||||
|
|
||||
|
// 7. 批量更新围栏报警
|
||||
|
if (!context.fenceAlarmsToUpdate.isEmpty()) { |
||||
|
fenceAlarmService.batchUpdateById(context.fenceAlarmsToUpdate); |
||||
|
log.debug("[批量持久化] 更新围栏报警: {} 条", context.fenceAlarmsToUpdate.size()); |
||||
|
} |
||||
|
|
||||
|
// 8. 批量更新 Redis
|
||||
|
if (!context.redisUpdates.isEmpty()) { |
||||
|
batchUpdateRedis(context.redisUpdates, context.snToTenantMap); |
||||
|
log.debug("[批量持久化] Redis 更新: {} 条", context.redisUpdates.size()); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
// ========== 基础数据获取方法 ==========
|
||||
|
|
||||
|
/** |
||||
|
* 批量获取租户ID映射 |
||||
|
*/ |
||||
|
private Map<String, Long> getTenantIdsInBatch(List<String> sns) { |
||||
|
Map<String, Long> result = new HashMap<>(); |
||||
|
|
||||
|
try { |
||||
|
List<Object> tenantIdObjs = redisUtil.hmget( |
||||
|
RedisKeyUtil.getDeviceTenantMappingKey(), |
||||
|
new ArrayList<>(sns) |
||||
|
); |
||||
|
|
||||
|
for (int i = 0; i < sns.size(); i++) { |
||||
|
Object tenantIdObj = tenantIdObjs.get(i); |
||||
|
if (tenantIdObj != null && StringUtils.isNotBlank(tenantIdObj.toString())) { |
||||
|
result.put(sns.get(i), Long.parseLong(tenantIdObj.toString())); |
||||
|
} |
||||
|
} |
||||
|
} catch (Exception e) { |
||||
|
log.error("[批量获取] 获取租户ID失败", e); |
||||
|
} |
||||
|
|
||||
|
return result; |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 批量获取设备信息（优先从 Redis） |
||||
|
*/ |
||||
|
private Map<String, HandDataVo> getDeviceVosInBatch(Map<Long, List<String>> tenantToSnsMap) { |
||||
|
Map<String, HandDataVo> result = new HashMap<>(); |
||||
|
List<String> cacheMissSns = new ArrayList<>(); |
||||
|
|
||||
|
// 从 Redis 批量获取
|
||||
|
tenantToSnsMap.forEach((tenantId, sns) -> { |
||||
|
try { |
||||
|
String redisKey = RedisKeyUtil.getTenantDeviceHashKey(tenantId); |
||||
|
List<Object> cachedObjects = redisUtil.hmget(redisKey, new ArrayList<>(sns)); |
||||
|
|
||||
|
for (int i = 0; i < sns.size(); i++) { |
||||
|
Object obj = cachedObjects.get(i); |
||||
|
String sn = sns.get(i); |
||||
|
|
||||
|
if (obj != null) { |
||||
|
result.put(sn, BeanUtils.toBean(obj, HandDataVo.class)); |
||||
|
} else { |
||||
|
cacheMissSns.add(sn); |
||||
|
} |
||||
|
} |
||||
|
} catch (Exception e) { |
||||
|
log.error("[批量获取] 从Redis获取设备信息失败，tenantId: {}", tenantId, e); |
||||
|
cacheMissSns.addAll(sns); |
||||
|
} |
||||
|
}); |
||||
|
|
||||
|
// 缓存未命中，从数据库加载
|
||||
|
if (!cacheMissSns.isEmpty()) { |
||||
|
try { |
||||
|
QueryWrapper<HandDetectorDO> doQueryWrapper = new QueryWrapper<>(); |
||||
|
doQueryWrapper.in("sn", cacheMissSns); |
||||
|
List<HandDetectorDO> detectors = handDetectorService.listAll(doQueryWrapper); |
||||
|
Map<String, HandDetectorDO> detectorMap = detectors.stream() |
||||
|
.collect(Collectors.toMap(HandDetectorDO::getSn, Function.identity())); |
||||
|
|
||||
|
cacheMissSns.forEach(sn -> { |
||||
|
HandDetectorDO detector = detectorMap.get(sn); |
||||
|
if (detector != null) { |
||||
|
result.put(sn, BeanUtils.toBean(detector, HandDataVo.class)); |
||||
|
} |
||||
|
}); |
||||
|
|
||||
|
log.info("[批量获取] 从数据库加载设备: {} 条", detectors.size()); |
||||
|
} catch (Exception e) { |
||||
|
log.error("[批量获取] 从数据库加载设备失败", e); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
return result; |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 批量获取报警规则 |
||||
|
*/ |
||||
|
private Map<Long, Map<Long, List<AlarmRuleDO>>> getAlarmRulesForTenants(Set<Long> tenantIds) { |
||||
|
Map<Long, Map<Long, List<AlarmRuleDO>>> result = new HashMap<>(); |
||||
|
|
||||
|
try { |
||||
|
for (Long tenantId : tenantIds) { |
||||
|
Map<Long, List<AlarmRuleDO>> rules = alarmRuleService.selectCacheListMap(tenantId); |
||||
|
if (rules != null && !rules.isEmpty()) { |
||||
|
result.put(tenantId, rules); |
||||
|
} |
||||
|
} |
||||
|
} catch (Exception e) { |
||||
|
log.error("[批量获取] 获取报警规则失败", e); |
||||
|
} |
||||
|
|
||||
|
return result; |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 批量获取围栏信息 |
||||
|
*/ |
||||
|
private Map<Long, List<Geofence>> getFenceInfoBatch(Collection<HandDataVo> devices) { |
||||
|
Map<Long, List<Geofence>> result = new HashMap<>(); |
||||
|
|
||||
|
try { |
||||
|
// 收集所有需要查询的围栏ID
|
||||
|
Set<Long> allFenceIds = devices.stream() |
||||
|
.map(HandDataVo::getFenceIds) |
||||
|
.filter(StringUtils::isNotBlank) |
||||
|
.flatMap(ids -> Arrays.stream(ids.split(","))) |
||||
|
.map(Long::parseLong) |
||||
|
.collect(Collectors.toSet()); |
||||
|
|
||||
|
if (!allFenceIds.isEmpty()) { |
||||
|
List<Geofence> fences = fenceService.getFenceList(new ArrayList<>(allFenceIds)); |
||||
|
Map<Long, Geofence> fenceMap = fences.stream() |
||||
|
.collect(Collectors.toMap(Geofence::getId, Function.identity())); |
||||
|
|
||||
|
// 为每个设备构建其对应的围栏列表
|
||||
|
devices.forEach(device -> { |
||||
|
if (StringUtils.isNotBlank(device.getFenceIds())) { |
||||
|
List<Geofence> deviceFences = Arrays.stream(device.getFenceIds().split(",")) |
||||
|
.map(Long::parseLong) |
||||
|
.map(fenceMap::get) |
||||
|
.filter(Objects::nonNull) |
||||
|
.toList(); |
||||
|
|
||||
|
if (!deviceFences.isEmpty()) { |
||||
|
result.put(device.getId(), deviceFences); |
||||
|
} |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
} catch (Exception e) { |
||||
|
log.error("[批量获取] 获取围栏信息失败", e); |
||||
|
} |
||||
|
|
||||
|
return result; |
||||
|
} |
||||
|
|
||||
|
// ========== 业务逻辑处理方法 ==========
|
||||
|
|
||||
|
/** |
||||
|
* 数据解析与转换 |
||||
|
*/ |
||||
|
private HandDataVo parseAndConvertData(String sn, String payload, HandDataVo device) { |
||||
|
try { |
||||
|
JSONObject json = JSONUtil.parseObj(payload); |
||||
|
|
||||
|
Double gasValue = json.getDouble("gas", null); |
||||
|
String loc = json.getStr("loc"); |
||||
|
String battery = json.getStr("battery"); |
||||
|
|
||||
|
device.setValue(gasValue); |
||||
|
device.setSn(sn); |
||||
|
device.setBattery(battery); |
||||
|
device.setTime(new Date()); |
||||
|
device.setOnlineStatus(OnlineStatusType.ONLINE.getType()); |
||||
|
|
||||
|
// 解析位置信息
|
||||
|
if (StringUtils.isNotBlank(loc)) { |
||||
|
String coords = loc.substring(1, loc.length() - 1); |
||||
|
String[] parts = coords.split(","); |
||||
|
|
||||
|
if (parts.length == 3) { |
||||
|
Map<String, Double> converted = CoordinateTransferUtils.wgs84ToGcj02( |
||||
|
Double.parseDouble(parts[0]), |
||||
|
Double.parseDouble(parts[1]) |
||||
|
); |
||||
|
|
||||
|
device.setLongitude(converted.get("lon")); |
||||
|
device.setLatitude(converted.get("lat")); |
||||
|
device.setGpsType(Integer.parseInt(parts[2])); |
||||
|
} |
||||
|
} |
||||
|
} catch (Exception e) { |
||||
|
log.error("[数据转换] 解析失败，SN: {}, payload: {}", sn, payload, e); |
||||
|
} |
||||
|
|
||||
|
return device; |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 气体报警处理 |
||||
|
*/ |
||||
|
private void processGasAlarm(HandDataVo handVo, AlarmRuleDO alarmRule, BatchContext context) { |
||||
|
if (alarmRule == null) { |
||||
|
handVo.setGasStatus(HandAlarmType.NORMAL.getType()); |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
LocalDateTime now = LocalDateTime.now(); |
||||
|
|
||||
|
// 处理离线报警恢复
|
||||
|
if (OnlineStatusType.ONLINE.getType().equals(handVo.getOnlineStatus()) |
||||
|
&& AlarmLevelEnum.OFFLINE.value().equals(handVo.getAlarmLevel()) |
||||
|
&& handVo.getAlarmId() != null) { |
||||
|
|
||||
|
HandAlarmDO alarmToEnd = new HandAlarmDO(); |
||||
|
alarmToEnd.setId(handVo.getAlarmId()); |
||||
|
alarmToEnd.setTAlarmEnd(now); |
||||
|
alarmToEnd.setStatus(EnableStatus.HANDLE.value()); |
||||
|
context.gasAlarmsToUpdate.add(alarmToEnd); |
||||
|
|
||||
|
handVo.setAlarmId(null); |
||||
|
handVo.setAlarmLevel(0); |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
boolean isCurrentlyAlarming = handVo.getAlarmLevel() != null && handVo.getAlarmLevel() > 0; |
||||
|
boolean shouldAlarm = alarmRule.getAlarmLevel() > 0; |
||||
|
|
||||
|
// 报警恢复
|
||||
|
if (isCurrentlyAlarming && !shouldAlarm) { |
||||
|
handVo.setAlarmLevel(0); |
||||
|
handVo.setTAlarmEnd(now); |
||||
|
handVo.setGasStatus(HandAlarmType.NORMAL.getType()); |
||||
|
|
||||
|
// 发送报警结束消息
|
||||
|
sendAlarmMessage(handVo, alarmRule.getGasTypeName(), handVo.getValue(), false, context); |
||||
|
handVo.setLastPushValue(null); |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
// 正常状态
|
||||
|
if (!shouldAlarm) { |
||||
|
handVo.setAlarmLevel(0); |
||||
|
handVo.setGasStatus(HandAlarmType.NORMAL.getType()); |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
// 报警触发或持续
|
||||
|
Integer newLevel = alarmRule.getAlarmLevel(); |
||||
|
|
||||
|
// 首次报警
|
||||
|
if (!isCurrentlyAlarming) { |
||||
|
handVo.setFirstValue(handVo.getValue()); |
||||
|
handVo.setTAlarmStart(now); |
||||
|
handVo.setMaxValue(handVo.getValue()); |
||||
|
handVo.setMaxAlarmLevel(newLevel); |
||||
|
|
||||
|
// 创建新报警
|
||||
|
HandAlarmSaveReqVO newAlarm = new HandAlarmSaveReqVO(); |
||||
|
newAlarm.setDetectorId(handVo.getId()); |
||||
|
newAlarm.setSn(handVo.getSn()); |
||||
|
newAlarm.setVAlarmFirst(handVo.getFirstValue()); |
||||
|
newAlarm.setGasType(handVo.getGasChemical()); |
||||
|
newAlarm.setPicX(handVo.getLongitude()); |
||||
|
newAlarm.setPicY(handVo.getLatitude()); |
||||
|
newAlarm.setAlarmType(AlarmType.GAS.getType()); |
||||
|
newAlarm.setAlarmLevel(newLevel); |
||||
|
newAlarm.setTAlarmStart(now); |
||||
|
newAlarm.setTenantId(Math.toIntExact(handVo.getTenantId())); |
||||
|
newAlarm.setUnit(handVo.getUnit()); |
||||
|
newAlarm.setCreator("system"); |
||||
|
newAlarm.setCreateTime(now); |
||||
|
|
||||
|
context.gasAlarmsToCreate.add(newAlarm); |
||||
|
context.pendingAlarmIds.put(handVo.getSn(), newAlarm); // 用于后续回填ID
|
||||
|
} |
||||
|
|
||||
|
handVo.setAlarmLevel(newLevel); |
||||
|
|
||||
|
// 报警升级
|
||||
|
if (handVo.getMaxAlarmLevel() == null || newLevel > handVo.getMaxAlarmLevel()) { |
||||
|
handVo.setMaxAlarmLevel(newLevel); |
||||
|
} |
||||
|
|
||||
|
// 更新最大值
|
||||
|
if (shouldUpdateMaxValue(handVo.getValue(), handVo.getMaxValue(), alarmRule.getDirection())) { |
||||
|
handVo.setMaxValue(handVo.getValue()); |
||||
|
|
||||
|
// 更新已存在的报警记录
|
||||
|
if (handVo.getAlarmId() != null) { |
||||
|
HandAlarmDO alarmToUpdate = new HandAlarmDO(); |
||||
|
alarmToUpdate.setId(handVo.getAlarmId()); |
||||
|
alarmToUpdate.setVAlarmMaximum(handVo.getMaxValue()); |
||||
|
alarmToUpdate.setAlarmLevel(newLevel); |
||||
|
alarmToUpdate.setUpdater("system"); |
||||
|
alarmToUpdate.setUpdateTime(now); |
||||
|
context.gasAlarmsToUpdate.add(alarmToUpdate); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
// 推送报警消息（值变化时）
|
||||
|
if (handVo.getLastPushValue() == null |
||||
|
|| !handVo.getLastPushValue().equals(handVo.getValue())) { |
||||
|
sendAlarmMessage(handVo, alarmRule.getGasTypeName(), handVo.getValue(), true, context); |
||||
|
handVo.setLastPushValue(handVo.getValue()); |
||||
|
} |
||||
|
|
||||
|
handVo.setGasStatus(HandAlarmType.ALARM.getType()); |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 电池报警处理 |
||||
|
*/ |
||||
|
private void processBatteryAlarm(HandDataVo handVo, BatchContext context) { |
||||
|
handVo.setBatteryStatus(EnableStatus.DISABLED.value()); |
||||
|
|
||||
|
if (handVo.getBatteryAlarmValue() == null) { |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
try { |
||||
|
int batteryPercentage = BatteryConverterUtils.getBatteryPercentage( |
||||
|
Integer.parseInt(handVo.getBattery()) |
||||
|
); |
||||
|
|
||||
|
int threshold = handVo.getBatteryAlarmValue().intValue(); |
||||
|
boolean isCurrentlyAlarming = EnableStatus.ENABLED.value().equals(handVo.getBatteryStatus()); |
||||
|
boolean shouldAlarm = batteryPercentage < threshold; |
||||
|
|
||||
|
LocalDateTime now = LocalDateTime.now(); |
||||
|
|
||||
|
// 从正常变为报警
|
||||
|
if (shouldAlarm && !isCurrentlyAlarming) { |
||||
|
HandAlarmSaveReqVO newAlarm = new HandAlarmSaveReqVO(); |
||||
|
newAlarm.setDetectorId(handVo.getId()); |
||||
|
newAlarm.setSn(handVo.getSn()); |
||||
|
newAlarm.setAlarmType(AlarmType.BATTERY.getType()); |
||||
|
newAlarm.setTAlarmStart(now); |
||||
|
newAlarm.setVAlarmFirst((double) batteryPercentage); |
||||
|
newAlarm.setPicX(handVo.getLatitude()); |
||||
|
newAlarm.setPicY(handVo.getLongitude()); |
||||
|
newAlarm.setTenantId(Math.toIntExact(handVo.getTenantId())); |
||||
|
newAlarm.setCreator("system"); |
||||
|
newAlarm.setCreateTime(now); |
||||
|
|
||||
|
context.gasAlarmsToCreate.add(newAlarm); |
||||
|
handVo.setBatteryStatus(EnableStatus.ENABLED.value()); |
||||
|
|
||||
|
} else if (!shouldAlarm && isCurrentlyAlarming && handVo.getBatteryStatusAlarmId() != null) { |
||||
|
// 从报警恢复正常
|
||||
|
HandAlarmDO alarmToEnd = new HandAlarmDO(); |
||||
|
alarmToEnd.setId(handVo.getBatteryStatusAlarmId()); |
||||
|
alarmToEnd.setTAlarmEnd(now); |
||||
|
alarmToEnd.setUpdater("system"); |
||||
|
alarmToEnd.setUpdateTime(now); |
||||
|
|
||||
|
context.gasAlarmsToUpdate.add(alarmToEnd); |
||||
|
handVo.setBatteryStatus(EnableStatus.DISABLED.value()); |
||||
|
handVo.setBatteryStatusAlarmId(null); |
||||
|
} |
||||
|
} catch (Exception e) { |
||||
|
log.error("[电池报警] 处理失败，SN: {}", handVo.getSn(), e); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 围栏报警处理 |
||||
|
*/ |
||||
|
private void processFenceAlarm(HandDataVo handVo, BatchContext context) { |
||||
|
if (StringUtils.isBlank(handVo.getFenceIds())) { |
||||
|
handVo.setFenceStatus(HandAlarmType.NORMAL.getType()); |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
List<Geofence> fenceList = context.fenceCache.get(handVo.getId()); |
||||
|
if (fenceList == null || fenceList.isEmpty()) { |
||||
|
handVo.setFenceStatus(HandAlarmType.NORMAL.getType()); |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
FenceType fenceType = FenceType.fromType(handVo.getFenceType()); |
||||
|
if (null == fenceType) { |
||||
|
log.error("[围栏报警] 围栏类型无效，SN: {}", handVo.getSn()); |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
boolean isInside = GeofenceUtils.isInsideAnyFence( |
||||
|
handVo.getLongitude(), |
||||
|
handVo.getLatitude(), |
||||
|
fenceList |
||||
|
); |
||||
|
|
||||
|
Boolean isViolating = switch (fenceType) { |
||||
|
case INSIDE -> isInside; |
||||
|
case OUTSIDE -> !isInside; |
||||
|
default -> null; |
||||
|
}; |
||||
|
|
||||
|
if (isViolating != null) { |
||||
|
handleFenceAlarmLifecycle(isViolating, handVo, fenceType, fenceList, context); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 处理围栏报警生命周期 |
||||
|
*/ |
||||
|
private void handleFenceAlarmLifecycle(boolean isViolating, HandDataVo handVo, |
||||
|
FenceType fenceType, List<Geofence> fenceList, BatchContext context) { |
||||
|
|
||||
|
boolean hasOngoingAlarm = handVo.getFenceAlarmId() != null; |
||||
|
LocalDateTime now = LocalDateTime.now(); |
||||
|
|
||||
|
if (isViolating) { |
||||
|
double distance = GeofenceUtils.fenceDistance(handVo, fenceType, fenceList); |
||||
|
double roundedDistance = Math.round(distance * 100.0) / 100.0; |
||||
|
|
||||
|
if (!hasOngoingAlarm) { |
||||
|
// 触发新报警
|
||||
|
FenceAlarmSaveReqVO newAlarm = new FenceAlarmSaveReqVO(); |
||||
|
newAlarm.setDetectorId(handVo.getId()); |
||||
|
newAlarm.setSn(handVo.getSn()); |
||||
|
newAlarm.setType(fenceType.getType()); |
||||
|
newAlarm.setTAlarmStart(now); |
||||
|
newAlarm.setPicX(handVo.getLongitude()); |
||||
|
newAlarm.setPicY(handVo.getLatitude()); |
||||
|
newAlarm.setDistance(roundedDistance); |
||||
|
newAlarm.setMaxDistance(roundedDistance); |
||||
|
newAlarm.setTenantId(handVo.getTenantId()); |
||||
|
newAlarm.setCreator("system"); |
||||
|
newAlarm.setCreateTime(now); |
||||
|
|
||||
|
context.fenceAlarmsToCreate.add(newAlarm); |
||||
|
context.pendingFenceAlarmIds.put(handVo.getSn(), newAlarm); |
||||
|
|
||||
|
handVo.setDistance(roundedDistance); |
||||
|
handVo.setMaxDistance(roundedDistance); |
||||
|
|
||||
|
} else { |
||||
|
// 更新持续报警
|
||||
|
handVo.setDistance(roundedDistance); |
||||
|
|
||||
|
FenceAlarmDO alarmToUpdate = new FenceAlarmDO(); |
||||
|
alarmToUpdate.setId(handVo.getFenceAlarmId()); |
||||
|
alarmToUpdate.setDistance(roundedDistance); |
||||
|
|
||||
|
if (handVo.getMaxDistance() == null || roundedDistance > handVo.getMaxDistance()) { |
||||
|
alarmToUpdate.setMaxDistance(roundedDistance); |
||||
|
handVo.setMaxDistance(roundedDistance); |
||||
|
} |
||||
|
|
||||
|
alarmToUpdate.setUpdater("system"); |
||||
|
alarmToUpdate.setUpdateTime(now); |
||||
|
context.fenceAlarmsToUpdate.add(alarmToUpdate); |
||||
|
} |
||||
|
|
||||
|
handVo.setFenceStatus(HandAlarmType.ALARM.getType()); |
||||
|
|
||||
|
} else { |
||||
|
// 结束报警
|
||||
|
if (hasOngoingAlarm) { |
||||
|
FenceAlarmDO alarmToEnd = new FenceAlarmDO(); |
||||
|
alarmToEnd.setId(handVo.getFenceAlarmId()); |
||||
|
alarmToEnd.setTAlarmEnd(now); |
||||
|
alarmToEnd.setStatus(EnableStatus.HANDLE.value()); |
||||
|
alarmToEnd.setUpdater("system"); |
||||
|
alarmToEnd.setUpdateTime(now); |
||||
|
|
||||
|
context.fenceAlarmsToUpdate.add(alarmToEnd); |
||||
|
|
||||
|
handVo.setFenceAlarmId(null); |
||||
|
handVo.setDistance(null); |
||||
|
handVo.setMaxDistance(null); |
||||
|
} |
||||
|
|
||||
|
handVo.setFenceStatus(HandAlarmType.NORMAL.getType()); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
// ========== 工具方法 ==========
|
||||
|
|
||||
|
/** |
||||
|
* 获取报警规则 |
||||
|
*/ |
||||
|
private AlarmRuleDO getAlarmRule(HandDataVo handVo, Map<Long, List<AlarmRuleDO>> ruleMap) { |
||||
|
if (handVo.getValue() == null || ruleMap == null) { |
||||
|
return null; |
||||
|
} |
||||
|
|
||||
|
double gasValue = handVo.getValue(); |
||||
|
Long gasTypeId = handVo.getGasTypeId(); |
||||
|
|
||||
|
List<AlarmRuleDO> rules = ruleMap.get(gasTypeId); |
||||
|
|
||||
|
// 兼容 Redis 反序列化的 String key
|
||||
|
if (rules == null && gasTypeId != null) { |
||||
|
rules = ruleMap.get(gasTypeId.toString()); |
||||
|
} |
||||
|
|
||||
|
if (rules == null || rules.isEmpty()) { |
||||
|
return null; |
||||
|
} |
||||
|
|
||||
|
AlarmRuleDO result = null; |
||||
|
for (AlarmRuleDO rule : rules) { |
||||
|
boolean inRange = (rule.getMin() == null || rule.getMin() <= gasValue) |
||||
|
&& (rule.getMax() == null || gasValue <= rule.getMax()); |
||||
|
|
||||
|
if (inRange) { |
||||
|
if (result == null || rule.getAlarmLevel() > result.getAlarmLevel()) { |
||||
|
result = rule; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
return result; |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 判断是否需要更新最大值 |
||||
|
*/ |
||||
|
private boolean shouldUpdateMaxValue(Double currentValue, Double maxValue, Integer direction) { |
||||
|
if (currentValue == null || maxValue == null || direction == null) { |
||||
|
return false; |
||||
|
} |
||||
|
|
||||
|
return (MaxDirection.DOWN.value().equals(direction) && currentValue < maxValue) |
||||
|
|| (MaxDirection.UP.value().equals(direction) && currentValue > maxValue); |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 发送报警消息（异步） |
||||
|
*/ |
||||
|
private void sendAlarmMessage(HandDataVo handVo, String gasName, Double value, |
||||
|
boolean isAlarming, BatchContext context) { |
||||
|
|
||||
|
String valueStr = (value != null && value % 1 == 0) |
||||
|
? String.valueOf(value.intValue()) |
||||
|
: String.valueOf(value); |
||||
|
|
||||
|
String statusText = isAlarming ? "报警" : "报警结束"; |
||||
|
String msgContent = String.format("%s%s，%s气体浓度为%s", |
||||
|
handVo.getName(), statusText, gasName, valueStr); |
||||
|
|
||||
|
// 记录报警消息日志
|
||||
|
AlarmMessageLog log = new AlarmMessageLog(); |
||||
|
log.setDetectorId(handVo.getId()); |
||||
|
log.setHolderName(handVo.getName()); |
||||
|
log.setSn(handVo.getSn()); |
||||
|
log.setDeptId(handVo.getDeptId()); |
||||
|
log.setTenantId(handVo.getTenantId()); |
||||
|
log.setMessage(msgContent); |
||||
|
log.setRemark("系统自动触发报警推送"); |
||||
|
|
||||
|
try { |
||||
|
List<String> targetSns = handDetectorService.getSnListByDept( |
||||
|
handVo.getDeptId(), |
||||
|
handVo.getTenantId(), |
||||
|
handVo.getSn() |
||||
|
); |
||||
|
|
||||
|
if (targetSns != null && !targetSns.isEmpty()) { |
||||
|
log.setPushSnList(String.join(",", targetSns)); |
||||
|
|
||||
|
// 异步推送 MQTT 消息
|
||||
|
publishAlarmToMqtt(targetSns, msgContent); |
||||
|
} |
||||
|
} catch (Exception e) { |
||||
|
this.log.error("[报警推送] 准备推送数据失败", e); |
||||
|
} |
||||
|
|
||||
|
context.alarmMessageLogs.add(log); |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 异步推送 MQTT 报警消息 |
||||
|
*/ |
||||
|
private void publishAlarmToMqtt(List<String> targetSns, String message) { |
||||
|
alarmExecutor.execute(() -> { |
||||
|
Map<String, String> payload = Map.of("message", message); |
||||
|
|
||||
|
try { |
||||
|
String json = OBJECT_MAPPER.writeValueAsString(payload); |
||||
|
|
||||
|
for (String sn : targetSns) { |
||||
|
if (StringUtils.isBlank(sn)) continue; |
||||
|
|
||||
|
try { |
||||
|
String topic = sn + "/zds_down"; |
||||
|
client.publish(topic, json); |
||||
|
} catch (Exception e) { |
||||
|
log.error("[MQTT推送] 失败，SN: {}", sn, e); |
||||
|
} |
||||
|
} |
||||
|
} catch (JsonProcessingException e) { |
||||
|
log.error("[MQTT推送] JSON序列化失败", e); |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 批量更新 Redis |
||||
|
*/ |
||||
|
private void batchUpdateRedis(Map<String, HandDataVo> updates, Map<String, Long> snToTenantMap) { |
||||
|
// 按租户分组
|
||||
|
Map<Long, Map<String, HandDataVo>> updatesByTenant = updates.entrySet().stream() |
||||
|
.filter(e -> snToTenantMap.containsKey(e.getKey())) |
||||
|
.collect(Collectors.groupingBy( |
||||
|
e -> snToTenantMap.get(e.getKey()), |
||||
|
Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue) |
||||
|
)); |
||||
|
|
||||
|
// 按租户批量更新
|
||||
|
updatesByTenant.forEach((tenantId, deviceMap) -> { |
||||
|
try { |
||||
|
handDetectorService.batchUpdateRedisData(tenantId, deviceMap); |
||||
|
} catch (Exception e) { |
||||
|
log.error("[Redis更新] 失败，tenantId: {}", tenantId, e); |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 创建原始日志对象 |
||||
|
*/ |
||||
|
private HandOriginalLog createOriginalLog(String sn, String payload, Long tenantId) { |
||||
|
HandOriginalLog log = new HandOriginalLog(); |
||||
|
log.setSn(sn); |
||||
|
log.setPayload(payload); |
||||
|
log.setTenantId(tenantId); |
||||
|
log.setTs(new Timestamp(System.currentTimeMillis())); |
||||
|
return log; |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 创建处理后的日志对象 |
||||
|
*/ |
||||
|
private TdengineDataVo createTdengineDataVo(HandDataVo handVo) { |
||||
|
TdengineDataVo vo = BeanUtils.toBean(handVo, TdengineDataVo.class); |
||||
|
vo.setTenantId(handVo.getTenantId()); |
||||
|
vo.setTs(new Timestamp(System.currentTimeMillis())); |
||||
|
return vo; |
||||
|
} |
||||
|
|
||||
|
// ========== 内部类：批处理上下文 ==========
|
||||
|
|
||||
|
/** |
||||
|
* 批处理上下文，存储整个批次的数据 |
||||
|
*/ |
||||
|
private static class BatchContext { |
||||
|
// 基础数据
|
||||
|
Map<String, Long> snToTenantMap = new HashMap<>(); |
||||
|
Map<String, HandDataVo> snToDeviceMap = new HashMap<>(); |
||||
|
Map<Long, Map<Long, List<AlarmRuleDO>>> tenantAlarmRules = new HashMap<>(); |
||||
|
Map<Long, List<Geofence>> fenceCache = new HashMap<>(); |
||||
|
|
||||
|
// 待保存的日志
|
||||
|
List<HandOriginalLog> originalLogs = new ArrayList<>(); |
||||
|
List<TdengineDataVo> processedLogs = new ArrayList<>(); |
||||
|
List<AlarmMessageLog> alarmMessageLogs = new ArrayList<>(); |
||||
|
|
||||
|
// 待保存的报警
|
||||
|
List<HandAlarmSaveReqVO> gasAlarmsToCreate = new ArrayList<>(); |
||||
|
List<HandAlarmDO> gasAlarmsToUpdate = new ArrayList<>(); |
||||
|
List<FenceAlarmSaveReqVO> fenceAlarmsToCreate = new ArrayList<>(); |
||||
|
List<FenceAlarmDO> fenceAlarmsToUpdate = new ArrayList<>(); |
||||
|
|
||||
|
// 待回填的ID（用于新创建的报警记录）
|
||||
|
Map<String, HandAlarmSaveReqVO> pendingAlarmIds = new HashMap<>(); |
||||
|
Map<String, FenceAlarmSaveReqVO> pendingFenceAlarmIds = new HashMap<>(); |
||||
|
|
||||
|
// 待更新的 Redis 数据
|
||||
|
Map<String, HandDataVo> redisUpdates = new HashMap<>(); |
||||
|
|
||||
|
boolean isEmpty() { |
||||
|
return snToDeviceMap.isEmpty(); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -1,107 +0,0 @@ |
|||||
package cn.iocoder.yudao.module.hand.controller.admin; |
|
||||
|
|
||||
import cn.iocoder.yudao.module.hand.dal.AlarmMessageDO; |
|
||||
import cn.iocoder.yudao.module.hand.service.AlarmMessageService; |
|
||||
import cn.iocoder.yudao.module.hand.vo.AlarmMessagePageReqVO; |
|
||||
import cn.iocoder.yudao.module.hand.vo.AlarmMessageRespVO; |
|
||||
import cn.iocoder.yudao.module.hand.vo.AlarmMessageSaveReqVO; |
|
||||
import org.springframework.web.bind.annotation.*; |
|
||||
import jakarta.annotation.Resource; |
|
||||
import org.springframework.validation.annotation.Validated; |
|
||||
import org.springframework.security.access.prepost.PreAuthorize; |
|
||||
import io.swagger.v3.oas.annotations.tags.Tag; |
|
||||
import io.swagger.v3.oas.annotations.Parameter; |
|
||||
import io.swagger.v3.oas.annotations.Operation; |
|
||||
|
|
||||
import jakarta.validation.constraints.*; |
|
||||
import jakarta.validation.*; |
|
||||
import jakarta.servlet.http.*; |
|
||||
import java.util.*; |
|
||||
import java.io.IOException; |
|
||||
|
|
||||
import cn.iocoder.yudao.framework.common.pojo.PageParam; |
|
||||
import cn.iocoder.yudao.framework.common.pojo.PageResult; |
|
||||
import cn.iocoder.yudao.framework.common.pojo.CommonResult; |
|
||||
import cn.iocoder.yudao.framework.common.util.object.BeanUtils; |
|
||||
import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success; |
|
||||
|
|
||||
import cn.iocoder.yudao.framework.excel.core.util.ExcelUtils; |
|
||||
|
|
||||
import cn.iocoder.yudao.framework.apilog.core.annotation.ApiAccessLog; |
|
||||
import static cn.iocoder.yudao.framework.apilog.core.enums.OperateTypeEnum.*; |
|
||||
|
|
||||
|
|
||||
|
|
||||
@Tag(name = "管理后台 - GAS手持探测器推送") |
|
||||
@RestController |
|
||||
@RequestMapping("/gas/alarm-message") |
|
||||
@Validated |
|
||||
public class AlarmMessageController { |
|
||||
|
|
||||
@Resource |
|
||||
private AlarmMessageService alarmMessageService; |
|
||||
|
|
||||
@PostMapping("/create") |
|
||||
@Operation(summary = "创建GAS手持探测器推送") |
|
||||
@PreAuthorize("@ss.hasPermission('gas:alarm-message:create')") |
|
||||
public CommonResult<Long> createAlarmMessage(@Valid @RequestBody AlarmMessageSaveReqVO createReqVO) { |
|
||||
return success(alarmMessageService.createAlarmMessage(createReqVO)); |
|
||||
} |
|
||||
|
|
||||
@PutMapping("/update") |
|
||||
@Operation(summary = "更新GAS手持探测器推送") |
|
||||
@PreAuthorize("@ss.hasPermission('gas:alarm-message:update')") |
|
||||
public CommonResult<Boolean> updateAlarmMessage(@Valid @RequestBody AlarmMessageSaveReqVO updateReqVO) { |
|
||||
alarmMessageService.updateAlarmMessage(updateReqVO); |
|
||||
return success(true); |
|
||||
} |
|
||||
|
|
||||
@DeleteMapping("/delete") |
|
||||
@Operation(summary = "删除GAS手持探测器推送") |
|
||||
@Parameter(name = "id", description = "编号", required = true) |
|
||||
@PreAuthorize("@ss.hasPermission('gas:alarm-message:delete')") |
|
||||
public CommonResult<Boolean> deleteAlarmMessage(@RequestParam("id") Long id) { |
|
||||
alarmMessageService.deleteAlarmMessage(id); |
|
||||
return success(true); |
|
||||
} |
|
||||
|
|
||||
@DeleteMapping("/delete-list") |
|
||||
@Parameter(name = "ids", description = "编号", required = true) |
|
||||
@Operation(summary = "批量删除GAS手持探测器推送") |
|
||||
@PreAuthorize("@ss.hasPermission('gas:alarm-message:delete')") |
|
||||
public CommonResult<Boolean> deleteAlarmMessageList(@RequestParam("ids") List<Long> ids) { |
|
||||
alarmMessageService.deleteAlarmMessageListByIds(ids); |
|
||||
return success(true); |
|
||||
} |
|
||||
|
|
||||
@GetMapping("/get") |
|
||||
@Operation(summary = "获得GAS手持探测器推送") |
|
||||
@Parameter(name = "id", description = "编号", required = true, example = "1024") |
|
||||
@PreAuthorize("@ss.hasPermission('gas:alarm-message:query')") |
|
||||
public CommonResult<AlarmMessageRespVO> getAlarmMessage(@RequestParam("id") Long id) { |
|
||||
AlarmMessageDO alarmMessage = alarmMessageService.getAlarmMessage(id); |
|
||||
return success(BeanUtils.toBean(alarmMessage, AlarmMessageRespVO.class)); |
|
||||
} |
|
||||
|
|
||||
@GetMapping("/page") |
|
||||
@Operation(summary = "获得GAS手持探测器推送分页") |
|
||||
@PreAuthorize("@ss.hasPermission('gas:alarm-message:query')") |
|
||||
public CommonResult<PageResult<AlarmMessageRespVO>> getAlarmMessagePage(@Valid AlarmMessagePageReqVO pageReqVO) { |
|
||||
PageResult<AlarmMessageDO> pageResult = alarmMessageService.getAlarmMessagePage(pageReqVO); |
|
||||
return success(BeanUtils.toBean(pageResult, AlarmMessageRespVO.class)); |
|
||||
} |
|
||||
|
|
||||
@GetMapping("/export-excel") |
|
||||
@Operation(summary = "导出GAS手持探测器推送 Excel") |
|
||||
@PreAuthorize("@ss.hasPermission('gas:alarm-message:export')") |
|
||||
@ApiAccessLog(operateType = EXPORT) |
|
||||
public void exportAlarmMessageExcel(@Valid AlarmMessagePageReqVO pageReqVO, |
|
||||
HttpServletResponse response) throws IOException { |
|
||||
pageReqVO.setPageSize(PageParam.PAGE_SIZE_NONE); |
|
||||
List<AlarmMessageDO> list = alarmMessageService.getAlarmMessagePage(pageReqVO).getList(); |
|
||||
// 导出 Excel
|
|
||||
ExcelUtils.write(response, "GAS手持探测器推送.xls", "数据", AlarmMessageRespVO.class, |
|
||||
BeanUtils.toBean(list, AlarmMessageRespVO.class)); |
|
||||
} |
|
||||
|
|
||||
} |
|
||||
@ -1,33 +0,0 @@ |
|||||
package cn.iocoder.yudao.module.hand.mapper; |
|
||||
|
|
||||
import java.util.*; |
|
||||
|
|
||||
import cn.iocoder.yudao.framework.common.pojo.PageResult; |
|
||||
import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX; |
|
||||
import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX; |
|
||||
import cn.iocoder.yudao.module.hand.dal.AlarmMessageDO; |
|
||||
import cn.iocoder.yudao.module.hand.vo.AlarmMessagePageReqVO; |
|
||||
import org.apache.ibatis.annotations.Mapper; |
|
||||
|
|
||||
/** |
|
||||
* GAS手持探测器推送 Mapper |
|
||||
* |
|
||||
* @author 超级管理员 |
|
||||
*/ |
|
||||
@Mapper |
|
||||
public interface AlarmMessageMapper extends BaseMapperX<AlarmMessageDO> { |
|
||||
|
|
||||
default PageResult<AlarmMessageDO> selectPage(AlarmMessagePageReqVO reqVO) { |
|
||||
return selectPage(reqVO, new LambdaQueryWrapperX<AlarmMessageDO>() |
|
||||
.eqIfPresent(AlarmMessageDO::getDetectorId, reqVO.getDetectorId()) |
|
||||
.likeIfPresent(AlarmMessageDO::getName, reqVO.getName()) |
|
||||
.eqIfPresent(AlarmMessageDO::getSn, reqVO.getSn()) |
|
||||
.eqIfPresent(AlarmMessageDO::getMessage, reqVO.getMessage()) |
|
||||
.eqIfPresent(AlarmMessageDO::getPushSnList, reqVO.getPushSnList()) |
|
||||
.eqIfPresent(AlarmMessageDO::getRemark, reqVO.getRemark()) |
|
||||
.eqIfPresent(AlarmMessageDO::getDeptId, reqVO.getDeptId()) |
|
||||
.betweenIfPresent(AlarmMessageDO::getCreateTime, reqVO.getCreateTime()) |
|
||||
.orderByDesc(AlarmMessageDO::getId)); |
|
||||
} |
|
||||
|
|
||||
} |
|
||||
@ -1,67 +0,0 @@ |
|||||
package cn.iocoder.yudao.module.hand.service; |
|
||||
|
|
||||
import java.util.*; |
|
||||
|
|
||||
import cn.iocoder.yudao.module.hand.dal.AlarmMessageDO; |
|
||||
import cn.iocoder.yudao.module.hand.dal.HandDetectorDO; |
|
||||
import cn.iocoder.yudao.module.hand.vo.AlarmMessagePageReqVO; |
|
||||
import cn.iocoder.yudao.module.hand.vo.AlarmMessageSaveReqVO; |
|
||||
import cn.iocoder.yudao.module.hand.vo.HandDataVo; |
|
||||
import jakarta.validation.*; |
|
||||
import cn.iocoder.yudao.framework.common.pojo.PageResult; |
|
||||
import cn.iocoder.yudao.framework.common.pojo.PageParam; |
|
||||
|
|
||||
/** |
|
||||
* GAS手持探测器推送 Service 接口 |
|
||||
* |
|
||||
* @author 超级管理员 |
|
||||
*/ |
|
||||
public interface AlarmMessageService { |
|
||||
|
|
||||
/** |
|
||||
* 创建GAS手持探测器推送 |
|
||||
* |
|
||||
* @param createReqVO 创建信息 |
|
||||
* @return 编号 |
|
||||
*/ |
|
||||
Long createAlarmMessage(@Valid AlarmMessageSaveReqVO createReqVO); |
|
||||
|
|
||||
/** |
|
||||
* 更新GAS手持探测器推送 |
|
||||
* |
|
||||
* @param updateReqVO 更新信息 |
|
||||
*/ |
|
||||
void updateAlarmMessage(@Valid AlarmMessageSaveReqVO updateReqVO); |
|
||||
|
|
||||
/** |
|
||||
* 删除GAS手持探测器推送 |
|
||||
* |
|
||||
* @param id 编号 |
|
||||
*/ |
|
||||
void deleteAlarmMessage(Long id); |
|
||||
|
|
||||
/** |
|
||||
* 批量删除GAS手持探测器推送 |
|
||||
* |
|
||||
* @param ids 编号 |
|
||||
*/ |
|
||||
void deleteAlarmMessageListByIds(List<Long> ids); |
|
||||
|
|
||||
/** |
|
||||
* 获得GAS手持探测器推送 |
|
||||
* |
|
||||
* @param id 编号 |
|
||||
* @return GAS手持探测器推送 |
|
||||
*/ |
|
||||
AlarmMessageDO getAlarmMessage(Long id); |
|
||||
|
|
||||
/** |
|
||||
* 获得GAS手持探测器推送分页 |
|
||||
* |
|
||||
* @param pageReqVO 分页查询 |
|
||||
* @return GAS手持探测器推送分页 |
|
||||
*/ |
|
||||
PageResult<AlarmMessageDO> getAlarmMessagePage(AlarmMessagePageReqVO pageReqVO); |
|
||||
|
|
||||
void createAlarmRecord(HandDataVo redisData, List<HandDetectorDO> listAll, String msgContent); |
|
||||
} |
|
||||
@ -1,117 +0,0 @@ |
|||||
package cn.iocoder.yudao.module.hand.service.impl; |
|
||||
|
|
||||
import cn.hutool.core.collection.CollUtil; |
|
||||
import cn.hutool.core.util.StrUtil; |
|
||||
import cn.iocoder.yudao.module.hand.dal.AlarmMessageDO; |
|
||||
import cn.iocoder.yudao.module.hand.dal.HandDetectorDO; |
|
||||
import cn.iocoder.yudao.module.hand.mapper.AlarmMessageMapper; |
|
||||
import cn.iocoder.yudao.module.hand.service.AlarmMessageService; |
|
||||
import cn.iocoder.yudao.module.hand.vo.AlarmMessagePageReqVO; |
|
||||
import cn.iocoder.yudao.module.hand.vo.AlarmMessageSaveReqVO; |
|
||||
import cn.iocoder.yudao.module.hand.vo.HandDataVo; |
|
||||
import org.springframework.stereotype.Service; |
|
||||
import jakarta.annotation.Resource; |
|
||||
import org.springframework.validation.annotation.Validated; |
|
||||
import org.springframework.transaction.annotation.Transactional; |
|
||||
|
|
||||
import java.util.*; |
|
||||
import java.util.stream.Collectors; |
|
||||
|
|
||||
import cn.iocoder.yudao.framework.common.pojo.PageResult; |
|
||||
import cn.iocoder.yudao.framework.common.pojo.PageParam; |
|
||||
import cn.iocoder.yudao.framework.common.util.object.BeanUtils; |
|
||||
|
|
||||
|
|
||||
import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception; |
|
||||
import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList; |
|
||||
import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.diffList; |
|
||||
import static cn.iocoder.yudao.module.hand.enums.ErrorCodeConstants.ALARM_MESSAGE_NOT_EXISTS; |
|
||||
|
|
||||
/** |
|
||||
* GAS手持探测器推送 Service 实现类 |
|
||||
* |
|
||||
* @author 超级管理员 |
|
||||
*/ |
|
||||
@Service |
|
||||
@Validated |
|
||||
public class AlarmMessageServiceImpl implements AlarmMessageService { |
|
||||
|
|
||||
@Resource |
|
||||
private AlarmMessageMapper alarmMessageMapper; |
|
||||
|
|
||||
@Override |
|
||||
public Long createAlarmMessage(AlarmMessageSaveReqVO createReqVO) { |
|
||||
// 插入
|
|
||||
AlarmMessageDO alarmMessage = BeanUtils.toBean(createReqVO, AlarmMessageDO.class); |
|
||||
alarmMessageMapper.insert(alarmMessage); |
|
||||
|
|
||||
// 返回
|
|
||||
return alarmMessage.getId(); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public void updateAlarmMessage(AlarmMessageSaveReqVO updateReqVO) { |
|
||||
// 校验存在
|
|
||||
validateAlarmMessageExists(updateReqVO.getId()); |
|
||||
// 更新
|
|
||||
AlarmMessageDO updateObj = BeanUtils.toBean(updateReqVO, AlarmMessageDO.class); |
|
||||
alarmMessageMapper.updateById(updateObj); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public void deleteAlarmMessage(Long id) { |
|
||||
// 校验存在
|
|
||||
validateAlarmMessageExists(id); |
|
||||
// 删除
|
|
||||
alarmMessageMapper.deleteById(id); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public void deleteAlarmMessageListByIds(List<Long> ids) { |
|
||||
// 删除
|
|
||||
alarmMessageMapper.deleteByIds(ids); |
|
||||
} |
|
||||
|
|
||||
|
|
||||
private void validateAlarmMessageExists(Long id) { |
|
||||
if (alarmMessageMapper.selectById(id) == null) { |
|
||||
throw exception(ALARM_MESSAGE_NOT_EXISTS); |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public AlarmMessageDO getAlarmMessage(Long id) { |
|
||||
return alarmMessageMapper.selectById(id); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public PageResult<AlarmMessageDO> getAlarmMessagePage(AlarmMessagePageReqVO pageReqVO) { |
|
||||
return alarmMessageMapper.selectPage(pageReqVO); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
@Transactional(rollbackFor = Exception.class) |
|
||||
public void createAlarmRecord(HandDataVo redisData, List<HandDetectorDO> listAll, String msgContent) { |
|
||||
String pushSnListStr = ""; |
|
||||
if (CollUtil.isNotEmpty(listAll)) { |
|
||||
pushSnListStr = listAll.stream() |
|
||||
.map(HandDetectorDO::getSn) |
|
||||
.filter(StrUtil::isNotBlank) |
|
||||
.collect(Collectors.joining(",")); |
|
||||
} |
|
||||
// 2. 构建实体对象
|
|
||||
AlarmMessageDO alarmDO = new AlarmMessageDO(); |
|
||||
alarmDO.setDetectorId(redisData.getId()); |
|
||||
alarmDO.setName(redisData.getName()); |
|
||||
alarmDO.setSn(redisData.getSn()); |
|
||||
alarmDO.setTenantId(redisData.getTenantId()); |
|
||||
alarmDO.setDeptId(redisData.getDeptId()); |
|
||||
alarmDO.setMessage(msgContent); |
|
||||
alarmDO.setPushSnList(pushSnListStr); |
|
||||
alarmDO.setRemark("系统自动触发报警推送"); |
|
||||
|
|
||||
// 3. 落库
|
|
||||
alarmMessageMapper.insert(alarmDO); |
|
||||
} |
|
||||
|
|
||||
} |
|
||||
@ -1,41 +0,0 @@ |
|||||
package cn.iocoder.yudao.module.hand.vo; |
|
||||
|
|
||||
import lombok.*; |
|
||||
import java.util.*; |
|
||||
import io.swagger.v3.oas.annotations.media.Schema; |
|
||||
import cn.iocoder.yudao.framework.common.pojo.PageParam; |
|
||||
import org.springframework.format.annotation.DateTimeFormat; |
|
||||
import java.time.LocalDateTime; |
|
||||
|
|
||||
import static cn.iocoder.yudao.framework.common.util.date.DateUtils.FORMAT_YEAR_MONTH_DAY_HOUR_MINUTE_SECOND; |
|
||||
|
|
||||
@Schema(description = "管理后台 - GAS手持探测器推送分页 Request VO") |
|
||||
@Data |
|
||||
public class AlarmMessagePageReqVO extends PageParam { |
|
||||
|
|
||||
@Schema(description = "手持表id", example = "10665") |
|
||||
private Long detectorId; |
|
||||
|
|
||||
@Schema(description = "持有人名称", example = "王五") |
|
||||
private String name; |
|
||||
|
|
||||
@Schema(description = "设备编号") |
|
||||
private String sn; |
|
||||
|
|
||||
@Schema(description = "消息") |
|
||||
private String message; |
|
||||
|
|
||||
@Schema(description = "推送设备sn，逗号分割") |
|
||||
private String pushSnList; |
|
||||
|
|
||||
@Schema(description = "备注", example = "随便") |
|
||||
private String remark; |
|
||||
|
|
||||
@Schema(description = "部门id", example = "12286") |
|
||||
private Long deptId; |
|
||||
|
|
||||
@Schema(description = "创建时间") |
|
||||
@DateTimeFormat(pattern = FORMAT_YEAR_MONTH_DAY_HOUR_MINUTE_SECOND) |
|
||||
private LocalDateTime[] createTime; |
|
||||
|
|
||||
} |
|
||||
@ -1,51 +0,0 @@ |
|||||
package cn.iocoder.yudao.module.hand.vo; |
|
||||
|
|
||||
import io.swagger.v3.oas.annotations.media.Schema; |
|
||||
import lombok.*; |
|
||||
import java.util.*; |
|
||||
import org.springframework.format.annotation.DateTimeFormat; |
|
||||
import java.time.LocalDateTime; |
|
||||
import com.alibaba.excel.annotation.*; |
|
||||
|
|
||||
@Schema(description = "管理后台 - GAS手持探测器推送 Response VO") |
|
||||
@Data |
|
||||
@ExcelIgnoreUnannotated |
|
||||
public class AlarmMessageRespVO { |
|
||||
|
|
||||
@Schema(description = "主键ID", requiredMode = Schema.RequiredMode.REQUIRED, example = "13733") |
|
||||
@ExcelProperty("主键ID") |
|
||||
private Long id; |
|
||||
|
|
||||
@Schema(description = "手持表id", example = "10665") |
|
||||
@ExcelProperty("手持表id") |
|
||||
private Long detectorId; |
|
||||
|
|
||||
@Schema(description = "持有人名称", example = "王五") |
|
||||
@ExcelProperty("持有人名称") |
|
||||
private String name; |
|
||||
|
|
||||
@Schema(description = "设备编号") |
|
||||
@ExcelProperty("设备编号") |
|
||||
private String sn; |
|
||||
|
|
||||
@Schema(description = "消息") |
|
||||
@ExcelProperty("消息") |
|
||||
private String message; |
|
||||
|
|
||||
@Schema(description = "推送设备sn，逗号分割") |
|
||||
@ExcelProperty("推送设备sn，逗号分割") |
|
||||
private String pushSnList; |
|
||||
|
|
||||
@Schema(description = "备注", example = "随便") |
|
||||
@ExcelProperty("备注") |
|
||||
private String remark; |
|
||||
|
|
||||
@Schema(description = "部门id", example = "12286") |
|
||||
@ExcelProperty("部门id") |
|
||||
private Long deptId; |
|
||||
|
|
||||
@Schema(description = "创建时间", requiredMode = Schema.RequiredMode.REQUIRED) |
|
||||
@ExcelProperty("创建时间") |
|
||||
private LocalDateTime createTime; |
|
||||
|
|
||||
} |
|
||||
@ -1,36 +0,0 @@ |
|||||
package cn.iocoder.yudao.module.hand.vo; |
|
||||
|
|
||||
import io.swagger.v3.oas.annotations.media.Schema; |
|
||||
import lombok.*; |
|
||||
import java.util.*; |
|
||||
import jakarta.validation.constraints.*; |
|
||||
|
|
||||
@Schema(description = "管理后台 - GAS手持探测器推送新增/修改 Request VO") |
|
||||
@Data |
|
||||
public class AlarmMessageSaveReqVO { |
|
||||
|
|
||||
@Schema(description = "主键ID", requiredMode = Schema.RequiredMode.REQUIRED, example = "13733") |
|
||||
private Long id; |
|
||||
|
|
||||
@Schema(description = "手持表id", example = "10665") |
|
||||
private Long detectorId; |
|
||||
|
|
||||
@Schema(description = "持有人名称", example = "王五") |
|
||||
private String name; |
|
||||
|
|
||||
@Schema(description = "设备编号") |
|
||||
private String sn; |
|
||||
|
|
||||
@Schema(description = "消息") |
|
||||
private String message; |
|
||||
|
|
||||
@Schema(description = "推送设备sn，逗号分割") |
|
||||
private String pushSnList; |
|
||||
|
|
||||
@Schema(description = "备注", example = "随便") |
|
||||
private String remark; |
|
||||
|
|
||||
@Schema(description = "部门id", example = "12286") |
|
||||
private Long deptId; |
|
||||
|
|
||||
} |
|
||||
@ -1,12 +0,0 @@ |
|||||
<?xml version="1.0" encoding="UTF-8"?> |
|
||||
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd"> |
|
||||
<mapper namespace="cn.iocoder.yudao.module.hand.mapper.AlarmMessageMapper"> |
|
||||
|
|
||||
<!-- |
|
||||
一般情况下，尽可能使用 Mapper 进行 CRUD 增删改查即可。 |
|
||||
无法满足的场景，例如说多表关联查询，才使用 XML 编写 SQL。 |
|
||||
代码生成器暂时只生成 Mapper XML 文件本身，更多推荐 MybatisX 快速开发插件来生成查询。 |
|
||||
文档可见：https://www.iocoder.cn/MyBatis/x-plugins/ |
|
||||
--> |
|
||||
|
|
||||
</mapper> |
|
||||
Loading…
Reference in new issue