root
/
agri
2
0
Fork 0
Code Issues Pull Requests Packages Projects 1 Releases Wiki Activity

Compare commits

base: root:a04d840f988fdaa60c78c3f1b7fa394722c1b6d1
Branches Tags
root:master
root:feasure
..
compare: root:6d62735dc4fa0566dfba85c91a0c795b916f1184
Branches Tags
root:master
root:feasure

No commits in common. "a04d840f988fdaa60c78c3f1b7fa394722c1b6d1" and "6d62735dc4fa0566dfba85c91a0c795b916f1184" have entirely different histories.
a04d840f98 ... 6d62735dc4

2 changed files with 109 additions and 208 deletions
Show Stats Expand all files Collapse all files

310
agri-quartz/src/main/java/com/agri/quartz/task/RollerAutoTask.java
View File

@ -1,8 +1,13 @@
package com.agri.quartz.task;
import com.agri.common.core.domain.entity.SysUser;
import com.agri.common.enums.TempCommandStatus;
import com.agri.common.utils.*;
import com.agri.common.utils.RollerTimeCalculator;
import com.agri.common.utils.TempJudgeUtil;
import com.agri.common.utils.TimeConvertUtil;
import com.agri.common.utils.TimeRangeUtil;
import com.agri.framework.config.MqttConfig;
import com.agri.framework.interceptor.FrontendControlHandler;
import com.agri.framework.manager.MqttAutoOffManager;
import com.agri.system.domain.SysAgriInfo;
import com.agri.system.domain.SysDevOperLog;
@ -11,6 +16,7 @@ import com.agri.system.service.ISysAgriInfoService;
import com.agri.system.service.ISysDevOperLogService;
import com.agri.system.service.ISysDtuDataService;
import com.agri.system.service.ISysRollerParamService;
import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.slf4j.Logger;
@ -26,6 +32,7 @@ import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
/**
@ -69,14 +76,13 @@ public class RollerAutoTask {
*/
private static final Logger log = LoggerFactory.getLogger(RollerAutoTask.class);
@Value("${spring.mqtt.dtu-ctl-lock-ttl}")
private int dtuCtlLockTTL;
public void checkAutoTerm() {
// 查询自动模式的大棚
List<SysAgriInfo> agriInfos = agriInfoService.lambdaQuery()
.select(SysAgriInfo::getImei, SysAgriInfo::getAgriName)
.select(SysAgriInfo::getImei)
.eq(SysAgriInfo::getWorkMode, 1)
.eq(SysAgriInfo::getIsDeleted, 0)
.list();
@ -108,262 +114,154 @@ public class RollerAutoTask {
// 查询每个IMEI今天的第一条日志
Map<String, Map<String, Integer>> todayLogCountByImeiMap = devOperLogService.getTodayLogCountByImeiMap(imeiList);
// 循环所有开启自动化的大棚
for (SysAgriInfo agriInfo : agriInfos) {
String imei = agriInfo.getImei();
String agriName = agriInfo.getAgriName();
for (String imei : imeiList) {
// 获取温湿度指定imei的数据
List<Map<String, Object>> dtuDataInfo = dtuDataByImeiMap.get(imei);
// 该大棚温湿度不存在
if (CollectionUtils.isEmpty(dtuDataInfo)) {
// todo 该大棚下1分钟内无最新温湿度，怀疑离线
continue;
break;
}
// 最后一条对应imei对应温度 及时间
// 获取今天对应imei的日志
Map<String, Integer> todayLogByRoller = todayLogCountByImeiMap.get(imei);
// 最后一条对应imei对应温度
Map<String, Object> dtuData = dtuDataInfo.get(0);
// 求温度上报时间
LocalDateTime dtuTime = TimeConvertUtil.strToLocalDateTimeSafe((String) dtuData.get("time"));
if (dtuTime == null) {
// todo 当前大棚温湿度时间为空 跳过
continue;
}
// 获取当前imei下的所有参数设置以及卷膜自动化条件设置
Map<String, List<RollerTermVO>> configTermByRollerMap = rollerTermMap.get(imei);
if (configTermByRollerMap.isEmpty()) {
// todo 当前大棚下没有设置条件或者参数
continue;
break;
}
// 获取今天对应imei的日志
Map<String, Integer> todayLogByRoller = todayLogCountByImeiMap.get(imei);
for (Map.Entry<String, List<RollerTermVO>> configEntry : configTermByRollerMap.entrySet()) {
String roller = configEntry.getKey(); // 当前卷膜
List<RollerTermVO> terms = configEntry.getValue(); // 当前卷膜条件
configTermByRollerMap.forEach((config, terms) ->{
// 每个卷膜分组只会有一个卷膜参数设置，所有取第一个即可
if (terms == null || terms.isEmpty()) {
// todo 当前卷膜 无参数设置，跳过当前roller
continue;
}
// 获取卷膜参数
RollerTermVO rollerConfig = terms.get(0);
String refTempCode = rollerConfig.getRefTempCode(); // 参考温度
// 整体参数
String refTempCode = rollerConfig.getRefTempCode();
String roller = rollerConfig.getRoller(); // 卷膜
BigDecimal ventTotalLen = rollerConfig.getVentTotalLen(); // 风口总长
BigDecimal reservedLen = rollerConfig.getReservedLen(); // 预留风口
// 防御性判断：避免dtuData.get(refTempCode)为null
Object tempObj = dtuData.get(refTempCode);
if (tempObj == null) {
// todo 当前卷膜参考温度设置为空
continue;
// 判断对应卷膜是否是今天第一次操作
boolean isFirstRun = true;
// todayLogByRoller为空铁定第一次操作，否则，就是todayLogByRoller.getRoller为空是第一次操作
if (!todayLogByRoller.isEmpty()) {
Integer logOfRoller = todayLogByRoller.getOrDefault(roller, 0);
if (logOfRoller>0) {
isFirstRun = false;
}
}
terms.forEach(term -> {
// 求温度上报时间
LocalDateTime dtuTime = TimeConvertUtil.strToLocalDateTimeSafe((String) dtuData.get("time"));
if (dtuTime == null) {
return; // 跳过该设备
}
// 优化：明确标注为当前roller的参考温度快照（仅解析一次）
BigDecimal currentTemp = new BigDecimal(tempObj.toString());
// 每个roller单独定义isFirstRun（作用域：当前roller）
boolean isFirstRun = todayLogByRoller.getOrDefault(roller, 0) == 0;
// 遍历当前roller的所有term（改用普通for循环，可读性更高）
for (RollerTermVO term : terms) {
// 判断该温度上报时间是否在该条件设置的时间范围内
boolean inRange = TimeRangeUtil.isTimeInRange(dtuTime, term.getStartTime(), term.getEndTime());
// 在范围内
if (inRange) {
//判断温度是否在适宜温度内
String redTempCode = dtuData.get(refTempCode).toString();
BigDecimal currentTemp = new BigDecimal(dtuData.get(redTempCode).toString());
TempCommandStatus tempCommandStatus = TempJudgeUtil.judgeTempCommand(currentTemp, term.getTemp());
if (tempCommandStatus == TempCommandStatus.OPEN) {
if (tempCommandStatus==TempCommandStatus.OPEN) {
// 开指令
sendOpenCommand(imei,agriName, roller, isFirstRun, term.getVent(), reservedLen);
sendOpenCommand(imei, roller, isFirstRun, term.getVent(),reservedLen);
isFirstRun = false;
} else if (tempCommandStatus == TempCommandStatus.CLOSE) {
} else if (tempCommandStatus==TempCommandStatus.CLOSE) {
// 关指令
sendCloseCommand(imei, agriName, roller, isFirstRun, term.getVent());
isFirstRun = false;
}
}
sendCloseCommand(imei, roller, isFirstRun, term.getVent());
}
}
// 不在掠过
});
});
}
}
/**
* 公共常量（可抽取到常量类）
*/
private static final String LOCK_PREFIX = "lock:";
private static final String AUTO_MODE = "auto_mode";
private static final String CREATE_BY = "条件控制";
private static final int OP_SOURCE = 3; // 操作来源：条件控制
private static final int LOCK_ACQUIRED = 1; // 是否获取锁
/**
* 发送开指令
* @param imei 设备唯一标识
* @param agriName 大棚名称
* @param roller 卷膜标识
* @param isFirstRun 是否第一次执行
* @param vent 每次执行的风口大小
* @param reservedLen 预留风口长度
*/
private void sendOpenCommand(String imei, String agriName, String roller, boolean isFirstRun,
private void sendOpenCommand(String imei, String roller, boolean isFirstRun,
BigDecimal vent, BigDecimal reservedLen) {
// ========== 1. 前置参数校验（防御性判断） ==========
validateBaseParams(imei, agriName, roller);
if (vent == null) {
log.error("【开指令】设备{}卷膜{}风口大小为空，指令发送失败", imei, roller);
return;
}
if (isFirstRun && reservedLen == null) {
log.error("【开指令】设备{}卷膜{}首次执行但预留风口为空，指令发送失败", imei, roller);
return;
}
// ========== 2. 计算开指令风口长度 ==========
try {
// 默认
BigDecimal openLen = vent;
// 是第一次，需要vent+预留风口
if (isFirstRun) {
log.info("【自动模式】设备【{}】卷膜【{}】今日首次执行开指令，叠加预留风口{}", imei, roller, reservedLen);
log.info("自动模式：设备【{}】开启自动模式。触发自动化条件，卷膜【{}】是今天第一次执行",imei,roller);
openLen = openLen.add(reservedLen);
}
String funcType = roller + "k1";
String message = buildMqttMessage(funcType);
// ========== 3. 执行核心指令逻辑 ==========
executeCommand(imei, agriName, roller, funcType, message, openLen, true);
String lockKey = "lock:" + imei + ":" + funcType;
Boolean lockSuccess = stringRedisTemplate.opsForValue().setIfAbsent(
lockKey, "auto_mode", dtuCtlLockTTL, TimeUnit.SECONDS // 延长至15秒，适配设备回执场景
);
if (lockSuccess == null || !lockSuccess) {
log.warn("【分布式锁】前端【自动模式】下触发【{}】操作设备【{}】的【{}】功能失败；可能其他用户正在操作此功能", roller, imei, funcType);
return;
}
// 3. 记录日志
log.info("【指令处理】前端{}于{}控制设备{}的{}功能，指令：{}",
clientId, LocalDateTime.now(), deviceId, funcType, payload);
SysUser sysUser = sysUserService.lambdaQuery()
.eq(SysUser::getClientId, clientId)
.one();
String operator = "手动控制";
if (sysUser!=null) {
operator = sysUser.getUserName();
}
SysAgriInfo agriInfo = sysAgriInfoService.lambdaQuery()
.eq(SysAgriInfo::getImei, deviceId)
.one();
String agriName = (agriInfo!=null && org.apache.commons.lang3.ObjectUtils.isNotEmpty(agriInfo.getAgriName()))?agriInfo.getAgriName():null;
SysDevOperLog logDto = new SysDevOperLog();
logDto.setAgriName(agriName);
logDto.setImei(deviceId);
logDto.setFuncCode(funcType);
logDto.setOpType(funcCodeMap.get(funcType));
logDto.setOpSource(1);
logDto.setPayload(payload);
logDto.setLockAcquired(1);
logDto.setLockHolder(clientId);
logDto.setCreateBy(operator);
sysDevOperLogService.save(logDto);
mqttMessageSender.publish("dtu/"+imei+"/down", "{\""+funcType+"\":1}");
// 获取运行时间
int runTime = RollerTimeCalculator.calculateRunTime(openLen);
if (runTime>0) {
autoOffManager.scheduleAutoOff(imei, roller+"k",runTime);
}
} catch (MqttException e) {
throw new RuntimeException(e);
}
}
/**
* 发送关指令
* @param imei 设备唯一标识
* @param agriName 大棚名称
* @param roller 卷膜标识
* @param isFirstRun 是否第一次执行
* @param vent 每次执行的风口大小
*/
private void sendCloseCommand(String imei, String agriName, String roller, boolean isFirstRun, BigDecimal vent) {
// ========== 1. 前置参数校验 ==========
validateBaseParams(imei, agriName, roller);
if (isFirstRun) {
log.info("【关指令】设备{}卷膜{}今日首次执行，直接返回不发送关指令", imei, roller);
return;
}
if (vent == null) {
log.error("【关指令】设备{}卷膜{}风口大小为空，指令发送失败", imei, roller);
return;
}
// ========== 2. 构建关指令参数 ==========
private void sendCloseCommand(String imei, String roller, boolean isFirstRun, BigDecimal vent) {
try {
// 如果是第一次直接返回
if (isFirstRun) return;
String funcType = roller + "g1";
String message = buildMqttMessage(funcType);
// ========== 3. 执行核心指令逻辑 ==========
executeCommand(imei, agriName, roller, funcType, message, vent, false);
}
// ========== 抽取公共方法：减少代码冗余 ==========
/**
* 基础参数校验
* @param imei 设备ID
* @param agriName 大棚名称
* @param roller 卷膜标识
*/
private void validateBaseParams(String imei, String agriName, String roller) {
if (StringUtils.isBlank(imei)) {
throw new IllegalArgumentException("设备IMEI不能为空");
}
if (StringUtils.isBlank(agriName)) {
log.warn("【指令】设备{}卷膜{}大棚名称为空，仍继续执行指令", imei, roller);
}
if (StringUtils.isBlank(roller)) {
throw new IllegalArgumentException("卷膜标识不能为空");
}
}
/**
* 构建MQTT指令消息
* @param funcType 功能类型（如roller1k1/roller1g1）
* @return 格式化的MQTT消息字符串
*/
private String buildMqttMessage(String funcType) {
return String.format("{\"%s\":1}", funcType);
}
/**
* 执行指令核心逻辑（分布式锁+日志记录+MQTT发布+自动关调度）
* @param imei 设备ID
* @param agriName 大棚名称
* @param roller 卷膜标识
* @param funcType 功能类型
* @param message MQTT消息
* @param len 风口长度（用于计算运行时间）
* @param isOpen 是否开指令
*/
private void executeCommand(String imei, String agriName, String roller, String funcType,
String message, BigDecimal len, boolean isOpen) {
String lockKey = LOCK_PREFIX + imei + ":" + funcType;
try {
// ========== 1. 获取分布式锁 ==========
Boolean lockSuccess = stringRedisTemplate.opsForValue().setIfAbsent(
lockKey, AUTO_MODE, dtuCtlLockTTL, TimeUnit.SECONDS
);
if (lockSuccess == null || !lockSuccess) {
log.warn("【分布式锁】自动模式下{}设备{}的{}功能失败，可能存在并发操作",
isOpen ? "开启" : "关闭", imei, funcType);
return;
}
// ========== 2. 记录操作日志 ==========
log.info("【指令处理】自动模式下触发{}设备{}的{}功能，指令：{}",
isOpen ? "开启" : "关闭", imei, funcType, message);
saveOperLog(imei, agriName, funcType, message, isOpen ? 1 : 0);
// ========== 3. 发布MQTT指令 ==========
mqttMessageSender.publish("dtu/" + imei + "/down", message);
// ========== 4. 计算运行时间并调度自动关 ==========
int runTime = RollerTimeCalculator.calculateRunTime(len);
mqttMessageSender.publish("dtu/"+imei+"/down", "{\""+funcType+"\":1}");
// 获取运行时间
int runTime = RollerTimeCalculator.calculateRunTime(vent);
if (runTime > 0) {
String autoOffKey = roller + (isOpen ? "k" : "g");
autoOffManager.scheduleAutoOff(imei, autoOffKey, runTime);
log.debug("【自动关调度】设备{}卷膜{}调度{}秒后自动关闭", imei, roller, runTime);
autoOffManager.scheduleAutoOff(imei, roller+"g",runTime);
}
} catch (MqttException e) {
// ========== 异常处理：记录详细日志，不抛运行时异常 ==========
log.error("【MQTT异常】{}设备{}的{}功能指令发布失败，指令：{}",
isOpen ? "开启" : "关闭", imei, funcType, message, e);
} catch (Exception e) {
// 兜底异常捕获：避免未知异常导致锁无法释放（依赖TTL兜底）
log.error("【指令执行异常】{}设备{}的{}功能执行失败",
isOpen ? "开启" : "关闭", imei, funcType, e);
}
}
/**
* 保存设备操作日志
* @param imei 设备ID
* @param agriName 大棚名称
* @param funcCode 功能码
* @param payload 指令内容
* @param opType 操作类型（1-开，0-关）
*/
private void saveOperLog(String imei, String agriName, String funcCode, String payload, int opType) {
SysDevOperLog logDto = new SysDevOperLog();
logDto.setAgriName(agriName);
logDto.setImei(imei);
logDto.setFuncCode(funcCode);
logDto.setOpType(opType);
logDto.setOpSource(OP_SOURCE);
logDto.setPayload(payload);
logDto.setLockAcquired(LOCK_ACQUIRED); // 已获取锁
logDto.setLockHolder(AUTO_MODE);
logDto.setCreateBy(CREATE_BY);
// 可选：增加异常捕获，避免日志保存失败影响指令执行
try {
devOperLogService.save(logDto);
} catch (Exception e) {
log.error("【日志保存失败】设备{}功能{}日志保存失败", imei, funcCode, e);
throw new RuntimeException(e);
}
}
}

7
agri-system/src/main/java/com/agri/system/service/impl/SysDevOperLogServiceImpl.java
View File

@ -3,7 +3,10 @@ package com.agri.system.service.impl;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import com.agri.common.utils.DateUtils;
@ -132,7 +135,7 @@ public class SysDevOperLogServiceImpl extends ServiceImpl<SysDevOperLogMapper, S
// 4.1 初始化所有固定IMEI的内层Map（避免取值时NPE）
for (String imei : imeiList) {
nestedCountMap.put(imei, Collections.emptyMap());
nestedCountMap.put(imei, new HashMap<>());
}
// 4.2 填充统计结果
for (Map<String, Object> row : sqlResult) {