Microservice/iot
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Compare commits

base: Microservice:6dcee1f6c38fd50ec50ab7ced80e781235f94507
Branches Tags
Microservice:master Microservice:wireless
Microservice:2026-05 Microservice:2026-04 Microservice:2026-03 Microservice:2026-02 Microservice:2026-01 Microservice:2025-10 Microservice:2025-09 Microservice:20250723 Microservice:2025-08 Microservice:2025-07 Microservice:2025-04 Microservice:2024-12 Microservice:2024-11 Microservice:2024-10 Microservice:2024-09 Microservice:2024-08 Microservice:2024-01
...
compare: Microservice:2026-04
Branches Tags
Microservice:master Microservice:wireless
Microservice:2026-05 Microservice:2026-04 Microservice:2026-03 Microservice:2026-02 Microservice:2026-01 Microservice:2025-10 Microservice:2025-09 Microservice:20250723 Microservice:2025-08 Microservice:2025-07 Microservice:2025-04 Microservice:2024-12 Microservice:2024-11 Microservice:2024-10 Microservice:2024-09 Microservice:2024-08 Microservice:2024-01

6 Commits
6dcee1f6c3 ... 2026-04

Author SHA1 Message Date
xy
0f532033b0 fix(mq): 修正消息消费成功后的Redis缓存时间 2026-05-07 13:29:08 +08:00
xy
b418855d02 refactor(realtime): 优化实时数据分析服务设备接入处理逻辑 
- 移除CldDevRunFlagConsumer中的注释代码，启用设备状态翻转功能
- 添加channelData2方法支持第二种设备类型的数据处理
- 在统计服务中新增通信记录功能，记录设备上线状态
 2026-04-29 13:57:22 +08:00
caozehui
9774172c0b 升级调整 2026-04-29 08:52:39 +08:00
xy
da75f74218 refactor(rt): 移除FloatUtils依赖并优化数据处理 
- 优化harmRealDataSet数据赋值方式，移除不必要的浮点数精度处理
 2026-04-24 09:42:26 +08:00
caozehui
7670793d5e SendMessageUtil 注入问题 2026-04-23 09:07:39 +08:00
caozehui
6f87784ddf 文件系统，设备升级 2026-04-23 08:58:13 +08:00
44 changed files with 2700 additions and 187 deletions
Expand all files Collapse all files

10
iot-access/access-api/src/main/java/com/njcn/access/enums/AccessResponseEnum.java
View File

@@ -75,7 +75,15 @@ public enum AccessResponseEnum {
FILE_CHECK_ERROR("A0312","文件校验码不一致!"), FILE_CHECK_ERROR("A0312","文件校验码不一致!"),
CLD_MODEL_EXIST("A0313","云前置模板已存在,请先删除再录入!"), CLD_MODEL_EXIST("A0313","云前置模板已存在,请先删除再录入!"),
;
/**
* A3001 ~ A3099 用于zlevent模块的枚举
* <p>
*/
FILE_DOWNLOAD_FAIL("A3002", "文件下载失败！"),
UNKNOWN_BUSINESS_TYPE_CODE("A3003", "未知业务type码"),
FILE_UPLOAD_FAIL("A3004", "文件上传失败！"),
TIME_OUT("A3005", "前置响应超时！"),;
private final String code; private final String code;

13
iot-access/access-api/src/main/java/com/njcn/access/enums/TypeEnum.java
View File

@@ -45,6 +45,19 @@ public enum TypeEnum {
TYPE_29("9217","设备心跳请求"), TYPE_29("9217","设备心跳请求"),
TYPE_30("4865","设备数据主动上送"), TYPE_30("4865","设备数据主动上送"),
TYPE_31("8503","设备控制命令"), TYPE_31("8503","设备控制命令"),
READ_FILE_DIR("1101", "读取文件目录"),
FILE_DOWNLOAD("1102", "文件下载"),
FIXED_VALUE("1103", "定值读取/写入"),
INNER_FIXED_VALUE("1104", "内部定值读取/写入"),
WORKING_LOG("1111","设备运行日志"),
DEVICE_VERSION("1112","设备版本信息"),
DEVICE_REBOOT("1114","设备重启"),
DEVICE_UPGRADE("1115","设备升级"),
FILE_UPLOAD("1116","文件上传"),
FILE_DELETE("1117","文件删除"),
MKDIR("1118","目录创建"),
DIR_DELETE("1119","目录删除"),
/** /**
* 数据类型 * 数据类型

29
iot-access/access-api/src/main/java/com/njcn/access/utils/SendMessageUtil.java
View File

@@ -1,7 +1,11 @@
package com.njcn.access.utils; package com.njcn.access.utils;
import com.alibaba.nacos.shaded.com.google.gson.Gson; import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.njcn.access.enums.AccessResponseEnum;
import com.njcn.access.pojo.dto.NoticeUserDto; import com.njcn.access.pojo.dto.NoticeUserDto;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.utils.RedisUtil;
import lombok.AllArgsConstructor; import lombok.AllArgsConstructor;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@@ -22,9 +26,9 @@ import java.nio.charset.StandardCharsets;
*/ */
@Component @Component
@Slf4j @Slf4j
@AllArgsConstructor
@RequiredArgsConstructor @RequiredArgsConstructor
public class SendMessageUtil { public class SendMessageUtil {
private final RedisUtil redisUtil;
@Value("${app.sendUrl:https://fc-mp-ff7b310f-94c9-4468-8260-109111c0a6b2.next.bspapp.com/push}") @Value("${app.sendUrl:https://fc-mp-ff7b310f-94c9-4468-8260-109111c0a6b2.next.bspapp.com/push}")
private String appSendUrl; private String appSendUrl;
@@ -67,4 +71,27 @@ public class SendMessageUtil {
e.getMessage(); e.getMessage();
} }
} }
/**
* 轮询 Redis 等待响应
*/
public String waitForResponse(String guid, int timeoutSeconds) {
long startTime = System.currentTimeMillis();
while (System.currentTimeMillis() - startTime < timeoutSeconds * 1000L) {
String response = redisUtil.getStringByKey(AppRedisKey.COMMON_RESOPNSE + guid);
if (response != null) {
redisUtil.delete(AppRedisKey.COMMON_REQUEST + guid);
redisUtil.delete(AppRedisKey.COMMON_RESOPNSE + guid);
return response;
}
try {
Thread.sleep(100); // 100ms 轮询一次
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
throw new BusinessException(AccessResponseEnum.TIME_OUT);
}
} }

4
iot-access/access-boot/pom.xml
View File

@@ -20,6 +20,10 @@
</properties> </properties>
<dependencies> <dependencies>
<dependency>
<groupId>com.github.tocrhz</groupId>
<artifactId>mqtt-spring-boot-starter</artifactId>
</dependency>
<dependency> <dependency>
<groupId>org.eclipse.paho</groupId> <groupId>org.eclipse.paho</groupId>

4
iot-analysis/analysis-rt/rt-boot/pom.xml
View File

@@ -21,6 +21,10 @@
</properties> </properties>
<dependencies> <dependencies>
<dependency>
<groupId>com.github.tocrhz</groupId>
<artifactId>mqtt-spring-boot-starter</artifactId>
</dependency>
<dependency> <dependency>
<groupId>com.njcn</groupId> <groupId>com.njcn</groupId>
<artifactId>common-web</artifactId> <artifactId>common-web</artifactId>

262
iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/service/impl/RtServiceImpl.java
View File

@@ -11,8 +11,10 @@ import com.njcn.common.utils.PubUtils;
import com.njcn.csdevice.api.CsLineFeignClient; import com.njcn.csdevice.api.CsLineFeignClient;
import com.njcn.csdevice.api.DataArrayFeignClient; import com.njcn.csdevice.api.DataArrayFeignClient;
import com.njcn.csdevice.api.DataSetFeignClient; import com.njcn.csdevice.api.DataSetFeignClient;
import com.njcn.csdevice.api.EquipmentFeignClient;
import com.njcn.csdevice.pojo.po.CsDataArray; import com.njcn.csdevice.pojo.po.CsDataArray;
import com.njcn.csdevice.pojo.po.CsDataSet; import com.njcn.csdevice.pojo.po.CsDataSet;
import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
import com.njcn.csdevice.pojo.po.CsLinePO; import com.njcn.csdevice.pojo.po.CsLinePO;
import com.njcn.mq.message.AppAutoDataMessage; import com.njcn.mq.message.AppAutoDataMessage;
import com.njcn.redis.utils.RedisUtil; import com.njcn.redis.utils.RedisUtil;
@@ -21,7 +23,6 @@ import com.njcn.rt.pojo.dto.BaseRealDataSet;
import com.njcn.rt.pojo.dto.HarmData; import com.njcn.rt.pojo.dto.HarmData;
import com.njcn.rt.pojo.dto.HarmRealDataSet; import com.njcn.rt.pojo.dto.HarmRealDataSet;
import com.njcn.rt.service.IRtService; import com.njcn.rt.service.IRtService;
import com.njcn.web.utils.FloatUtils;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@@ -51,6 +52,7 @@ public class RtServiceImpl implements IRtService {
private final ChannelObjectUtil channelObjectUtil; private final ChannelObjectUtil channelObjectUtil;
private final MqttPublisher publisher; private final MqttPublisher publisher;
private final RedisSetUtil redisSetUtil; private final RedisSetUtil redisSetUtil;
private final EquipmentFeignClient equipmentFeignClient;
@Override @Override
public void analysis(AppAutoDataMessage appAutoDataMessage) { public void analysis(AppAutoDataMessage appAutoDataMessage) {
@@ -74,12 +76,14 @@ public class RtServiceImpl implements IRtService {
} }
//根据dataArray解析数据 //根据dataArray解析数据
AppAutoDataMessage.DataArray item = appAutoDataMessage.getMsg().getDataArray().get(0); AppAutoDataMessage.DataArray item = appAutoDataMessage.getMsg().getDataArray().get(0);
//获取设备类型
CsEquipmentDeliveryPO po1 =equipmentFeignClient.getDevByLineId(lineId).getData();
//fixme 这边先根据数据集的名称来返回对应实体，这边感觉不太合适，后期有好方案再调整 //fixme 这边先根据数据集的名称来返回对应实体，这边感觉不太合适，后期有好方案再调整
//基础数据 //基础数据
if (dataSet.getName().contains("Ds$Pqd$Rt$Basic$")) { if (dataSet.getName().contains("Ds$Pqd$Rt$Basic$")) {
//用户Id //用户Id
String userId = redisUtil.getObjectByKey("rtDataUserId:"+lineId).toString(); String userId = redisUtil.getObjectByKey("rtDataUserId:"+lineId).toString();
BaseRealDataSet baseRealDataSet = assembleData(dataArrayList,item,po.getConType()); BaseRealDataSet baseRealDataSet = assembleData(dataArrayList,item,po.getConType(),po1.getDevAccessMethod());
baseRealDataSet.setUserId(userId); baseRealDataSet.setUserId(userId);
baseRealDataSet.setLineId(lineId); baseRealDataSet.setLineId(lineId);
baseRealDataSet.setPt(po.getPtRatio().floatValue()); baseRealDataSet.setPt(po.getPtRatio().floatValue());
@@ -94,7 +98,7 @@ public class RtServiceImpl implements IRtService {
if (ObjectUtil.isNotNull(redisObject)) { if (ObjectUtil.isNotNull(redisObject)) {
Set<String> userSet = redisSetUtil.convertToSet(redisObject); Set<String> userSet = redisSetUtil.convertToSet(redisObject);
userSet.forEach(userId->{ userSet.forEach(userId->{
BaseRealDataSet baseRealDataSet = assembleData(dataArrayList,item,po.getConType()); BaseRealDataSet baseRealDataSet = assembleData(dataArrayList,item,po.getConType(),po1.getDevAccessMethod());
baseRealDataSet.setUserId(userId); baseRealDataSet.setUserId(userId);
baseRealDataSet.setLineId(lineId); baseRealDataSet.setLineId(lineId);
baseRealDataSet.setPt(po.getPtRatio().floatValue()); baseRealDataSet.setPt(po.getPtRatio().floatValue());
@@ -102,7 +106,7 @@ public class RtServiceImpl implements IRtService {
baseRealDataSet.setDataLevel(dataSet.getDataLevel()); baseRealDataSet.setDataLevel(dataSet.getDataLevel());
long timestamp = item.getDataTimeSec(); long timestamp = item.getDataTimeSec();
baseRealDataSet.setDataTime(getTime(timestamp)); baseRealDataSet.setDataTime(getTime(timestamp));
publisher.send("/Web/RealData/" + userId, new Gson().toJson(baseRealDataSet), 1, false); publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false);
}); });
} }
} }
@@ -201,114 +205,194 @@ public class RtServiceImpl implements IRtService {
return dataMap; return dataMap;
} }
public BaseRealDataSet assembleData(List<CsDataArray> dataArrayList,AppAutoDataMessage.DataArray dataArray,Integer conType) { public BaseRealDataSet assembleData(List<CsDataArray> dataArrayList,AppAutoDataMessage.DataArray dataArray,Integer conType,String accessMethod) {
Map<String,Float> dataMap = getData(dataArrayList,dataArray); Map<String,Float> dataMap = getData(dataArrayList,dataArray);
if (Objects.equals("CLD",accessMethod)) {
return channelData(dataMap,conType); return channelData(dataMap,conType);
} else {
return channelData2(dataMap);
}
} }
public BaseRealDataSet channelData(Map<String,Float> map,Integer conType) { public BaseRealDataSet channelData(Map<String,Float> map,Integer conType) {
BaseRealDataSet baseRealDataSet = new BaseRealDataSet(); BaseRealDataSet baseRealDataSet = new BaseRealDataSet();
//频率 //频率
baseRealDataSet.setFreq(FloatUtils.get2Float(map.get("Pq_FreqM"))); baseRealDataSet.setFreq(map.get("Pq_FreqM"));
//频率偏差 //频率偏差
baseRealDataSet.setFreqDev(FloatUtils.get2Float(map.get("Pq_FreqDevM"))); baseRealDataSet.setFreqDev(map.get("Pq_FreqDevM"));
//判断监测点的接线方式，不同接线方式电压有效值取值不同 //判断监测点的接线方式，不同接线方式电压有效值取值不同
//星型-相电压 角形、V型-线电压 //星型-相电压 角形、V型-线电压
//电压有效值 //电压有效值
if (conType == 0) { if (conType == 0) {
baseRealDataSet.setVRmsA(FloatUtils.get2Float(map.get("Pq_RmsUA"))); baseRealDataSet.setVRmsA(map.get("Pq_RmsUA"));
baseRealDataSet.setVRmsB(FloatUtils.get2Float(map.get("Pq_RmsUB"))); baseRealDataSet.setVRmsB(map.get("Pq_RmsUB"));
baseRealDataSet.setVRmsC(FloatUtils.get2Float(map.get("Pq_RmsUC"))); baseRealDataSet.setVRmsC(map.get("Pq_RmsUC"));
} else { } else {
baseRealDataSet.setVRmsA(FloatUtils.get2Float(map.get("Pq_RmsLUAB"))); baseRealDataSet.setVRmsA(map.get("Pq_RmsLUAB"));
baseRealDataSet.setVRmsB(FloatUtils.get2Float(map.get("Pq_RmsLUBC"))); baseRealDataSet.setVRmsB(map.get("Pq_RmsLUBC"));
baseRealDataSet.setVRmsC(FloatUtils.get2Float(map.get("Pq_RmsLUCA"))); baseRealDataSet.setVRmsC(map.get("Pq_RmsLUCA"));
} }
//基波电压幅值 //基波电压幅值
if (conType == 0) { if (conType == 0) {
baseRealDataSet.setV1A(FloatUtils.get2Float(map.get("Pq_RmsFundUA"))); baseRealDataSet.setV1A(map.get("Pq_RmsFundUA"));
baseRealDataSet.setV1B(FloatUtils.get2Float(map.get("Pq_RmsFundUB"))); baseRealDataSet.setV1B(map.get("Pq_RmsFundUB"));
baseRealDataSet.setV1C(FloatUtils.get2Float(map.get("Pq_RmsFundUC"))); baseRealDataSet.setV1C(map.get("Pq_RmsFundUC"));
} else { } else {
baseRealDataSet.setV1A(FloatUtils.get2Float(map.get("Pq_RmsFundLUAB"))); baseRealDataSet.setV1A(map.get("Pq_RmsFundLUAB"));
baseRealDataSet.setV1B(FloatUtils.get2Float(map.get("Pq_RmsFundLUBC"))); baseRealDataSet.setV1B(map.get("Pq_RmsFundLUBC"));
baseRealDataSet.setV1C(FloatUtils.get2Float(map.get("Pq_RmsFundLUCA"))); baseRealDataSet.setV1C(map.get("Pq_RmsFundLUCA"));
} }
//电流有效值 //电流有效值
baseRealDataSet.setIRmsA(FloatUtils.get2Float(map.get("Pq_RmsIA"))); baseRealDataSet.setIRmsA(map.get("Pq_RmsIA"));
baseRealDataSet.setIRmsB(FloatUtils.get2Float(map.get("Pq_RmsIB"))); baseRealDataSet.setIRmsB(map.get("Pq_RmsIB"));
baseRealDataSet.setIRmsC(FloatUtils.get2Float(map.get("Pq_RmsIC"))); baseRealDataSet.setIRmsC(map.get("Pq_RmsIC"));
//基波电流幅值 //基波电流幅值
baseRealDataSet.setI1A(FloatUtils.get2Float(map.get("Pq_RmsFundIA"))); baseRealDataSet.setI1A(map.get("Pq_RmsFundIA"));
baseRealDataSet.setI1B(FloatUtils.get2Float(map.get("Pq_RmsFundIB"))); baseRealDataSet.setI1B(map.get("Pq_RmsFundIB"));
baseRealDataSet.setI1C(FloatUtils.get2Float(map.get("Pq_RmsFundIC"))); baseRealDataSet.setI1C(map.get("Pq_RmsFundIC"));
//电压偏差 //电压偏差
if (conType == 0) { if (conType == 0) {
baseRealDataSet.setVDevA(FloatUtils.get2Float(map.get("Pq_UDevA"))); baseRealDataSet.setVDevA(map.get("Pq_UDevA"));
baseRealDataSet.setVDevB(FloatUtils.get2Float(map.get("Pq_UDevB"))); baseRealDataSet.setVDevB(map.get("Pq_UDevB"));
baseRealDataSet.setVDevC(FloatUtils.get2Float(map.get("Pq_UDevC"))); baseRealDataSet.setVDevC(map.get("Pq_UDevC"));
} else { } else {
baseRealDataSet.setVDevA(FloatUtils.get2Float(map.get("Pq_LUDevAB"))); baseRealDataSet.setVDevA(map.get("Pq_LUDevAB"));
baseRealDataSet.setVDevB(FloatUtils.get2Float(map.get("Pq_LUDevBC"))); baseRealDataSet.setVDevB(map.get("Pq_LUDevBC"));
baseRealDataSet.setVDevC(FloatUtils.get2Float(map.get("Pq_LUDevCA"))); baseRealDataSet.setVDevC(map.get("Pq_LUDevCA"));
} }
//基波电压相位 //基波电压相位
if (conType == 0) { if (conType == 0) {
baseRealDataSet.setV1AngA(FloatUtils.get2Float(map.get("Pq_FundUAngA"))); baseRealDataSet.setV1AngA(map.get("Pq_FundUAngA"));
baseRealDataSet.setV1AngB(FloatUtils.get2Float(map.get("Pq_FundUAngB"))); baseRealDataSet.setV1AngB(map.get("Pq_FundUAngB"));
baseRealDataSet.setV1AngC(FloatUtils.get2Float(map.get("Pq_FundUAngC"))); baseRealDataSet.setV1AngC(map.get("Pq_FundUAngC"));
} else { } else {
baseRealDataSet.setV1AngA(FloatUtils.get2Float(map.get("Pq_FundLUAngAB"))); baseRealDataSet.setV1AngA(map.get("Pq_FundLUAngAB"));
baseRealDataSet.setV1AngB(FloatUtils.get2Float(map.get("Pq_FundLUAngBC"))); baseRealDataSet.setV1AngB(map.get("Pq_FundLUAngBC"));
baseRealDataSet.setV1AngC(FloatUtils.get2Float(map.get("Pq_FundLUAngCA"))); baseRealDataSet.setV1AngC(map.get("Pq_FundLUAngCA"));
} }
//基波电流相位 //基波电流相位
baseRealDataSet.setI1AngA(FloatUtils.get2Float(map.get("Pq_FundIAngA"))); baseRealDataSet.setI1AngA(map.get("Pq_FundIAngA"));
baseRealDataSet.setI1AngB(FloatUtils.get2Float(map.get("Pq_FundIAngB"))); baseRealDataSet.setI1AngB(map.get("Pq_FundIAngB"));
baseRealDataSet.setI1AngC(FloatUtils.get2Float(map.get("Pq_FundIAngC"))); baseRealDataSet.setI1AngC(map.get("Pq_FundIAngC"));
//电压总谐波畸变率 //电压总谐波畸变率
if (conType == 0) { if (conType == 0) {
baseRealDataSet.setVThdA(FloatUtils.get2Float(map.get("Pq_ThdUA"))); baseRealDataSet.setVThdA(map.get("Pq_ThdUA"));
baseRealDataSet.setVThdB(FloatUtils.get2Float(map.get("Pq_ThdUB"))); baseRealDataSet.setVThdB(map.get("Pq_ThdUB"));
baseRealDataSet.setVThdC(FloatUtils.get2Float(map.get("Pq_ThdUC"))); baseRealDataSet.setVThdC(map.get("Pq_ThdUC"));
} else { } else {
baseRealDataSet.setVThdA(FloatUtils.get2Float(map.get("Pq_ThdLUAB"))); baseRealDataSet.setVThdA(map.get("Pq_ThdLUAB"));
baseRealDataSet.setVThdB(FloatUtils.get2Float(map.get("Pq_ThdLUBC"))); baseRealDataSet.setVThdB(map.get("Pq_ThdLUBC"));
baseRealDataSet.setVThdC(FloatUtils.get2Float(map.get("Pq_ThdLUCA"))); baseRealDataSet.setVThdC(map.get("Pq_ThdLUCA"));
} }
//电流总谐波畸变率 //电流总谐波畸变率
baseRealDataSet.setIThdA(FloatUtils.get2Float(map.get("Pq_ThdIA"))); baseRealDataSet.setIThdA(map.get("Pq_ThdIA"));
baseRealDataSet.setIThdB(FloatUtils.get2Float(map.get("Pq_ThdIB"))); baseRealDataSet.setIThdB(map.get("Pq_ThdIB"));
baseRealDataSet.setIThdC(FloatUtils.get2Float(map.get("Pq_ThdIC"))); baseRealDataSet.setIThdC(map.get("Pq_ThdIC"));
//电压不平衡度 //电压不平衡度
baseRealDataSet.setVUnbalance(FloatUtils.get2Float(map.get("Pq_UnbalNegUM"))); baseRealDataSet.setVUnbalance(map.get("Pq_UnbalNegUM"));
//电流不平衡度 //电流不平衡度
baseRealDataSet.setIUnbalance(FloatUtils.get2Float(map.get("Pq_UnbalNegIM"))); baseRealDataSet.setIUnbalance(map.get("Pq_UnbalNegIM"));
//有功功率 //有功功率
baseRealDataSet.setPA(FloatUtils.get2Float(map.get("Pq_PA"))); baseRealDataSet.setPA(map.get("Pq_PA"));
baseRealDataSet.setPB(FloatUtils.get2Float(map.get("Pq_PB"))); baseRealDataSet.setPB(map.get("Pq_PB"));
baseRealDataSet.setPC(FloatUtils.get2Float(map.get("Pq_PC"))); baseRealDataSet.setPC(map.get("Pq_PC"));
baseRealDataSet.setPTot(FloatUtils.get2Float(map.get("Pq_TotPM"))); baseRealDataSet.setPTot(map.get("Pq_TotPM"));
//无功功率 //无功功率
baseRealDataSet.setQA(FloatUtils.get2Float(map.get("Pq_QA"))); baseRealDataSet.setQA(map.get("Pq_QA"));
baseRealDataSet.setQB(FloatUtils.get2Float(map.get("Pq_QB"))); baseRealDataSet.setQB(map.get("Pq_QB"));
baseRealDataSet.setQC(FloatUtils.get2Float(map.get("Pq_QC"))); baseRealDataSet.setQC(map.get("Pq_QC"));
baseRealDataSet.setQTot(FloatUtils.get2Float(map.get("Pq_TotQM"))); baseRealDataSet.setQTot(map.get("Pq_TotQM"));
//视在功率 //视在功率
baseRealDataSet.setSA(FloatUtils.get2Float(map.get("Pq_SA"))); baseRealDataSet.setSA(map.get("Pq_SA"));
baseRealDataSet.setSB(FloatUtils.get2Float(map.get("Pq_SB"))); baseRealDataSet.setSB(map.get("Pq_SB"));
baseRealDataSet.setSC(FloatUtils.get2Float(map.get("Pq_SC"))); baseRealDataSet.setSC(map.get("Pq_SC"));
baseRealDataSet.setSTot(FloatUtils.get2Float(map.get("Pq_TotSM"))); baseRealDataSet.setSTot(map.get("Pq_TotSM"));
//视在功率因数 //视在功率因数
baseRealDataSet.setPfA(FloatUtils.get2Float(map.get("Pq_PFA"))); baseRealDataSet.setPfA(map.get("Pq_PFA"));
baseRealDataSet.setPfB(FloatUtils.get2Float(map.get("Pq_PFB"))); baseRealDataSet.setPfB(map.get("Pq_PFB"));
baseRealDataSet.setPfC(FloatUtils.get2Float(map.get("Pq_PFC"))); baseRealDataSet.setPfC(map.get("Pq_PFC"));
baseRealDataSet.setPfTot(FloatUtils.get2Float(map.get("Pq_TotPFM"))); baseRealDataSet.setPfTot(map.get("Pq_TotPFM"));
//位移功率因数 //位移功率因数
baseRealDataSet.setDpfA(FloatUtils.get2Float(map.get("Pq_DFA"))); baseRealDataSet.setDpfA(map.get("Pq_DFA"));
baseRealDataSet.setDpfB(FloatUtils.get2Float(map.get("Pq_DFB"))); baseRealDataSet.setDpfB(map.get("Pq_DFB"));
baseRealDataSet.setDpfC(FloatUtils.get2Float(map.get("Pq_DFC"))); baseRealDataSet.setDpfC(map.get("Pq_DFC"));
baseRealDataSet.setDpfTot(FloatUtils.get2Float(map.get("Pq_TotDFM"))); baseRealDataSet.setDpfTot(map.get("Pq_TotDFM"));
return baseRealDataSet;
}
public BaseRealDataSet channelData2(Map<String,Float> map) {
BaseRealDataSet baseRealDataSet = new BaseRealDataSet();
//频率
baseRealDataSet.setFreq(map.get("Pq_FreqM"));
//频率偏差
baseRealDataSet.setFreqDev(map.get("Pq_FreqDevM"));
//判断监测点的接线方式，不同接线方式电压有效值取值不同
//星型-相电压 角形、V型-线电压
//电压有效值
baseRealDataSet.setVRmsA(map.get("Pq_RmsUA"));
baseRealDataSet.setVRmsB(map.get("Pq_RmsUB"));
baseRealDataSet.setVRmsC(map.get("Pq_RmsUC"));
//基波电压幅值
baseRealDataSet.setV1A(map.get("Pq_RmsFundUA"));
baseRealDataSet.setV1B(map.get("Pq_RmsFundUB"));
baseRealDataSet.setV1C(map.get("Pq_RmsFundUC"));
//电流有效值
baseRealDataSet.setIRmsA(map.get("Pq_RmsIA"));
baseRealDataSet.setIRmsB(map.get("Pq_RmsIB"));
baseRealDataSet.setIRmsC(map.get("Pq_RmsIC"));
//基波电流幅值
baseRealDataSet.setI1A(map.get("Pq_RmsFundIA"));
baseRealDataSet.setI1B(map.get("Pq_RmsFundIB"));
baseRealDataSet.setI1C(map.get("Pq_RmsFundIC"));
//电压偏差
baseRealDataSet.setVDevA(map.get("Pq_UDevA"));
baseRealDataSet.setVDevB(map.get("Pq_UDevB"));
baseRealDataSet.setVDevC(map.get("Pq_UDevC"));
//基波电压相位
baseRealDataSet.setV1AngA(map.get("Pq_FundUAngA"));
baseRealDataSet.setV1AngB(map.get("Pq_FundUAngB"));
baseRealDataSet.setV1AngC(map.get("Pq_FundUAngC"));
//基波电流相位
baseRealDataSet.setI1AngA(map.get("Pq_FundIAngA"));
baseRealDataSet.setI1AngB(map.get("Pq_FundIAngB"));
baseRealDataSet.setI1AngC(map.get("Pq_FundIAngC"));
//电压总谐波畸变率
baseRealDataSet.setVThdA(map.get("Pq_ThdUA"));
baseRealDataSet.setVThdB(map.get("Pq_ThdUB"));
baseRealDataSet.setVThdC(map.get("Pq_ThdUC"));
//电流总谐波畸变率
baseRealDataSet.setIThdA(map.get("Pq_ThdIA"));
baseRealDataSet.setIThdB(map.get("Pq_ThdIB"));
baseRealDataSet.setIThdC(map.get("Pq_ThdIC"));
//电压不平衡度
baseRealDataSet.setVUnbalance(map.get("Pq_UnbalNegUM"));
//电流不平衡度
baseRealDataSet.setIUnbalance(map.get("Pq_UnbalNegIM"));
//有功功率
baseRealDataSet.setPA(map.get("Pq_PA"));
baseRealDataSet.setPB(map.get("Pq_PB"));
baseRealDataSet.setPC(map.get("Pq_PC"));
baseRealDataSet.setPTot(map.get("Pq_TotPM"));
//无功功率
baseRealDataSet.setQA(map.get("Pq_QA"));
baseRealDataSet.setQB(map.get("Pq_QB"));
baseRealDataSet.setQC(map.get("Pq_QC"));
baseRealDataSet.setQTot(map.get("Pq_TotQM"));
//视在功率
baseRealDataSet.setSA(map.get("Pq_SA"));
baseRealDataSet.setSB(map.get("Pq_SB"));
baseRealDataSet.setSC(map.get("Pq_SC"));
baseRealDataSet.setSTot(map.get("Pq_TotSM"));
//视在功率因数
baseRealDataSet.setPfA(map.get("Pq_PFA"));
baseRealDataSet.setPfB(map.get("Pq_PFB"));
baseRealDataSet.setPfC(map.get("Pq_PFC"));
baseRealDataSet.setPfTot(map.get("Pq_TotPFM"));
//位移功率因数
baseRealDataSet.setDpfA(map.get("Pq_DFA"));
baseRealDataSet.setDpfB(map.get("Pq_DFB"));
baseRealDataSet.setDpfC(map.get("Pq_DFC"));
baseRealDataSet.setDpfTot(map.get("Pq_TotDFM"));
return baseRealDataSet; return baseRealDataSet;
} }
@@ -346,14 +430,14 @@ public class RtServiceImpl implements IRtService {
if (Objects.equals(item.getHarmName(),"Pq_RmsFundI")) { if (Objects.equals(item.getHarmName(),"Pq_RmsFundI")) {
if ("Secondary".equals(dataLevel)) { if ("Secondary".equals(dataLevel)) {
double data = item.getData() * ct; double data = item.getData() * ct;
harmRealDataSet.setData1(FloatUtils.get2Float((float)data)); harmRealDataSet.setData1((float)data);
} else { } else {
harmRealDataSet.setData1(FloatUtils.get2Float(item.getData())); harmRealDataSet.setData1(item.getData());
} }
} else if (Objects.equals(item.getHarmName(),"Pq_RmsFundU")) { } else if (Objects.equals(item.getHarmName(),"Pq_RmsFundU")) {
harmRealDataSet.setData1(FloatUtils.get2Float(item.getData())); harmRealDataSet.setData1(item.getData());
} else if (Objects.equals(item.getHarmName(),"Pq_ThdU")) { } else if (Objects.equals(item.getHarmName(),"Pq_ThdU")) {
harmRealDataSet.setData1(FloatUtils.get2Float(item.getData())); harmRealDataSet.setData1(item.getData());
} else { } else {
String numberStr = item.getHarmName().substring(item.getHarmName().lastIndexOf('_') + 1); String numberStr = item.getHarmName().substring(item.getHarmName().lastIndexOf('_') + 1);
String fieldName = "data" + numberStr; String fieldName = "data" + numberStr;
@@ -363,12 +447,12 @@ public class RtServiceImpl implements IRtService {
if (item.getHarmName().contains("Pq_HarmI_")) { if (item.getHarmName().contains("Pq_HarmI_")) {
if ("Secondary".equals(dataLevel)) { if ("Secondary".equals(dataLevel)) {
double data = item.getData() * ct; double data = item.getData() * ct;
field.set(harmRealDataSet,FloatUtils.get2Float((float)data)); field.set(harmRealDataSet,(float)data);
} else { } else {
field.set(harmRealDataSet,FloatUtils.get2Float(item.getData())); field.set(harmRealDataSet,item.getData());
} }
} else { } else {
field.set(harmRealDataSet,FloatUtils.get2Float(item.getData())); field.set(harmRealDataSet,item.getData());
} }
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
@@ -378,26 +462,4 @@ public class RtServiceImpl implements IRtService {
return harmRealDataSet; return harmRealDataSet;
} }
private Set<String> convertObjectToSetSafe(Object obj) {
if (obj == null) {
return new HashSet<>();
}
if (obj instanceof Set) {
// 类型安全的转换
Set<?> rawSet = (Set<?>) obj;
return rawSet.stream()
.filter(Objects::nonNull)
.map(Object::toString)
.collect(Collectors.toSet());
} else if (obj instanceof Collection) {
return ((Collection<?>) obj).stream()
.filter(Objects::nonNull)
.map(Object::toString)
.collect(Collectors.toSet());
} else {
log.warn("Redis中的对象类型不是Set或Collection: {}", obj.getClass().getName());
return new HashSet<>();
}
}
} }

4
iot-analysis/analysis-stat/stat-boot/pom.xml
View File

@@ -21,6 +21,10 @@
</properties> </properties>
<dependencies> <dependencies>
<dependency>
<groupId>com.github.tocrhz</groupId>
<artifactId>mqtt-spring-boot-starter</artifactId>
</dependency>
<dependency> <dependency>
<groupId>com.njcn</groupId> <groupId>com.njcn</groupId>
<artifactId>common-web</artifactId> <artifactId>common-web</artifactId>

21
iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/impl/StatServiceImpl.java
View File

@@ -1,6 +1,7 @@
package com.njcn.stat.service.impl; package com.njcn.stat.service.impl;
import cn.hutool.core.collection.CollectionUtil; import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import com.alibaba.nacos.shaded.com.google.gson.Gson; import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.njcn.access.api.CsDeviceFeignClient; import com.njcn.access.api.CsDeviceFeignClient;
import com.njcn.access.api.CsLineLatestDataFeignClient; import com.njcn.access.api.CsLineLatestDataFeignClient;
@@ -9,13 +10,13 @@ import com.njcn.access.pojo.po.CsLineLatestData;
import com.njcn.access.utils.ChannelObjectUtil; import com.njcn.access.utils.ChannelObjectUtil;
import com.njcn.common.pojo.exception.BusinessException; import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.common.utils.PubUtils; import com.njcn.common.utils.PubUtils;
import com.njcn.csdevice.api.CsLineFeignClient; import com.njcn.csdevice.api.CsCommunicateFeignClient;
import com.njcn.csdevice.api.DataArrayFeignClient; import com.njcn.csdevice.api.DataArrayFeignClient;
import com.njcn.csdevice.api.DeviceMessageFeignClient; import com.njcn.csdevice.api.DeviceMessageFeignClient;
import com.njcn.csdevice.pojo.dto.PqsCommunicateDto;
import com.njcn.csdevice.pojo.param.DataArrayParam; import com.njcn.csdevice.pojo.param.DataArrayParam;
import com.njcn.csdevice.pojo.po.CsDataArray; import com.njcn.csdevice.pojo.po.CsDataArray;
import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO; import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
import com.njcn.csdevice.pojo.po.CsLinePO;
import com.njcn.influx.pojo.constant.InfluxDBTableConstant; import com.njcn.influx.pojo.constant.InfluxDBTableConstant;
import com.njcn.influx.utils.InfluxDbUtils; import com.njcn.influx.utils.InfluxDbUtils;
import com.njcn.mq.message.AppAutoDataMessage; import com.njcn.mq.message.AppAutoDataMessage;
@@ -23,9 +24,7 @@ import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.utils.RedisUtil; import com.njcn.redis.utils.RedisUtil;
import com.njcn.stat.enums.StatResponseEnum; import com.njcn.stat.enums.StatResponseEnum;
import com.njcn.stat.service.IStatService; import com.njcn.stat.service.IStatService;
import com.njcn.system.api.DicDataFeignClient;
import com.njcn.system.enums.DicDataEnum; import com.njcn.system.enums.DicDataEnum;
import com.njcn.system.pojo.po.DictData;
import com.njcn.system.pojo.po.SysDicTreePO; import com.njcn.system.pojo.po.SysDicTreePO;
import lombok.AllArgsConstructor; import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@@ -38,6 +37,7 @@ import org.springframework.transaction.annotation.Transactional;
import java.time.Instant; import java.time.Instant;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.time.ZoneId; import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*; import java.util.*;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
@@ -60,11 +60,7 @@ public class StatServiceImpl implements IStatService {
private final CsLineLatestDataFeignClient csLineLatestDataFeignClient; private final CsLineLatestDataFeignClient csLineLatestDataFeignClient;
private final CsDeviceFeignClient csDeviceFeignClient; private final CsDeviceFeignClient csDeviceFeignClient;
private final DeviceMessageFeignClient deviceMessageFeignClient; private final DeviceMessageFeignClient deviceMessageFeignClient;
private static final Map<String, String> PHASE_MAPPING = new HashMap<String, String>() {{ private final CsCommunicateFeignClient csCommunicateFeignClient;
put("AB", "A");
put("BC", "B");
put("CA", "C");
}};
@Override @Override
@Transactional(rollbackFor = Exception.class) @Transactional(rollbackFor = Exception.class)
@@ -166,6 +162,13 @@ public class StatServiceImpl implements IStatService {
//判断设备运行状态 //判断设备运行状态
if (!Objects.isNull(po.getRunStatus()) && po.getRunStatus() == 1) { if (!Objects.isNull(po.getRunStatus()) && po.getRunStatus() == 1) {
csDeviceFeignClient.updateRunStatus(appAutoDataMessage.getId(), AccessEnum.ONLINE.getCode()); csDeviceFeignClient.updateRunStatus(appAutoDataMessage.getId(), AccessEnum.ONLINE.getCode());
//记录设备上线
PqsCommunicateDto dto = new PqsCommunicateDto();
dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
dto.setDevId(appAutoDataMessage.getId());
dto.setType(1);
dto.setDescription("通讯正常");
csCommunicateFeignClient.insertion(dto);
} }
} }
System.gc(); System.gc();

46
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/CommonBaseMessage.java Normal file
View File

@@ -0,0 +1,46 @@
package com.njcn.zlevent.pojo.dto;
import com.alibaba.fastjson.annotation.JSONField;
import com.fasterxml.jackson.annotation.JsonAlias;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.njcn.middle.rocket.domain.BaseMessage;
import lombok.Data;
/**
* @author caozehui
* @data 2026-03-31
*/
@Data
public class CommonBaseMessage extends BaseMessage {
/**
* 消息请求的唯一标识
*/
@JsonProperty("guid")
@JsonAlias({"guid"})
private String guid;
/**
* 设备Mac
*/
@JsonProperty("devMac")
@JsonAlias({"Dev_mac"})
private String devMac;
/**
* 前置Id
*/
@JsonProperty("frontId")
@JsonAlias({"FrontId"})
private String frontId;
/**
* 前置进程号
*/
@JsonProperty("node")
@JsonAlias({"Node"})
private Integer node;
@JsonProperty("detail")
@JsonAlias({"Detail"})
private Object detail;
}

205
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/DevVersionResponeDTO.java Normal file
View File

@@ -0,0 +1,205 @@
package com.njcn.zlevent.pojo.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import java.time.LocalDate;
/**
* @author caozehui
* @data 2026-03-16
*/
@Data
public class DevVersionResponeDTO extends CommonBaseMessage {
@JsonProperty("Detail")
private DevVersionResponeDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JsonProperty("Type")
private Integer type;
@JsonProperty("Msg")
private DevVersionResponeDTO.Msg msg;
}
@Data
public static class Msg {
@JsonProperty("Code")
private Integer code;
@JsonProperty("VersionInfo")
private VersionInfo versionInfo;
}
@Data
public static class VersionInfo {
/**
* 装置基础型号（只用于程序升级鉴别）
*/
@JsonProperty("BaseModel")
private String baseModel;
/**
* 云服务协议版本
*/
@JsonProperty("CloudProtocolVer")
private String cloudProtocolVer;
/**
* 应用程序版本号
*/
@JsonProperty("AppVersion")
private String appVersion;
/**
* 应用程序版本日期
*/
@JsonProperty("AppDate")
private LocalDate appDate;
/**
* 应用程序校验码
*/
@JsonProperty("AppChecksum")
private String appChecksum;
/**
* 电压接线方式（0-星1-三角2-V）
*/
@JsonProperty("VoltageWiring")
private String voltageWiring;
/**
* 电流B相是否合成（0-否1-是）
*/
@JsonProperty("CurrentBSynthetic")
private String currentBSynthetic;
/**
* 数据统计时间间隔（单位分钟）
*/
@JsonProperty("DataStatInterval")
private Integer dataStatInterval;
/**
* 额定电压（二次值，单位V）
*/
@JsonProperty("RatedVoltage")
private Double ratedVoltage;
/**
* PT变比
*/
@JsonProperty("PTRatio")
private Integer ptRatio;
/**
* CT变比
*/
@JsonProperty("CTRatio")
private Integer ctRatio;
/**
* sntp对时IP
*/
@JsonProperty("SntpIP")
private String sntpIP;
/**
* sntp对时端口
*/
@JsonProperty("SntpPort")
private Integer sntpPort;
/**
* sntp对时间隔（单位分钟）
*/
@JsonProperty("SntpInterval")
private Integer sntpInterval;
/**
* Web端口
*/
@JsonProperty("WebPort")
private Integer webPort;
/**
* FTP端口
*/
@JsonProperty("FtpPort")
private Integer ftpPort;
/**
* Pqdif文件时间间隔（单位小时）
*/
@JsonProperty("PqdifInterval")
private Integer pqdifInterval;
/**
* 录波文件包含文件类型数
*/
@JsonProperty("WaveFileTypeCount")
private Integer waveFileTypeCount;
/**
* 特殊程序版本信息
*/
@JsonProperty("SpecialVersion")
private String specialVersion;
/**
* 装置型号（具体型号全称）
*/
@JsonProperty("DeviceModel")
private String deviceModel;
/**
* 谐波电度版本标志（0-否1-是）
*/
@JsonProperty("HarmonicEnergyFlag")
private Integer harmonicEnergyFlag;
/**
* 物理设备名称（仅用于上位机录波文件拼接）
*/
@JsonProperty("PhysicalName")
private String physicalName;
/**
* 录波LD名称（仅用于上位机录波文件拼接）
*/
@JsonProperty("WaveLDName")
private String waveLDName;
/**
* 高频谐波功能标志（0-否1-是）
*/
@JsonProperty("HighFreqHarmonicFlag")
private Integer highFreqHarmonicFlag;
/**
* 投入的通讯协议（2字节十六进制数）
*/
@JsonProperty("CommProtocols")
private Integer commProtocols;
/**
* 投入的对时方式选择（2字节十六进制数）
*/
@JsonProperty("TimeSyncMethods")
private Integer timeSyncMethods;
/**
* 装置功能配置（2字节十六进制数）
*/
@JsonProperty("DeviceFunctions")
private Integer deviceFunctions;
}
}

55
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/DeviceVersionRequestDTO.java Normal file
View File

@@ -0,0 +1,55 @@
package com.njcn.zlevent.pojo.dto;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.Data;
import java.util.Map;
/**
* @author caozehui
* @data 2026-04-01
*/
@Data
public class DeviceVersionRequestDTO {
/**
* 消息请求的唯一标识
*/
@JSONField(name = "guid")
private String guid;
/**
* 设备ID
*/
@JSONField(name = "Dev_id")
private String devId;
/**
* 前置Id
*/
@JSONField(name = "FrontId")
private String frontId;
/**
* 前置进程号
*/
@JSONField(name = "Node")
private Integer node;
@JSONField(name = "Detail")
private DeviceVersionRequestDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JSONField(name = "Type")
private Integer type;
/**
* 详情
*/
@JSONField(name = "Msg")
private Map<String, Object> msg;
}
}

60
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/FileDownloadRequestDTO.java Normal file
View File

@@ -0,0 +1,60 @@
package com.njcn.zlevent.pojo.dto;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.Data;
/**
* @author caozehui
* @data 2026-03-16
*/
@Data
public class FileDownloadRequestDTO {
/**
* 消息请求的唯一标识
*/
@JSONField(name = "guid")
private String guid;
/**
* 设备ID
*/
@JSONField(name = "Dev_id")
private String devId;
/**
* 前置Id
*/
@JSONField(name = "FrontId")
private String frontId;
/**
* 前置进程号
*/
@JSONField(name = "Node")
private Integer node;
@JSONField(name = "Detail")
private FileDownloadRequestDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JSONField(name = "Type")
private Integer type;
@JSONField(name = "Msg")
private FileDownloadRequestDTO.Msg msg;
}
@Data
public static class Msg {
/**
* 设备的文件名，例如：/etc/vol1_stat.txt
*/
@JSONField(name = "Name")
private String name;
}
}

42
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/FileDownloadResponeDTO.java Normal file
View File

@@ -0,0 +1,42 @@
package com.njcn.zlevent.pojo.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
/**
* @author caozehui
* @data 2026-03-16
*/
@Data
public class FileDownloadResponeDTO extends CommonBaseMessage {
@JsonProperty("Detail")
private FileDownloadResponeDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JsonProperty("Type")
private Integer type;
@JsonProperty("Msg")
private FileDownloadResponeDTO.Msg msg;
}
@Data
public static class Msg {
/**
* 文件名称，例如 /etc/vol1_stat.txt
*/
@JsonProperty("Name")
private String name;
/**
* 远端文件名，例如 /download/vol1_stat.txt
*/
@JsonProperty("RemoteName")
private String remoteName;
}
}

57
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/FileInfoRequestDTO.java Normal file
View File

@@ -0,0 +1,57 @@
package com.njcn.zlevent.pojo.dto;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.Data;
/**
* @author caozehui
* @data 2026-03-18
*/
@Data
public class FileInfoRequestDTO {
/**
* 消息请求的唯一标识
*/
@JSONField(name = "guid")
private String guid;
/**
* 设备ID
*/
@JSONField(name = "Dev_id")
private String devId;
/**
* 前置Id
*/
@JSONField(name = "FrontId")
private String frontId;
/**
* 前置进程号
*/
@JSONField(name = "Node")
private Integer node;
@JSONField(name = "Detail")
private FileInfoRequestDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JSONField(name = "Type")
private Integer type;
@JSONField(name = "Msg")
private FileInfoRequestDTO.Msg msg;
}
@Data
public static class Msg {
@JSONField(name = "Name")
private String name;
}
}

79
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/FileInfoResponseDTO.java Normal file
View File

@@ -0,0 +1,79 @@
package com.njcn.zlevent.pojo.dto;
import com.fasterxml.jackson.annotation.JsonAlias;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import java.util.List;
/**
* @author caozehui
* @data 2026-03-18
*/
@Data
public class FileInfoResponseDTO extends CommonBaseMessage {
@JsonProperty("detail")
@JsonAlias({"Detail"})
private FileInfoResponseDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JsonProperty("type")
@JsonAlias({"Type"})
private Integer type;
/**
* 消息主体
*/
@JsonProperty("msg")
@JsonAlias({"Msg"})
private FileInfoResponseDTO.Msg msg;
/**
* 响应码
*/
@JsonProperty("code")
@JsonAlias({"Code"})
private Integer code;
}
@Data
public static class Msg {
/**
* 目录信息数组
*/
@JsonProperty("dirInfo")
@JsonAlias({"DirInfo"})
private List<FileInfoResponseDTO.ResourceElement> dirInfo;
}
@Data
public static class ResourceElement {
/**
* 文件名/文件夹名称
*/
@JsonProperty("name")
@JsonAlias({"Name"})
private String name;
/**
* 类型，文件/文件夹
*/
@JsonProperty("type")
@JsonAlias({"Type"})
private String type;
/**
* 目录信息数组单个元素的数据成员大小
*/
@JsonProperty("size")
@JsonAlias({"Size"})
private Integer size;
private String prjDataPath;
}
}

58
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/FileOrDirDeleteRequestDTO.java Normal file
View File

@@ -0,0 +1,58 @@
package com.njcn.zlevent.pojo.dto;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.Data;
/**
* @author caozehui
* @data 2026-04-01
*/
@Data
public class FileOrDirDeleteRequestDTO {
/**
* 消息请求的唯一标识
*/
@JSONField(name = "guid")
private String guid;
/**
* 设备ID
*/
@JSONField(name = "Dev_id")
private String devId;
/**
* 前置Id
*/
@JSONField(name = "FrontId")
private String frontId;
/**
* 前置进程号
*/
@JSONField(name = "Node")
private Integer node;
@JSONField(name = "Detail")
private FileOrDirDeleteRequestDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JSONField(name = "Type")
private Integer type;
@JSONField(name = "Msg")
private FileOrDirDeleteRequestDTO.Msg msg;
}
@Data
public static class Msg {
@JSONField(name = "Name")
private String name;
}
}

39
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/FileOrDirDeleteResponeDTO.java Normal file
View File

@@ -0,0 +1,39 @@
package com.njcn.zlevent.pojo.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import java.util.Map;
/**
* @author caozehui
* @data 2026-03-30
*/
@Data
public class FileOrDirDeleteResponeDTO extends CommonBaseMessage {
@JsonProperty("Detail")
private FileOrDirDeleteResponeDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JsonProperty("Type")
private Integer type;
/**
* 响应主体
*/
@JsonProperty("Msg")
private Map<String, Object> msg;
/**
* 响应码
*/
@JsonProperty("Code")
private Integer code;
}
}

60
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/FileUploadRequestDTO.java Normal file
View File

@@ -0,0 +1,60 @@
package com.njcn.zlevent.pojo.dto;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.Data;
/**
* @author caozehui
* @data 2026-03-18
*/
@Data
public class FileUploadRequestDTO {
/**
* 消息请求的唯一标识
*/
@JSONField(name = "guid")
private String guid;
/**
* 设备ID
*/
@JSONField(name = "Dev_id")
private String devId;
/**
* 前置Id
*/
@JSONField(name = "FrontId")
private String frontId;
/**
* 前置进程号
*/
@JSONField(name = "Node")
private Integer node;
@JSONField(name = "Detail")
private FileUploadRequestDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JSONField(name = "Type")
private Integer type;
@JSONField(name = "Msg")
private FileUploadRequestDTO.Msg msg;
}
@Data
public static class Msg {
@JSONField(name = "Name")
private String name;
@JSONField(name = "RemoteName")
private String remoteName;
}
}

39
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/FileUploadResponeDTO.java Normal file
View File

@@ -0,0 +1,39 @@
package com.njcn.zlevent.pojo.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import java.util.Map;
/**
* @author caozehui
* @data 2026-03-30
*/
@Data
public class FileUploadResponeDTO extends CommonBaseMessage {
@JsonProperty("Detail")
private FileUploadResponeDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JsonProperty("Type")
private Integer type;
/**
* 响应主体
*/
@JsonProperty("Msg")
private Map<String, Object> msg;
/**
* 响应码
*/
@JsonProperty("Code")
private Integer code;
}
}

57
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/MkdirRequestDTO.java Normal file
View File

@@ -0,0 +1,57 @@
package com.njcn.zlevent.pojo.dto;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.Data;
/**
* @author caozehui
* @data 2026-03-18
*/
@Data
public class MkdirRequestDTO {
/**
* 消息请求的唯一标识
*/
@JSONField(name = "guid")
private String guid;
/**
* 设备ID
*/
@JSONField(name = "Dev_id")
private String devId;
/**
* 前置Id
*/
@JSONField(name = "FrontId")
private String frontId;
/**
* 前置进程号
*/
@JSONField(name = "Node")
private Integer node;
@JSONField(name = "Detail")
private MkdirRequestDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JSONField(name = "Type")
private Integer type;
@JSONField(name = "Msg")
private MkdirRequestDTO.Msg msg;
}
@Data
public static class Msg {
@JSONField(name = "Name")
private String name;
}
}

39
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/MkdirResponeDTO.java Normal file
View File

@@ -0,0 +1,39 @@
package com.njcn.zlevent.pojo.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import java.util.Map;
/**
* @author caozehui
* @data 2026-03-30
*/
@Data
public class MkdirResponeDTO extends CommonBaseMessage {
@JsonProperty("Detail")
private MkdirResponeDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JsonProperty("Type")
private Integer type;
/**
* 响应主体
*/
@JsonProperty("Msg")
private Map<String, Object> msg;
/**
* 响应码
*/
@JsonProperty("Code")
private Integer code;
}
}

55
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/RebootRequestDTO.java Normal file
View File

@@ -0,0 +1,55 @@
package com.njcn.zlevent.pojo.dto;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.Data;
import java.util.Map;
/**
* @author caozehui
* @data 2026-04-01
*/
@Data
public class RebootRequestDTO {
/**
* 消息请求的唯一标识
*/
@JSONField(name = "guid")
private String guid;
/**
* 设备ID
*/
@JSONField(name = "Dev_id")
private String devId;
/**
* 前置Id
*/
@JSONField(name = "FrontId")
private String frontId;
/**
* 前置进程号
*/
@JSONField(name = "Node")
private Integer node;
@JSONField(name = "Detail")
private RebootRequestDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JSONField(name = "Type")
private Integer type;
/**
* 详情
*/
@JSONField(name = "Msg")
private Map<String, Object> msg;
}
}

39
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/RebootResponeDTO.java Normal file
View File

@@ -0,0 +1,39 @@
package com.njcn.zlevent.pojo.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import java.util.Map;
/**
* @author caozehui
* @data 2026-03-30
*/
@Data
public class RebootResponeDTO extends CommonBaseMessage {
@JsonProperty("Detail")
private RebootResponeDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JsonProperty("Type")
private Integer type;
/**
* 响应主体
*/
@JsonProperty("Msg")
private Map<String, Object> msg;
/**
* 响应码
*/
@JsonProperty("Code")
private Integer code;
}
}

56
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/TimeSyncRequestDTO.java Normal file
View File

@@ -0,0 +1,56 @@
package com.njcn.zlevent.pojo.dto;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.Data;
import java.util.Map;
/**
* @author caozehui
* @data 2026-04-27
*/
@Data
public class TimeSyncRequestDTO {
/**
* 消息请求的唯一标识
*/
@JSONField(name = "guid")
private String guid;
/**
* 设备ID
*/
@JSONField(name = "Dev_id")
private String devId;
/**
* 前置Id
*/
@JSONField(name = "FrontId")
private String frontId;
/**
* 前置进程号
*/
@JSONField(name = "Node")
private Integer node;
@JSONField(name = "Detail")
private TimeSyncRequestDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JSONField(name = "Type")
private Integer type;
/**
* 详情
*/
@JSONField(name = "Msg")
private Map<String, Object> msg;
}
}

61
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/UpgradeRequestDTO.java Normal file
View File

@@ -0,0 +1,61 @@
package com.njcn.zlevent.pojo.dto;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.Data;
import java.util.Map;
/**
* @author caozehui
* @data 2026-04-01
*/
@Data
public class UpgradeRequestDTO {
/**
* 消息请求的唯一标识
*/
@JSONField(name = "guid")
private String guid;
/**
* 设备ID
*/
@JSONField(name = "Dev_id")
private String devId;
/**
* 前置Id
*/
@JSONField(name = "FrontId")
private String frontId;
/**
* 前置进程号
*/
@JSONField(name = "Node")
private Integer node;
@JSONField(name = "Detail")
private UpgradeRequestDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JSONField(name = "Type")
private Integer type;
/**
* 详情
*/
@JSONField(name = "Msg")
private UpgradeRequestDTO.Msg msg;
}
@Data
public static class Msg {
@JSONField(name = "Name")
private String name;
}
}

39
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/UpgradeResponeDTO.java Normal file
View File

@@ -0,0 +1,39 @@
package com.njcn.zlevent.pojo.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import java.util.Map;
/**
* @author caozehui
* @data 2026-03-30
*/
@Data
public class UpgradeResponeDTO extends CommonBaseMessage {
@JsonProperty("Detail")
private UpgradeResponeDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JsonProperty("Type")
private Integer type;
/**
* 响应主体
*/
@JsonProperty("Msg")
private Map<String, Object> msg;
/**
* 响应码
*/
@JsonProperty("Code")
private Integer code;
}
}

55
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/WorkingLogRequestDTO.java Normal file
View File

@@ -0,0 +1,55 @@
package com.njcn.zlevent.pojo.dto;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.Data;
import java.util.Map;
/**
* @author caozehui
* @data 2026-03-30
*/
@Data
public class WorkingLogRequestDTO {
/**
* 消息请求的唯一标识
*/
@JSONField(name = "guid")
private String guid;
/**
* 设备ID
*/
@JSONField(name = "Dev_id")
private String devId;
/**
* 前置Id
*/
@JSONField(name = "FrontId")
private String frontId;
/**
* 前置进程号
*/
@JSONField(name = "Node")
private Integer node;
@JSONField(name = "Detail")
private WorkingLogRequestDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JSONField(name = "Type")
private Integer type;
/**
* 详情
*/
@JSONField(name = "Msg")
private Map<String, Object> msg;
}
}

101
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/dto/WorkingLogResponeDTO.java Normal file
View File

@@ -0,0 +1,101 @@
package com.njcn.zlevent.pojo.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import java.time.LocalDateTime;
/**
* @author caozehui
* @data 2026-03-30
*/
@Data
public class WorkingLogResponeDTO extends CommonBaseMessage {
@JsonProperty("Detail")
private WorkingLogResponeDTO.Detail detail;
@Data
public static class Detail {
/**
* 数据类型，代表特定功能
*/
@JsonProperty("Type")
private Integer type;
/**
* 响应主体
*/
@JsonProperty("Msg")
private WorkingLogResponeDTO.Msg msg;
/**
* 响应码
*/
@JsonProperty("Code")
private Integer code;
}
@Data
public static class Msg {
/**
* 时标
*/
@JsonProperty("Time")
private LocalDateTime time;
/**
* CPU负荷（单位%） 39_38(双核，单核的为一个)
*/
@JsonProperty("CpuLoad")
private String cpuLoad;
/**
* 装置剩余内存（单位MB）
*/
@JsonProperty("FreeMemory")
private String freeMemory;
/**
* 装置总内存（单位MB）
*/
@JsonProperty("TotalMemory")
private String totalMemory;
/**
* 装置主存储器剩余空间(单位GB)
*/
@JsonProperty("FreeStorage")
private String freeStorage;
/**
* 装置主存储器总空间（单位GB）
*/
@JsonProperty("TotalStorage")
private String totalStorage;
/**
* 硬对时最后时标（B码或秒秒冲）
*/
@JsonProperty("HardTimeSync")
private LocalDateTime hardTimeSync;
/**
* Sntp对时最后时标
*/
@JsonProperty("SntpTimeSync")
private LocalDateTime sntpTimeSync;
/**
* 云服务协议对时最后时标
*/
@JsonProperty("CloudTimeSync")
private LocalDateTime cloudTimeSync;
/**
* 无线模块信号强度
*/
@JsonProperty("SignalStrength")
private String signalStrength;
}
}

13
iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/param/FileParam.java Normal file
View File

@@ -0,0 +1,13 @@
package com.njcn.zlevent.pojo.param;
import lombok.Data;
/**
* @author caozehui
* @data 2026-03-17
*/
@Data
public class FileParam {
private String filePath;
private String devId;
}

4
iot-analysis/analysis-zl-event/zl-event-boot/pom.xml
View File

@@ -20,6 +20,10 @@
</properties> </properties>
<dependencies> <dependencies>
<dependency>
<groupId>com.github.tocrhz</groupId>
<artifactId>mqtt-spring-boot-starter</artifactId>
</dependency>
<dependency> <dependency>
<groupId>com.baomidou</groupId> <groupId>com.baomidou</groupId>
<artifactId>dynamic-datasource-spring-boot-starter</artifactId> <artifactId>dynamic-datasource-spring-boot-starter</artifactId>

63
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/config/TaskSchedulerConfig.java Normal file
View File

@@ -0,0 +1,63 @@
package com.njcn.zlevent.config;
import cn.hutool.core.util.ObjectUtil;
import com.njcn.zlevent.service.IDeviceService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import java.time.Instant;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
@Slf4j
@Configuration
public class TaskSchedulerConfig {
@Autowired
private IDeviceService deviceService;
private final Map<String, ScheduledFuture<?>> scheduledTasks = new ConcurrentHashMap<>();
private final TaskScheduler taskScheduler;
public TaskSchedulerConfig() {
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.setPoolSize(3);
scheduler.setThreadNamePrefix("device-task-");
scheduler.setWaitForTasksToCompleteOnShutdown(true);
scheduler.initialize();
this.taskScheduler = scheduler;
}
public void startTask(String devId, long intervalSeconds) {
if (scheduledTasks.containsKey(devId)) {
return;
}
ScheduledFuture<?> future = taskScheduler.schedule(
() -> deviceService.getWorkingLog(devId),
triggerContext -> Date.from(Instant.now().plusSeconds(intervalSeconds))
);
scheduledTasks.put(devId, future);
log.info("启动设备 {} 的定时任务成功，间隔={}秒", devId, intervalSeconds);
}
public void stopTask(String devId) {
ScheduledFuture<?> future = scheduledTasks.remove(devId);
if (ObjectUtil.isNotNull(future)) {
future.cancel(false);
log.info("停止设备 {} 的定时任务成功", devId);
}
}
public boolean isTaskRunning(String devId) {
ScheduledFuture<?> future = scheduledTasks.get(devId);
return ObjectUtil.isNotNull(future) && !future.isDone() && !future.isCancelled();
}
}

80
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/controller/DeviceController.java Normal file
View File

@@ -0,0 +1,80 @@
package com.njcn.zlevent.controller;
import com.njcn.common.pojo.annotation.OperateInfo;
import com.njcn.common.pojo.enums.common.LogEnum;
import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.common.utils.HttpResultUtil;
import com.njcn.web.controller.BaseController;
import com.njcn.zlevent.service.IDeviceService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
/**
* @author caozehui
* @data 2026-03-20
*/
@Slf4j
@RestController
@RequestMapping("/device")
@Api(tags = "操作设备")
@AllArgsConstructor
public class DeviceController extends BaseController {
private IDeviceService deviceService;
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@GetMapping("/workingLog")
@ApiOperation("开始获取装置运行日志")
public HttpResult<Boolean> startWorkingLog(@RequestParam("devId") String devId) {
String methodDescribe = getMethodDescribe("startWorkingLog");
deviceService.startWorkingLog(devId);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, true, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@GetMapping("/stopWorkingLog")
@ApiOperation("停止获取装置运行日志")
public HttpResult<Boolean> stopWorkingLog(@RequestParam("devId") String devId) {
String methodDescribe = getMethodDescribe("stopWorkingLog");
deviceService.stopWorkingLogTask(devId);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, true, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@GetMapping("/upgrade")
@ApiOperation("装置升级")
@ApiImplicitParams({
@ApiImplicitParam(name = "devId", value = "设备ID", required = true),
@ApiImplicitParam(name = "edDataId", value = "程序版本Id", required = true)
})
public HttpResult<Boolean> upgrade(@RequestParam("devId") String devId, @RequestParam("edDataId") String edDataId) {
String methodDescribe = getMethodDescribe("upgrade");
boolean res = deviceService.upgrade(devId, edDataId);
return HttpResultUtil.assembleCommonResponseResult(res ? CommonResponseEnum.SUCCESS : CommonResponseEnum.FAIL, res, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@GetMapping("/reboot")
@ApiOperation("装置重启")
public HttpResult<Boolean> reboot(@RequestParam("devId") String devId) {
String methodDescribe = getMethodDescribe("reboot");
boolean res = deviceService.reboot(devId);
return HttpResultUtil.assembleCommonResponseResult(res ? CommonResponseEnum.SUCCESS : CommonResponseEnum.FAIL, res, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@GetMapping("/timeSync")
@ApiOperation("装置对时")
public HttpResult<Boolean> timeSync(@RequestParam("devId") String devId) {
String methodDescribe = getMethodDescribe("listDir");
boolean res = deviceService.timeSync(devId);
return HttpResultUtil.assembleCommonResponseResult(res ? CommonResponseEnum.SUCCESS : CommonResponseEnum.FAIL, res, methodDescribe);
}
}

56
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/controller/FileController.java
View File

@@ -7,13 +7,20 @@ import com.njcn.common.pojo.response.HttpResult;
import com.njcn.common.utils.HttpResultUtil; import com.njcn.common.utils.HttpResultUtil;
import com.njcn.mq.message.AppFileMessage; import com.njcn.mq.message.AppFileMessage;
import com.njcn.web.controller.BaseController; import com.njcn.web.controller.BaseController;
import com.njcn.zlevent.pojo.dto.FileInfoResponseDTO;
import com.njcn.zlevent.pojo.param.FileParam;
import com.njcn.zlevent.service.IFileService; import com.njcn.zlevent.service.IFileService;
import io.swagger.annotations.Api; import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam; import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation; import io.swagger.annotations.ApiOperation;
import lombok.AllArgsConstructor; import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.*; import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import javax.servlet.http.HttpServletResponse;
import java.util.List;
/** /**
* 类的介绍： * 类的介绍：
@@ -61,4 +68,53 @@ public class FileController extends BaseController {
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe); return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
} }
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/listDir")
@ApiOperation("获取目录列表")
@ApiImplicitParam(name = "fileParam", value = "文件路径", required = true)
public HttpResult<List<FileInfoResponseDTO.ResourceElement>> listDir(@RequestBody FileParam fileParam) {
String methodDescribe = getMethodDescribe("listDir");
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, fileService.listDir(fileParam.getFilePath(), fileParam.getDevId()), methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/downloadFileFromFront")
@ApiOperation("下载文件")
@ApiImplicitParam(name = "fileParam", value = "文件参数", required = true)
public void downloadFileFromFront(@RequestBody FileParam fileParam, HttpServletResponse response) {
fileService.downloadFileFromFront(fileParam.getFilePath(), fileParam.getDevId(), response);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/mkdir")
@ApiOperation("创建目录")
@ApiImplicitParam(name = "fileParam", value = "文件参数", required = true)
public HttpResult<Boolean> mkdir(@RequestBody FileParam fileParam) {
String methodDescribe = getMethodDescribe("mkdir");
boolean res = fileService.mkdir(fileParam.getFilePath(), fileParam.getDevId());
return HttpResultUtil.assembleCommonResponseResult(res ? CommonResponseEnum.SUCCESS : CommonResponseEnum.FAIL, res, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/uploadFileToFront")
@ApiOperation("上传文件")
@ApiImplicitParams({
@ApiImplicitParam(name = "file", value = "文件", required = true),
@ApiImplicitParam(name = "devId", value = "设备ID", required = true),
@ApiImplicitParam(name = "dirPath", value = "文件所在路径", required = true)
})
public HttpResult<Boolean> uploadFileToFront(@RequestPart("file") MultipartFile file, @RequestParam("devId") String devId, @RequestParam("dirPath") String dirPath) {
boolean res = fileService.uploadFileToFront(file, devId, dirPath);
return HttpResultUtil.assembleCommonResponseResult(res ? CommonResponseEnum.SUCCESS : CommonResponseEnum.FAIL, res, getMethodDescribe("uploadFileToFront"));
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/delete")
@ApiOperation("删除文件/目录")
@ApiImplicitParam(name = "fileParam", value = "文件参数", required = true)
public HttpResult<Boolean> delete(@RequestBody FileParam fileParam) {
boolean res = fileService.delete(fileParam.getFilePath(), fileParam.getDevId());
return HttpResultUtil.assembleCommonResponseResult(res ? CommonResponseEnum.SUCCESS : CommonResponseEnum.FAIL, res, getMethodDescribe("delete"));
}
} }

33
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/producer/CommonProducer.java Normal file
View File

@@ -0,0 +1,33 @@
package com.njcn.zlevent.producer;
import com.njcn.middle.rocket.domain.BaseMessage;
import com.njcn.middle.rocket.template.RocketMQEnhanceTemplate;
import com.njcn.mq.constant.BusinessTopic;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component;
/**
* @author caozehui
* @data 2026-03-16
*/
@Component
public class CommonProducer extends RocketMQEnhanceTemplate {
public CommonProducer(RocketMQTemplate template) {
super(template);
}
/**
* @param message
* @param frontId
* @return
*/
public SendResult send(BaseMessage message, String frontId) {
return send(BusinessTopic.CLOUD_TOPIC, frontId, message);
}
public SendResult send(BaseMessage message) {
return send(BusinessTopic.CLOUD_REPLY_TOPIC, message);
}
}

67
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/IDeviceService.java Normal file
View File

@@ -0,0 +1,67 @@
package com.njcn.zlevent.service;
import com.njcn.zlevent.pojo.dto.DevVersionResponeDTO;
import com.njcn.zlevent.pojo.dto.DeviceVersionRequestDTO;
import org.springframework.web.multipart.MultipartFile;
/**
* @author caozehui
* @data 2026-03-20
*/
public interface IDeviceService {
/**
* 开始运行日志任务
*
* @param devId 设备id
*/
void startWorkingLog(String devId);
/**
* 停止运行日志任务
*
* @param devId
*/
void stopWorkingLogTask(String devId);
/**
* 运行日志任务是否正在运行
*
* @param devId
* @return
*/
boolean isWorkingLogTaskRunning(String devId);
void getWorkingLog(String devId);
/**
* 设备对时
*
* @param devId
* @return
*/
boolean timeSync(String devId);
/**
* 获取设备版本信息
*
* @param devId
* @return
*/
DevVersionResponeDTO.VersionInfo getDeviceVersion(String devId);
/**
* 设备升级
*
* @param devId
* @param edDataId
*/
boolean upgrade(String devId, String edDataId);
/**
* 重启设备
*
* @param devId
*/
boolean reboot(String devId);
}

38
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/IFileService.java
View File

@@ -1,6 +1,11 @@
package com.njcn.zlevent.service; package com.njcn.zlevent.service;
import com.njcn.mq.message.AppFileMessage; import com.njcn.mq.message.AppFileMessage;
import com.njcn.zlevent.pojo.dto.FileInfoResponseDTO;
import org.springframework.web.multipart.MultipartFile;
import javax.servlet.http.HttpServletResponse;
import java.util.List;
/** /**
* 类的介绍： * 类的介绍：
@@ -16,12 +21,14 @@ public interface IFileService {
* 解析文件流之前需要获取文件的信息,可能要特殊处理 * 解析文件流之前需要获取文件的信息,可能要特殊处理
* 1.文件过大要分片获取(单次请求文件大小不超过50k) * 1.文件过大要分片获取(单次请求文件大小不超过50k)
* 2.校验文件(md5或者crc) * 2.校验文件(md5或者crc)
*
* @param appFileMessage * @param appFileMessage
*/ */
void analysisFileInfo(AppFileMessage appFileMessage); void analysisFileInfo(AppFileMessage appFileMessage);
/** /**
* 获取文件流，解析文件 * 获取文件流，解析文件
*
* @param appFileMessage * @param appFileMessage
*/ */
void analysisFileStream(AppFileMessage appFileMessage); void analysisFileStream(AppFileMessage appFileMessage);
@@ -30,4 +37,35 @@ public interface IFileService {
* 下载补召文件 * 下载补召文件
*/ */
void downloadMakeUpFile(String nDid); void downloadMakeUpFile(String nDid);
/**
* 获取目录列表
*
* @param filePath
* @param devId
* @return
*/
List<FileInfoResponseDTO.ResourceElement> listDir(String filePath, String devId);
/**
* 从前置下载文件
*
* @param filePath
* @param devId
* @param response
*/
void downloadFileFromFront(String filePath, String devId, HttpServletResponse response);
/**
* 上传文件到文件服务器
*
* @param file
* @param devId
* @param dirPath
*/
boolean uploadFileToFront(MultipartFile file, String devId, String dirPath);
boolean mkdir(String filePath, String devId);
boolean delete(String filePath, String devId);
} }

288
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/DeviceServiceImpl.java Normal file
View File

@@ -0,0 +1,288 @@
package com.njcn.zlevent.service.impl;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.github.tocrhz.mqtt.publisher.MqttPublisher;
import com.njcn.access.enums.TypeEnum;
import com.njcn.access.utils.SendMessageUtil;
import com.njcn.csdevice.api.CsEdDataFeignClient;
import com.njcn.csdevice.api.CsSoftInfoFeignClient;
import com.njcn.csdevice.api.CsUpgradeLogsFeignClient;
import com.njcn.csdevice.api.EquipmentFeignClient;
import com.njcn.csdevice.pojo.dto.CsEquipmentDeliveryDTO;
import com.njcn.csdevice.pojo.po.CsEdDataPO;
import com.njcn.csdevice.pojo.po.CsSoftInfoPO;
import com.njcn.csdevice.pojo.po.CsUpgradeLogs;
import com.njcn.middle.rocket.domain.BaseMessage;
import com.njcn.oss.utils.FileStorageUtil;
import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.utils.RedisUtil;
import com.njcn.zlevent.config.TaskSchedulerConfig;
import com.njcn.zlevent.pojo.dto.*;
import com.njcn.zlevent.producer.CommonProducer;
import com.njcn.zlevent.service.IDeviceService;
import lombok.AllArgsConstructor;
import org.springframework.stereotype.Service;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
/**
* @author caozehui
* @data 2026-03-20
*/
@Service
@AllArgsConstructor
public class DeviceServiceImpl implements IDeviceService {
private final EquipmentFeignClient equipmentFeignClient;
private final CsSoftInfoFeignClient csSoftInfoFeignClient;
private final CsEdDataFeignClient csEdDataFeignClient;
private final CommonProducer commonProducer;
private final SendMessageUtil sendMessageUtil;
private final TaskSchedulerConfig taskSchedulerConfig;
private final RedisUtil redisUtil;
private final MqttPublisher publisher;
private final FileStorageUtil fileStorageUtil;
private final CsUpgradeLogsFeignClient csUpgradeLogsFeignClient;
@Override
public void startWorkingLog(String devId) {
taskSchedulerConfig.startTask(devId, 5);
}
@Override
public void stopWorkingLogTask(String devId) {
taskSchedulerConfig.stopTask(devId);
}
@Override
public boolean isWorkingLogTaskRunning(String devId) {
return taskSchedulerConfig.isTaskRunning(devId);
}
public void getWorkingLog(String devId) {
List<CsEquipmentDeliveryDTO> listHttpResult = equipmentFeignClient.queryDeviceById(Collections.singletonList(devId)).getData();
WorkingLogRequestDTO requestDTO = new WorkingLogRequestDTO();
requestDTO.setDevId(devId);
requestDTO.setGuid(IdUtil.simpleUUID());
requestDTO.setNode(listHttpResult.get(0).getNodeProcess());
requestDTO.setFrontId(listHttpResult.get(0).getNodeId());
WorkingLogRequestDTO.Detail detail = new WorkingLogRequestDTO.Detail();
detail.setType(Integer.valueOf(TypeEnum.WORKING_LOG.getCode()));
detail.setMsg(new HashMap<>());
requestDTO.setDetail(detail);
BaseMessage message = new BaseMessage();
message.setMessageBody(JSON.toJSONString(requestDTO));
// 使用 Redis 存储 guid 用于后续查询
redisUtil.saveByKeyWithExpire(AppRedisKey.COMMON_REQUEST + requestDTO.getGuid(), "pending", 120L);
// 发送
commonProducer.send(message, requestDTO.getFrontId());
// 轮询 Redis 等待响应
WorkingLogResponeDTO workingLogResponeDTO = JSON.parseObject(sendMessageUtil.waitForResponse(requestDTO.getGuid(), 10), WorkingLogResponeDTO.class);
WorkingLogResponeDTO.Detail detail1 = workingLogResponeDTO.getDetail();
WorkingLogResponeDTO.Msg msg1 = detail1.getMsg();
//mqtt推送给前端
publisher.send("/afafaidfasd", JSON.toJSONString(msg1), 1, false);
}
@Override
public boolean timeSync(String devId) {
List<CsEquipmentDeliveryDTO> listHttpResult = equipmentFeignClient.queryDeviceById(Collections.singletonList(devId)).getData();
BaseMessage message = new BaseMessage();
// FileDownloadRequestDTO requestDTO = new FileDownloadRequestDTO();
// requestDTO.setGuid(listHttpResult.get(0).getNodeId());
// requestDTO.setNode(listHttpResult.get(0).getNodeProcess());
// BaseRequestDTO message = new BaseRequestDTO<>();
// message.setGuid(IdUtil.simpleUUID());
// message.setFrontId(listHttpResult.get(0).getNodeId());
// message.setNode(listHttpResult.get(0).getNodeProcess());
// message.setDevId(devId);
// BaseRequestDTO.Detail<FileInfoRequestDTO> detail = new BaseRequestDTO.Detail<>();
// detail.setType(1113);
// detail.setMsg(null);
// message.setDetail(detail);
// pendingResponsesMap.put(message.getGuid(), new CompletableFuture<>());
// 发送
//deviceProducer.send(message);
// BaseMessage baseMessage = null;
// CompletableFuture<BaseMessage> future = pendingResponsesMap.get(message.getGuid());
// try {
// baseMessage = future.get(5, TimeUnit.SECONDS);
// } catch (Exception e) {
// throw new BusinessException(ZleventResoponseEnum.RESPONSE_ERROR);
// } finally {
// pendingResponsesMap.remove(message.getGuid());
// }
//
// Integer code = baseMessage.getDetail().getCode();
// return code == 200;
return true;
}
@Override
public DevVersionResponeDTO.VersionInfo getDeviceVersion(String devId) {
List<CsEquipmentDeliveryDTO> listHttpResult = equipmentFeignClient.queryDeviceById(Collections.singletonList(devId)).getData();
CsEquipmentDeliveryDTO csEquipmentDeliveryDTO = listHttpResult.get(0);
// 先询问一下旧的版本信息
DeviceVersionRequestDTO deviceVersionRequestDTO = new DeviceVersionRequestDTO();
deviceVersionRequestDTO.setDevId(devId);
deviceVersionRequestDTO.setGuid(IdUtil.simpleUUID());
deviceVersionRequestDTO.setNode(csEquipmentDeliveryDTO.getNodeProcess());
deviceVersionRequestDTO.setFrontId(csEquipmentDeliveryDTO.getNodeId());
DeviceVersionRequestDTO.Detail detail1 = new DeviceVersionRequestDTO.Detail();
detail1.setMsg(new HashMap<>());
detail1.setType(Integer.parseInt(TypeEnum.DEVICE_VERSION.getCode()));
deviceVersionRequestDTO.setDetail(detail1);
BaseMessage message = new BaseMessage();
message.setMessageBody(JSON.toJSONString(deviceVersionRequestDTO));
// 使用 Redis 存储 guid 用于后续查询
redisUtil.saveByKeyWithExpire(AppRedisKey.COMMON_REQUEST + deviceVersionRequestDTO.getGuid(), "pending", 120L);
// 发送
commonProducer.send(message, deviceVersionRequestDTO.getFrontId());
// 轮询 Redis 等待响应
DevVersionResponeDTO responseDTO = JSON.parseObject(sendMessageUtil.waitForResponse(deviceVersionRequestDTO.getGuid(), 10), DevVersionResponeDTO.class);
DevVersionResponeDTO.Detail detail2 = responseDTO.getDetail();
if (detail2.getMsg().getCode() == 200) {
DevVersionResponeDTO.VersionInfo versionInfo = detail2.getMsg().getVersionInfo();
return versionInfo;
}
return null;
}
@Override
public boolean upgrade(String devId, String edDataId) {
// 先获取旧的版本信息
//DevVersionResponeDTO.VersionInfo oldVersionInfo = this.getDeviceVersion(devId);
List<CsEquipmentDeliveryDTO> listHttpResult = equipmentFeignClient.queryDeviceById(Collections.singletonList(devId)).getData();
CsEquipmentDeliveryDTO csEquipmentDeliveryDTO = listHttpResult.get(0);
CsEdDataPO csEdDataPO = csEdDataFeignClient.getById(edDataId).getData();
String filePath = csEdDataPO.getFilePath();
// 装置升级日志
CsUpgradeLogs csUpgradeLogs = new CsUpgradeLogs();
csUpgradeLogs.setDevId(devId);
csUpgradeLogs.setVersionNo(csEdDataPO.getVersionNo());
csUpgradeLogs.setResult(0);
UpgradeRequestDTO requestDTO = new UpgradeRequestDTO();
requestDTO.setDevId(devId);
requestDTO.setGuid(IdUtil.simpleUUID());
requestDTO.setNode(csEquipmentDeliveryDTO.getNodeProcess());
requestDTO.setFrontId(csEquipmentDeliveryDTO.getNodeId());
UpgradeRequestDTO.Detail detail1 = new UpgradeRequestDTO.Detail();
detail1.setType(Integer.valueOf(TypeEnum.DEVICE_UPGRADE.getCode()));
UpgradeRequestDTO.Msg msg = new UpgradeRequestDTO.Msg();
msg.setName(filePath);
detail1.setMsg(msg);
requestDTO.setDetail(detail1);
BaseMessage message = new BaseMessage();
message.setMessageBody(JSON.toJSONString(requestDTO));
// 使用 Redis 存储 guid 用于后续查询
redisUtil.saveByKeyWithExpire(AppRedisKey.COMMON_REQUEST + requestDTO.getGuid(), "pending", 120L);
// 发送
commonProducer.send(message, requestDTO.getFrontId());
// 轮询 Redis 等待响应
UpgradeResponeDTO responeDTO = JSON.parseObject(sendMessageUtil.waitForResponse(requestDTO.getGuid(), 10), UpgradeResponeDTO.class);
UpgradeResponeDTO.Detail detail2 = responeDTO.getDetail();
if (detail2.getCode() == 200) {
// 修改数据库记录
String softinfoId = csEquipmentDeliveryDTO.getSoftinfoId();
if (StrUtil.isNotBlank(softinfoId)) {
csSoftInfoFeignClient.removeSoftInfo(softinfoId);
}
CsSoftInfoPO softInfoPO = new CsSoftInfoPO();
softInfoPO.setId(IdUtil.fastSimpleUUID());
softInfoPO.setAppCheck(csEdDataPO.getCrc());
softInfoPO.setAppDate(csEdDataPO.getVersionDate());
softInfoPO.setAppVersion(csEdDataPO.getVersionNo());
softInfoPO.setOpAttr("r");
softInfoPO.setOsName("VxWorks");
softInfoPO.setOsVersion("VxWorks");
softInfoPO.setSoftUpdate("yes");
csSoftInfoFeignClient.saveSoftInfo(softInfoPO);
equipmentFeignClient.updateSoftInfo(csEquipmentDeliveryDTO.getNdid(), softInfoPO.getId());
// 重新获取升级后的版本信息
DevVersionResponeDTO.VersionInfo newVersionInfo = this.getDeviceVersion(devId);
if (newVersionInfo.getAppVersion().equals(csEdDataPO.getVersionNo()) && newVersionInfo.getCloudProtocolVer().equals(csEdDataPO.getVersionAgreement())) {
// 修改数据库记录
equipmentFeignClient.updateSoftInfo(csEquipmentDeliveryDTO.getNdid(), softInfoPO.getId());
csUpgradeLogs.setResult(1);
csUpgradeLogsFeignClient.add(csUpgradeLogs);
return true;
}
csUpgradeLogsFeignClient.add(csUpgradeLogs);
return false;
}
csUpgradeLogsFeignClient.add(csUpgradeLogs);
return false;
}
@Override
public boolean reboot(String devId) {
List<CsEquipmentDeliveryDTO> listHttpResult = equipmentFeignClient.queryDeviceById(Collections.singletonList(devId)).getData();
RebootRequestDTO requestDTO = new RebootRequestDTO();
requestDTO.setDevId(devId);
requestDTO.setGuid(IdUtil.simpleUUID());
requestDTO.setNode(listHttpResult.get(0).getNodeProcess());
requestDTO.setFrontId(listHttpResult.get(0).getNodeId());
RebootRequestDTO.Detail detail1 = new RebootRequestDTO.Detail();
detail1.setType(Integer.parseInt(TypeEnum.DEVICE_REBOOT.getCode()));
detail1.setMsg(new HashMap<>());
requestDTO.setDetail(detail1);
BaseMessage message = new BaseMessage();
message.setMessageBody(JSON.toJSONString(requestDTO));
// 使用 Redis 存储 guid 用于后续查询
redisUtil.saveByKeyWithExpire(AppRedisKey.COMMON_REQUEST + requestDTO.getGuid(), "pending", 120L);
// 发送
commonProducer.send(message, requestDTO.getFrontId());
// 轮询 Redis 等待响应
RebootResponeDTO responeDTO = JSON.parseObject(sendMessageUtil.waitForResponse(requestDTO.getGuid(), 10), RebootResponeDTO.class);
if (responeDTO.getDetail().getCode() == 200) {
return true;
}
return false;
}
}

350
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/FileServiceImpl.java
View File

@@ -2,6 +2,7 @@ package com.njcn.zlevent.service.impl;
import cn.hutool.core.collection.CollectionUtil; import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.text.StrPool; import cn.hutool.core.text.StrPool;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil; import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
import com.alibaba.nacos.shaded.com.google.gson.Gson; import com.alibaba.nacos.shaded.com.google.gson.Gson;
@@ -12,19 +13,18 @@ import com.njcn.access.enums.AccessResponseEnum;
import com.njcn.access.enums.TypeEnum; import com.njcn.access.enums.TypeEnum;
import com.njcn.access.pojo.dto.ReqAndResDto; import com.njcn.access.pojo.dto.ReqAndResDto;
import com.njcn.access.pojo.dto.file.FileDto; import com.njcn.access.pojo.dto.file.FileDto;
import com.njcn.access.utils.CRC32Utils; import com.njcn.access.utils.*;
import com.njcn.access.utils.ChannelObjectUtil;
import com.njcn.access.utils.FileCommonUtils;
import com.njcn.access.utils.MqttUtil;
import com.njcn.common.config.GeneralInfo; import com.njcn.common.config.GeneralInfo;
import com.njcn.common.pojo.exception.BusinessException; import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.csdevice.api.DeviceFtpFeignClient; import com.njcn.csdevice.api.DeviceFtpFeignClient;
import com.njcn.csdevice.api.EquipmentFeignClient; import com.njcn.csdevice.api.EquipmentFeignClient;
import com.njcn.csdevice.api.PortableOffLogFeignClient; import com.njcn.csdevice.api.PortableOffLogFeignClient;
import com.njcn.csdevice.enums.AlgorithmResponseEnum; import com.njcn.csdevice.enums.AlgorithmResponseEnum;
import com.njcn.csdevice.pojo.dto.CsEquipmentDeliveryDTO;
import com.njcn.csharmonic.api.WavePicFeignClient; import com.njcn.csharmonic.api.WavePicFeignClient;
import com.njcn.csharmonic.enums.CsHarmonicResponseEnum; import com.njcn.csharmonic.enums.CsHarmonicResponseEnum;
import com.njcn.csharmonic.pojo.dto.DownloadMakeUpDto; import com.njcn.csharmonic.pojo.dto.DownloadMakeUpDto;
import com.njcn.middle.rocket.domain.BaseMessage;
import com.njcn.mq.message.AppFileMessage; import com.njcn.mq.message.AppFileMessage;
import com.njcn.oss.constant.GeneralConstant; import com.njcn.oss.constant.GeneralConstant;
import com.njcn.oss.constant.OssPath; import com.njcn.oss.constant.OssPath;
@@ -32,11 +32,10 @@ import com.njcn.oss.utils.FileStorageUtil;
import com.njcn.redis.pojo.enums.AppRedisKey; import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.utils.RedisUtil; import com.njcn.redis.utils.RedisUtil;
import com.njcn.zlevent.param.CsEventParam; import com.njcn.zlevent.param.CsEventParam;
import com.njcn.zlevent.pojo.dto.FileInfoDto; import com.njcn.zlevent.pojo.dto.*;
import com.njcn.zlevent.pojo.dto.FileStreamDto;
import com.njcn.zlevent.pojo.dto.WaveTimeDto;
import com.njcn.zlevent.pojo.po.CsEventFileLogs; import com.njcn.zlevent.pojo.po.CsEventFileLogs;
import com.njcn.zlevent.pojo.po.CsWave; import com.njcn.zlevent.pojo.po.CsWave;
import com.njcn.zlevent.producer.CommonProducer;
import com.njcn.zlevent.service.ICsEventFileLogsService; import com.njcn.zlevent.service.ICsEventFileLogsService;
import com.njcn.zlevent.service.ICsEventService; import com.njcn.zlevent.service.ICsEventService;
import com.njcn.zlevent.service.ICsWaveService; import com.njcn.zlevent.service.ICsWaveService;
@@ -46,12 +45,16 @@ import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject; import net.sf.json.JSONObject;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;
import javax.servlet.http.HttpServletResponse;
import java.io.*; import java.io.*;
import java.net.URLEncoder;
import java.nio.file.Files; import java.nio.file.Files;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
import java.util.*; import java.util.*;
import java.util.concurrent.CompletableFuture;
/** /**
* 类的介绍： * 类的介绍：
@@ -83,6 +86,11 @@ public class FileServiceImpl implements IFileService {
private final DeviceFtpFeignClient deviceFtpFeignClient; private final DeviceFtpFeignClient deviceFtpFeignClient;
private final PortableOffLogFeignClient portableOffLogFeignClient; private final PortableOffLogFeignClient portableOffLogFeignClient;
private final CommonProducer commonProducer;
private final SendMessageUtil sendMessageUtil;
public final static String UPLOAD_PATH = "upload";
@Override @Override
public void analysisFileInfo(AppFileMessage appFileMessage) { public void analysisFileInfo(AppFileMessage appFileMessage) {
if (Objects.equals(appFileMessage.getCode(), AccessEnum.SUCCESS.getCode())) { if (Objects.equals(appFileMessage.getCode(), AccessEnum.SUCCESS.getCode())) {
@@ -458,6 +466,332 @@ public class FileServiceImpl implements IFileService {
} }
} }
@Override
public List<FileInfoResponseDTO.ResourceElement> listDir(String filePath, String devId) {
List<CsEquipmentDeliveryDTO> listHttpResult = equipmentFeignClient.queryDeviceById(Collections.singletonList(devId)).getData();
FileInfoRequestDTO requestDTO = new FileInfoRequestDTO();
requestDTO.setGuid(IdUtil.simpleUUID());
requestDTO.setFrontId(listHttpResult.get(0).getNodeId());
requestDTO.setNode(listHttpResult.get(0).getNodeProcess());
requestDTO.setDevId(devId);
FileInfoRequestDTO.Detail detail = new FileInfoRequestDTO.Detail();
detail.setType(Integer.parseInt(TypeEnum.READ_FILE_DIR.getCode()));
FileInfoRequestDTO.Msg msg = new FileInfoRequestDTO.Msg();
msg.setName(filePath);
detail.setMsg(msg);
requestDTO.setDetail(detail);
BaseMessage message = new BaseMessage();
message.setMessageBody(JSON.toJSONString(requestDTO));
// 使用 Redis 存储 guid 用于后续查询
redisUtil.saveByKeyWithExpire(AppRedisKey.COMMON_REQUEST + requestDTO.getGuid(), "pending", 120L);
// 发送
commonProducer.send(message, requestDTO.getFrontId());
//this.simulation1(requestDTO.getGuid());
// 轮询 Redis 等待响应
FileInfoResponseDTO responseDTO = JSON.parseObject(sendMessageUtil.waitForResponse(requestDTO.getGuid(), 10), FileInfoResponseDTO.class);
FileInfoResponseDTO.Detail detail1 = responseDTO.getDetail();
FileInfoResponseDTO.Msg msg1 = detail1.getMsg();
msg1.getDirInfo().forEach(resourceElement -> {
resourceElement.setPrjDataPath(StrUtil.SLASH.equals(filePath) ? resourceElement.getName() : filePath + StrUtil.SLASH + resourceElement.getName());
});
return msg1.getDirInfo();
}
private void simulation1(String guid) {
// 模拟异步处理，实际场景中应由消息队列或回调触发
CompletableFuture.runAsync(() -> {
// 模拟耗时操作，例如等待设备响应
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
FileInfoResponseDTO message = new FileInfoResponseDTO();
List<FileInfoResponseDTO.ResourceElement> dirInfo = new ArrayList<>();
FileInfoResponseDTO.ResourceElement resourceElement1 = new FileInfoResponseDTO.ResourceElement();
resourceElement1.setName("/ram0");
resourceElement1.setType("dir");
resourceElement1.setSize(1);
dirInfo.add(resourceElement1);
FileInfoResponseDTO.ResourceElement resourceElement2 = new FileInfoResponseDTO.ResourceElement();
resourceElement2.setName("/etc");
resourceElement2.setType("dir");
resourceElement2.setSize(1);
dirInfo.add(resourceElement2);
FileInfoResponseDTO.ResourceElement resourceElement3 = new FileInfoResponseDTO.ResourceElement();
resourceElement3.setName("/sd0:1");
resourceElement3.setType("dir");
resourceElement3.setSize(1);
dirInfo.add(resourceElement3);
FileInfoResponseDTO.ResourceElement resourceElement4 = new FileInfoResponseDTO.ResourceElement();
resourceElement4.setName("1773986668375094.xls");
resourceElement4.setType("file");
resourceElement4.setSize(1000);
dirInfo.add(resourceElement4);
FileInfoResponseDTO.Detail detail = new FileInfoResponseDTO.Detail();
FileInfoResponseDTO.Msg msg = new FileInfoResponseDTO.Msg();
msg.setDirInfo(dirInfo);
detail.setMsg(msg);
detail.setCode(200);
detail.setType(4657);
detail.setMsg(msg);
message.setGuid(guid);
message.setFrontId("sdghsfdhfdhdfhdfghd234234534534");
message.setNode(1);
message.setDevMac("A0BC7B4A5D8A");
message.setDetail(detail);
BaseMessage baseMessage = new BaseMessage();
baseMessage.setSendTime(LocalDateTime.now());
baseMessage.setMessageBody(JSON.toJSONString(message));
commonProducer.send(baseMessage);
});
}
private void simulation2(String guid) {
// 模拟异步处理，实际场景中应由消息队列或回调触发
CompletableFuture.runAsync(() -> {
// 模拟耗时操作，例如等待设备响应
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
FileDownloadResponeDTO message = new FileDownloadResponeDTO();
message.setGuid(guid);
message.setFrontId("sdghsfdhfdhdfhdfghd234234534534");
message.setNode(1);
message.setDevMac("A0BC7B4A5D8A");
FileDownloadResponeDTO.Detail detail = new FileDownloadResponeDTO.Detail();
detail.setType(Integer.parseInt(TypeEnum.FILE_DOWNLOAD.getCode()));
FileDownloadResponeDTO.Msg msg = new FileDownloadResponeDTO.Msg();
msg.setName("a.txt");
msg.setRemoteName("https://www.jswsrc.com.cn/data/upload/ueditor/file/20260320/1773986668375094.xls");
detail.setMsg(msg);
message.setDetail(detail);
BaseMessage baseMessage = new BaseMessage();
baseMessage.setSendTime(LocalDateTime.now());
baseMessage.setMessageBody(JSON.toJSONString(message));
commonProducer.send(baseMessage);
});
}
@Override
public void downloadFileFromFront(String filePath, String devId, HttpServletResponse response) {
List<CsEquipmentDeliveryDTO> listHttpResult = equipmentFeignClient.queryDeviceById(Collections.singletonList(devId)).getData();
FileDownloadRequestDTO requestDTO = new FileDownloadRequestDTO();
requestDTO.setGuid(IdUtil.simpleUUID());
requestDTO.setFrontId(listHttpResult.get(0).getNodeId());
requestDTO.setNode(listHttpResult.get(0).getNodeProcess());
requestDTO.setDevId(devId);
FileDownloadRequestDTO.Detail detail = new FileDownloadRequestDTO.Detail();
detail.setType(Integer.parseInt(TypeEnum.FILE_DOWNLOAD.getCode()));
FileDownloadRequestDTO.Msg msg = new FileDownloadRequestDTO.Msg();
msg.setName(filePath);
detail.setMsg(msg);
requestDTO.setDetail(detail);
BaseMessage message = new BaseMessage();
message.setMessageBody(JSON.toJSONString(requestDTO));
// 使用 Redis 存储 guid 用于后续查询
redisUtil.saveByKeyWithExpire(AppRedisKey.COMMON_REQUEST + requestDTO.getGuid(), "pending", 120L);
// 发送
commonProducer.send(message, requestDTO.getFrontId());
//this.simulation2(requestDTO.getGuid());
// 轮询 Redis 等待响应
FileDownloadResponeDTO responseDTO = JSON.parseObject(sendMessageUtil.waitForResponse(requestDTO.getGuid(), 10), FileDownloadResponeDTO.class);
String remoteName = responseDTO.getDetail().getMsg().getRemoteName();
// String remoteName = "https://yunpan.360.cn/uploads/20230710/037ca576a421eb0bc23d717a7b076c5f.jpg";
// String remoteName = "/PQ_PQLD1_001429_20251010_143805_792.dat";
String fileName = remoteName.substring(remoteName.lastIndexOf(StrUtil.SLASH) + 1);
try {
response.setHeader("Content-Disposition", "attachment; filename=" + URLEncoder.encode(fileName, "UTF-8") + "\";filename*=UTF-8''" + URLEncoder.encode(fileName, "UTF-8"));
fileStorageUtil.downloadStream(response, remoteName);
// 下载完后删除文件
//fileStorageUtil.deleteFile(filePath);
} catch (Exception e) {
throw new BusinessException(AccessResponseEnum.FILE_DOWNLOAD_FAIL);
}
}
@Override
public boolean uploadFileToFront(MultipartFile file, String devId, String dirPath) {
dirPath = (dirPath.endsWith(StrUtil.SLASH) ? dirPath : dirPath + StrUtil.SLASH);
String remotePath = StrUtil.SLASH + UPLOAD_PATH + StrUtil.SLASH + devId + dirPath;
List<CsEquipmentDeliveryDTO> listHttpResult = equipmentFeignClient.queryDeviceById(Collections.singletonList(devId)).getData();
String frontId = listHttpResult.get(0).getNodeId();
try {
fileStorageUtil.uploadMultipart(file, remotePath, true);
} catch (Exception e) {
throw new BusinessException(AccessResponseEnum.FILE_UPLOAD_FAIL);
}
// 告诉前置上传文件所在的路径、设备等信息
FileUploadRequestDTO requestDTO = new FileUploadRequestDTO();
requestDTO.setGuid(IdUtil.simpleUUID());
requestDTO.setFrontId(frontId);
requestDTO.setNode(listHttpResult.get(0).getNodeProcess());
requestDTO.setDevId(devId);
FileUploadRequestDTO.Detail detail = new FileUploadRequestDTO.Detail();
detail.setType(Integer.parseInt(TypeEnum.FILE_UPLOAD.getCode()));
FileUploadRequestDTO.Msg msg = new FileUploadRequestDTO.Msg();
msg.setName(remotePath + file.getOriginalFilename());
msg.setRemoteName(dirPath + file.getOriginalFilename());
detail.setMsg(msg);
requestDTO.setDetail(detail);
BaseMessage message = new BaseMessage();
message.setMessageBody(JSON.toJSONString(requestDTO));
// 使用 Redis 存储 guid 用于后续查询
redisUtil.saveByKeyWithExpire(AppRedisKey.COMMON_REQUEST + requestDTO.getGuid(), "pending", 120L);
// 发送
commonProducer.send(message, requestDTO.getFrontId());
// 轮询 Redis 等待响应
FileUploadResponeDTO responseDTO = JSON.parseObject(sendMessageUtil.waitForResponse(requestDTO.getGuid(), 10), FileUploadResponeDTO.class);
FileUploadResponeDTO.Detail detail1 = responseDTO.getDetail();
return detail1.getCode()==200;
}
@Override
public boolean mkdir(String filePath, String devId) {
List<CsEquipmentDeliveryDTO> listHttpResult = equipmentFeignClient.queryDeviceById(Collections.singletonList(devId)).getData();
String frontId = listHttpResult.get(0).getNodeId();
// 告诉前置上传文件所在的路径、设备等信息
MkdirRequestDTO requestDTO = new MkdirRequestDTO();
requestDTO.setGuid(IdUtil.simpleUUID());
requestDTO.setFrontId(frontId);
requestDTO.setNode(listHttpResult.get(0).getNodeProcess());
requestDTO.setDevId(devId);
MkdirRequestDTO.Detail detail = new MkdirRequestDTO.Detail();
detail.setType(Integer.parseInt(TypeEnum.MKDIR.getCode()));
MkdirRequestDTO.Msg msg = new MkdirRequestDTO.Msg();
msg.setName(filePath.endsWith(StrUtil.SLASH) ? filePath : filePath + StrUtil.SLASH);
detail.setMsg(msg);
requestDTO.setDetail(detail);
BaseMessage message = new BaseMessage();
message.setMessageBody(JSON.toJSONString(requestDTO));
// 使用 Redis 存储 guid 用于后续查询
redisUtil.saveByKeyWithExpire(AppRedisKey.COMMON_REQUEST + requestDTO.getGuid(), "pending", 120L);
// 发送
commonProducer.send(message, requestDTO.getFrontId());
// 轮询 Redis 等待响应
MkdirResponeDTO responseDTO = JSON.parseObject(sendMessageUtil.waitForResponse(requestDTO.getGuid(), 10), MkdirResponeDTO.class);
MkdirResponeDTO.Detail detail1 = responseDTO.getDetail();
return detail1.getCode()==200;
}
@Override
public boolean delete(String filePath, String devId) {
boolean isDir = isDirectory(filePath);
List<CsEquipmentDeliveryDTO> listHttpResult = equipmentFeignClient.queryDeviceById(Collections.singletonList(devId)).getData();
String frontId = listHttpResult.get(0).getNodeId();
// 告诉前置上传文件所在的路径、设备等信息
FileOrDirDeleteRequestDTO requestDTO = new FileOrDirDeleteRequestDTO();
requestDTO.setGuid(IdUtil.simpleUUID());
requestDTO.setFrontId(frontId);
requestDTO.setNode(listHttpResult.get(0).getNodeProcess());
requestDTO.setDevId(devId);
FileOrDirDeleteRequestDTO.Detail detail = new FileOrDirDeleteRequestDTO.Detail();
detail.setType(Integer.parseInt(isDir ? TypeEnum.DIR_DELETE.getCode() : TypeEnum.FILE_DELETE.getCode()));
FileOrDirDeleteRequestDTO.Msg msg = new FileOrDirDeleteRequestDTO.Msg();
msg.setName(isDir ? (filePath.endsWith(StrUtil.SLASH) ? filePath : filePath + StrUtil.SLASH) : filePath);
detail.setMsg(msg);
requestDTO.setDetail(detail);
BaseMessage message = new BaseMessage();
message.setMessageBody(JSON.toJSONString(requestDTO));
// 使用 Redis 存储 guid 用于后续查询
redisUtil.saveByKeyWithExpire(AppRedisKey.COMMON_REQUEST + requestDTO.getGuid(), "pending", 120L);
// 发送
commonProducer.send(message, requestDTO.getFrontId());
// 轮询 Redis 等待响应
FileOrDirDeleteResponeDTO responseDTO = JSON.parseObject(sendMessageUtil.waitForResponse(requestDTO.getGuid(), 10), FileOrDirDeleteResponeDTO.class);
FileOrDirDeleteResponeDTO.Detail detail1 = responseDTO.getDetail();
return detail1.getCode()==200;
}
/**
* 根据文件扩展名判断是文件还是目录
* 规则：包含小数点的是文件，否则是目录
*
* @param filePath 文件路径
* @return true=目录，false=文件
*/
private boolean isDirectory(String filePath) {
String fileName = filePath.substring(filePath.lastIndexOf("/") + 1);
// 判断文件名中是否包含小数点（从第二个字符开始，排除隐藏文件的情况）
// 例如：.gitignore 是文件，不是目录
if (fileName.startsWith(".")) {
// 隐藏文件/目录：判断除第一个点外是否还有其他点
return !fileName.substring(1).contains(".");
} else {
// 普通文件/目录：直接判断是否包含点
return !fileName.contains(".");
}
}
/** /**
* 处理补召文件 * 处理补召文件
*/ */
@@ -680,6 +1014,7 @@ public class FileServiceImpl implements IFileService {
public MyObjectOutputStream(OutputStream out) throws IOException { public MyObjectOutputStream(OutputStream out) throws IOException {
super(out); super(out);
} }
@Override @Override
protected void writeStreamHeader() throws IOException { protected void writeStreamHeader() throws IOException {
//重写读取头部信息方法：不写入头部信息 //重写读取头部信息方法：不写入头部信息
@@ -691,6 +1026,7 @@ public class FileServiceImpl implements IFileService {
public MyObjectInputStream(InputStream in) throws IOException { public MyObjectInputStream(InputStream in) throws IOException {
super(in); super(in);
} }
@Override @Override
protected void readStreamHeader() throws IOException { protected void readStreamHeader() throws IOException {
//重写读取头部信息方法：什么也不做 //重写读取头部信息方法：什么也不做

33
iot-analysis/analysis-zl-event/zl-event-boot/src/test/java/com/njcn/zlevent/FileDownloadTest.java Normal file
View File

@@ -0,0 +1,33 @@
package com.njcn.zlevent;
import org.springframework.boot.test.context.SpringBootTest;
/**
* @author caozehui
* @data 2026-03-17
*/
@SpringBootTest
public class FileDownloadTest {
// @Resource
// private FileDownloadProducer fileDownloadProducer;
//
// @Test
// public void testSend() {
// BaseRequestDTO<FileDownloadRequestDTO> message = new BaseRequestDTO<>();
// message.setGuid(IdUtil.simpleUUID());
// message.setNode(1);
// message.setDevId("167456737637374567");
// message.setFrontId("dhdfhdfghd2342");
// // 设置 detail（重要！）
// BaseRequestDTO.Detail<FileDownloadRequestDTO> detail = new BaseRequestDTO.Detail<>();
// detail.setType(8498); // 设置类型
// FileDownloadRequestDTO fileDownloadRequestDTO = new FileDownloadRequestDTO();
// fileDownloadRequestDTO.setName("/etc/vol1_stat.txt");
// detail.setMsg(fileDownloadRequestDTO); // 设置消息体
// message.setDetail(detail);
// SendResult send = fileDownloadProducer.send(message);
// System.out.println(JSON.toJSON(send));
// }
}

5
iot-message/message-boot/src/main/java/com/njcn/message/consumer/AppAutoDataConsumer.java
View File

@@ -86,11 +86,12 @@ public class AppAutoDataConsumer extends EnhanceConsumerMessageHandler<AppAutoDa
} }
/** /**
* 消费成功，缓存到redis72小时，避免重复消费 * 消费成功，缓存到redis 5分钟，避免重复消费
*/ */
@Override @Override
protected void consumeSuccess(AppAutoDataMessage message) { protected void consumeSuccess(AppAutoDataMessage message) {
redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime()); // redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, 5 * 60L);
} }
/** /**

5
iot-message/message-boot/src/main/java/com/njcn/message/consumer/CldHeartConsumer.java
View File

@@ -68,11 +68,12 @@ public class CldHeartConsumer extends EnhanceConsumerMessageHandler<CldHeartBeat
} }
/** /**
* 消费成功，缓存到redis72小时，避免重复消费 * 消费成功，缓存到redis 5分钟，避免重复消费
*/ */
@Override @Override
protected void consumeSuccess(CldHeartBeatMessage message) { protected void consumeSuccess(CldHeartBeatMessage message) {
redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime()); // redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, 5 * 60L);
} }
/** /**

133
iot-message/message-boot/src/main/java/com/njcn/message/consumer/CommonConsumer.java Normal file
View File

@@ -0,0 +1,133 @@
package com.njcn.message.consumer;
import com.alibaba.fastjson.JSON;
import com.njcn.middle.rocket.constant.EnhanceMessageConstant;
import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler;
import com.njcn.mq.constant.BusinessTopic;
import com.njcn.mq.constant.MessageStatus;
import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.pojo.enums.RedisKeyEnum;
import com.njcn.redis.utils.RedisUtil;
import com.njcn.system.api.RocketMqLogFeignClient;
import com.njcn.system.pojo.po.RocketmqMsgErrorLog;
import com.njcn.zlevent.pojo.dto.CommonBaseMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.Objects;
/**
* @author caozehui
* @data 2026-03-18
*/
@Service
@RocketMQMessageListener(
topic = BusinessTopic.CLOUD_REPLY_TOPIC,
consumerGroup = BusinessTopic.CLOUD_REPLY_TOPIC,
consumeThreadNumber = 10,
enableMsgTrace = true
)
@Slf4j
public class CommonConsumer extends EnhanceConsumerMessageHandler<CommonBaseMessage> implements RocketMQListener<CommonBaseMessage> {
@Resource
private RedisUtil redisUtil;
@Resource
private RocketMqLogFeignClient rocketMqLogFeignClient;
@Override
protected void handleMessage(CommonBaseMessage message) throws Exception {
log.info("@@@@@处理Common信息");
System.out.println(JSON.toJSON(message));
String guid = message.getGuid();
// 将响应结果存入 Redis
redisUtil.saveByKeyWithExpire(AppRedisKey.COMMON_RESOPNSE + guid, JSON.toJSONString(message), 120L);
}
/***
* 通过redis分布式锁判断当前消息所处状态
* 1、null 查不到该key的数据，属于第一次消费，放行
* 2、fail 上次消息消费时发生异常，放行
* 3、being processed 正在处理，打回去
* 4、success 最近72小时消费成功，避免重复消费，打回去
*/
@Override
protected boolean filter(CommonBaseMessage message) {
String keyStatus = redisUtil.getStringByKey(AppRedisKey.RMQ_FILE_CONSUME_KEY.concat(message.getKey()));
if (Objects.isNull(keyStatus) || keyStatus.equalsIgnoreCase(MessageStatus.FAIL)) {
redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_FILE_CONSUME_KEY.concat(message.getKey()), MessageStatus.BEING_PROCESSED, 60L);
return false;
}
return true;
}
/**
* 消费成功，缓存到redis72小时，避免重复消费
*/
@Override
protected void consumeSuccess(CommonBaseMessage message) {
redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_FILE_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
}
/**
* 发生异常时，进行错误信息入库保存
* 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存
*/
@Override
protected void saveExceptionMsgLog(CommonBaseMessage message, String identity, Exception exception) {
redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_FILE_CONSUME_KEY.concat(message.getKey()), MessageStatus.FAIL, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
RocketmqMsgErrorLog rocketmqMsgErrorLog = new RocketmqMsgErrorLog();
rocketmqMsgErrorLog.setMsgKey(message.getKey());
rocketmqMsgErrorLog.setResource(message.getSource());
if (identity.equalsIgnoreCase(EnhanceMessageConstant.IDENTITY_SINGLE)) {
//数据库字段配置长度200，避免插入失败，大致分析异常原因
String exceptionMsg = exception.getMessage();
if (exceptionMsg.length() > 200) {
exceptionMsg = exceptionMsg.substring(0, 180);
}
rocketmqMsgErrorLog.setRecord(exceptionMsg);
//如果是当前消息重试的则略过
if (!message.getSource().startsWith(EnhanceMessageConstant.RETRY_PREFIX)) {
//单次消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
} else {
rocketmqMsgErrorLog.setRecord("重试消费" + super.getMaxRetryTimes() + "次，依旧消费失败。");
//重试N次后，依然消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
}
/***
* 处理失败后，是否重试
* 一般开启
*/
@Override
protected boolean isRetry() {
return true;
}
/***
* 消费失败是否抛出异常，抛出异常后就不再消费了
*/
@Override
protected boolean throwException() {
return false;
}
/***
* 调用父类handler处理消息的元信息
*/
@Override
public void onMessage(CommonBaseMessage message) {
super.dispatchMessage(message);
}
}

4
iot-message/message-boot/src/main/java/com/njcn/message/consumer/RealDataConsumer.java
View File

@@ -71,11 +71,11 @@ public class RealDataConsumer extends EnhanceConsumerMessageHandler<AppAutoDataM
} }
/** /**
* 消费成功，缓存到redis72小时，避免重复消费 * 消费成功，缓存到redis 5分钟，避免重复消费
*/ */
@Override @Override
protected void consumeSuccess(AppAutoDataMessage message) { protected void consumeSuccess(AppAutoDataMessage message) {
redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime()); redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, 5 * 60L);
} }
/** /**