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

新增云协议设备实时数据功能

Browse Source
This commit is contained in:
xy
2025-09-04 14:00:15 +08:00
parent f88baa52be
commit 1a3e443be1
15 changed files with 398 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,15 +1,8 @@
package com.njcn.access.api;
import com.njcn.access.api.fallback.AskDeviceDataClientFallbackFactory;
import com.njcn.common.pojo.annotation.OperateInfo;
import com.njcn.common.pojo.constant.ServerInfo;
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 io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
@@ -41,4 +34,7 @@ public interface AskDeviceDataFeignClient {
@PostMapping("/askRealData")
HttpResult<String> askRealData(@RequestParam("nDid") String nDid, @RequestParam("idx") Integer idx, @RequestParam("clDId") Integer clDId);
@PostMapping("/askCldRealData")
HttpResult<String> askCldRealData(@RequestParam("devId") String devId, @RequestParam("lineId") String lineId);
}

6
iot-access/access-api/src/main/java/com/njcn/access/api/fallback/AskDeviceDataClientFallbackFactory.java
View File

@@ -73,6 +73,12 @@ public class AskDeviceDataClientFallbackFactory implements FallbackFactory<AskDe
log.error("{}异常，降级处理，异常为：{}","询问装置实时数据",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
@Override
public HttpResult<String> askCldRealData(String devId, String lineId) {
log.error("{}异常，降级处理，异常为：{}","询问云前置实时数据",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
};
}
}

68
iot-access/access-api/src/main/java/com/njcn/access/utils/RedisSetUtil.java Normal file
View File

@@ -0,0 +1,68 @@
package com.njcn.access.utils;
import com.njcn.redis.utils.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
/**
* @author 徐扬
*/
@Component
@Slf4j
public class RedisSetUtil {
@Autowired
private RedisUtil redisUtil;
/**
* 向Redis Set中添加元素
*/
public void addToSet(String key, String value, long expireSeconds) {
try {
Object existing = redisUtil.getObjectByKey(key);
Set<String> set = convertToSet(existing);
set.add(value);
redisUtil.saveByKeyWithExpire(key, set, expireSeconds);
} catch (Exception e) {
log.error("向Redis Set添加元素失败，key: {}", key, e);
}
}
/**
* 从Redis Set中移除元素
*/
public void removeFromSet(String key, String value) {
try {
Object existing = redisUtil.getObjectByKey(key);
if (existing != null) {
Set<String> set = convertToSet(existing);
set.remove(value);
redisUtil.saveByKey(key, set);
}
} catch (Exception e) {
log.error("从Redis Set移除元素失败，key: {}", key, e);
}
}
/**
* 安全的对象到Set转换
*/
public Set<String> convertToSet(Object obj) {
if (obj == null) {
return new HashSet<>();
}
if (obj instanceof Set) {
return new HashSet<>((Set<String>) obj);
}
if (obj instanceof Collection) {
return new HashSet<>((Collection<String>) obj);
}
log.warn("无法转换的对象类型: {}", obj.getClass().getName());
return new HashSet<>();
}
}

12
iot-access/access-boot/src/main/java/com/njcn/access/controller/AskDeviceDataController.java
View File

@@ -123,5 +123,17 @@ public class AskDeviceDataController extends BaseController {
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/askCldRealData")
@ApiOperation("询问云前置实时数据")
@ApiImplicitParams({
@ApiImplicitParam(name = "devId", value = "装置id"),
@ApiImplicitParam(name = "lineId", value = "监测点id")
})
public HttpResult<String> askCldRealData(@RequestParam("devId") String devId, @RequestParam("lineId") String lineId){
String methodDescribe = getMethodDescribe("askCldRealData");
askDeviceDataService.askCldRealData(devId,lineId);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
}
}

95
iot-access/access-boot/src/main/java/com/njcn/access/listener/RedisKeyExpirationListener.java
View File

@@ -1,14 +1,17 @@
package com.njcn.access.listener;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.github.tocrhz.mqtt.publisher.MqttPublisher;
import com.njcn.access.enums.AccessEnum;
import com.njcn.access.pojo.dto.NoticeUserDto;
import com.njcn.access.pojo.po.CsDeviceOnlineLogs;
import com.njcn.access.service.ICsDeviceOnlineLogsService;
import com.njcn.access.service.ICsEquipmentDeliveryService;
import com.njcn.access.service.ICsTopicService;
import com.njcn.access.service.impl.CsDeviceServiceImpl;
import com.njcn.access.utils.MqttUtil;
import com.njcn.access.utils.RedisSetUtil;
import com.njcn.access.utils.SendMessageUtil;
import com.njcn.common.pojo.dto.DeviceLogDTO;
import com.njcn.csdevice.api.*;
@@ -17,6 +20,7 @@ import com.njcn.csdevice.pojo.dto.PqsCommunicateDto;
import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.utils.RedisUtil;
import com.njcn.rt.pojo.dto.BaseRealDataSet;
import com.njcn.user.api.AppUserFeignClient;
import com.njcn.user.api.UserFeignClient;
import com.njcn.user.pojo.po.User;
@@ -32,6 +36,7 @@ import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@@ -46,8 +51,6 @@ import java.util.stream.Collectors;
@Component
public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
@Resource
private ICsTopicService csTopicService;
@Resource
private ICsEquipmentDeliveryService csEquipmentDeliveryService;
@Resource
@@ -73,9 +76,11 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
@Resource
private SendMessageUtil sendMessageUtil;
@Resource
private CsDeviceServiceImpl csDeviceServiceImpl;
@Resource
private CsCommunicateFeignClient csCommunicateFeignClient;
@Resource
private MqttPublisher publisher;
@Resource
private RedisSetUtil redisSetUtil;
private final Object lock = new Object();
@@ -101,6 +106,20 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
String nDid = expiredKey.split(":")[1];
executeMainTask(nDid);
}
if(expiredKey.startsWith("cldRtDataOverTime:")){
String lineId = expiredKey.split(":")[1];
Object redisObject = redisUtil.getObjectByKey("rtDataUserId:"+lineId);
if (ObjectUtil.isNotNull(redisObject)) {
Set<String> userSet = redisSetUtil.convertToSet(redisObject);
userSet.forEach(userId->{
BaseRealDataSet baseRealDataSet = new BaseRealDataSet();
baseRealDataSet.setUserId(userId);
baseRealDataSet.setResult(false);
baseRealDataSet.setContent("设备未响应，超时中断");
publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false);
});
}
}
}
//主任务
@@ -118,10 +137,6 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
logDto.setOperate(nDid +"装置离线");
sendMessage(nDid);
//记录装置掉线时间
// CsDeviceOnlineLogs record = onlineLogsService.findLastData(nDid);
// record.setOfflineTime(LocalDateTime.now());
// onlineLogsService.updateById(record);
//记录装置掉线时间
PqsCommunicateDto dto = new PqsCommunicateDto();
dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
dto.setDevId(nDid);
@@ -131,68 +146,6 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
csLogsFeignClient.addUserLog(logDto);
}
//主任务
//1.装置心跳断连
//2.MQTT客户端不在线
// private void executeMainTask(String nDid, String version) {
// ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
// log.info("{}->装置离线", nDid);
// DeviceLogDTO logDto = new DeviceLogDTO();
// logDto.setUserName("运维管理员");
// logDto.setLoginName("njcnyw");
// //判断mqtt
// String clientName = "NJCN-" + nDid.substring(nDid.length() - 6);
// boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
// //心跳异常，但是客户端在线，则发送接入请求
// if (mqttClient) {
// csDeviceService.devAccessAskTemplate(nDid,version,1);
// redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,1);
// try {
// Thread.sleep(5000);
// Object object = redisUtil.getObjectByKey("online" + nDid);
// if (Objects.nonNull(object)) {
// scheduler.shutdown();
// logDto.setOperate(nDid + "客户端在线重连成功");
// } else {
// //装置下线
// csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
// //装置调整为注册状态
// csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode());
// logDto.setOperate(nDid +"装置离线");
// sendMessage(nDid);
//
// //记录装置掉线时间
// CsDeviceOnlineLogs record = onlineLogsService.findLastData(nDid);
// record.setOfflineTime(LocalDateTime.now());
// onlineLogsService.updateById(record);
//
// scheduler.shutdown();
// }
// } catch (InterruptedException e) {
// scheduler.shutdown();
// throw new RuntimeException(e);
// }
// csLogsFeignClient.addUserLog(logDto);
// }
// //客户端不在线则修改装置状态，进入定时任务
// else {
// //装置下线
// csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
// //装置调整为注册状态
// csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode());
// logDto.setOperate(nDid +"主任务执行失败，装置下线");
// csLogsFeignClient.addUserLog(logDto);
// sendMessage(nDid);
//
// //记录装置掉线时间
// CsDeviceOnlineLogs record = onlineLogsService.findLastData(nDid);
// record.setOfflineTime(LocalDateTime.now());
// onlineLogsService.updateById(record);
//
// scheduler.shutdown();
// }
// }
private void startScheduledTask(ScheduledExecutorService scheduler, String nDid, String version) {
synchronized (lock) {
//判断是否推送消息

3
iot-access/access-boot/src/main/java/com/njcn/access/runner/AutoAccessTimer.java
View File

@@ -203,8 +203,7 @@ public class AutoAccessTimer implements ApplicationRunner {
// }
// };
// //第一次执行的时间为120s，然后在前一个任务执行完毕后，等待120s再执行下一个任务
//// scheduler.scheduleWithFixedDelay(task, AUTO_TIME, AUTO_TIME, TimeUnit.SECONDS);
// scheduler.scheduleWithFixedDelay(task, 10, 10, TimeUnit.SECONDS);
// scheduler.scheduleWithFixedDelay(task, AUTO_TIME, AUTO_TIME, TimeUnit.SECONDS);
// }
//
// public void accessDev(List<CsEquipmentDeliveryPO> list) {

2
iot-access/access-boot/src/main/java/com/njcn/access/service/AskDeviceDataService.java
View File

@@ -21,4 +21,6 @@ public interface AskDeviceDataService {
* 实时数据请求报文
*/
void askRealData(String nDid, Integer idx, Integer size);
void askCldRealData(String devId, String lineId);
}

13
iot-access/access-boot/src/main/java/com/njcn/access/service/impl/AskDeviceDataServiceImpl.java
View File

@@ -11,6 +11,8 @@ import com.njcn.access.pojo.dto.ReqAndResDto;
import com.njcn.access.service.AskDeviceDataService;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.csdevice.enums.AlgorithmResponseEnum;
import com.njcn.mq.message.RealDataMessage;
import com.njcn.mq.template.RealDataMessageTemplate;
import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.utils.RedisUtil;
import lombok.RequiredArgsConstructor;
@@ -32,6 +34,7 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
private final MqttPublisher publisher;
private final CsTopicFeignClient csTopicFeignClient;
private final RedisUtil redisUtil;
private final RealDataMessageTemplate realDataMessageTemplate;
private static Integer mid = 1;
private static Integer range = 51200;
@@ -209,6 +212,16 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,mid);
}
@Override
public void askCldRealData(String devId, String lineId) {
RealDataMessage realDataMessage = new RealDataMessage();
realDataMessage.setDevSeries(devId);
realDataMessage.setLine(lineId);
realDataMessage.setRealData(true);
realDataMessage.setSoeData(true);
realDataMessage.setLimit(20);
realDataMessageTemplate.sendMember(realDataMessage);
}
public Object getDeviceMid(String nDid) {
return redisUtil.getObjectByKey(AppRedisKey.DEVICE_MID + nDid);

2
iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDeviceServiceImpl.java
View File

@@ -726,7 +726,7 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
}
}
publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(getJson(mid,TypeEnum.TYPE_5.getCode())), 1, false);
redisUtil.saveByKeyWithExpire("startFile:" + nDid, null, 60L);
// redisUtil.saveByKeyWithExpire("startFile:" + nDid, null, 60L);
result = true;
} catch (Exception e) {
DeviceLogDTO logDto = new DeviceLogDTO();

6
iot-analysis/analysis-rt/rt-api/src/main/java/com/njcn/rt/pojo/dto/BaseRealDataSet.java
View File

@@ -14,6 +14,12 @@ public class BaseRealDataSet implements Serializable {
@ApiModelProperty("用户ID")
private String userId;
@ApiModelProperty("结果(仅超时使用)")
private boolean result = true;
@ApiModelProperty("描述")
private String content;
@ApiModelProperty("监测点id")
private String lineId;

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

@@ -1,9 +1,11 @@
package com.njcn.rt.service.impl;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.github.tocrhz.mqtt.publisher.MqttPublisher;
import com.njcn.access.utils.ChannelObjectUtil;
import com.njcn.access.utils.RedisSetUtil;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.common.utils.PubUtils;
import com.njcn.csdevice.api.CsLineFeignClient;
@@ -21,6 +23,7 @@ import com.njcn.rt.pojo.dto.HarmRealDataSet;
import com.njcn.rt.service.IRtService;
import com.njcn.web.utils.FloatUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.lang.reflect.Field;
@@ -36,6 +39,7 @@ import java.util.stream.Collectors;
/**
* @author xy
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class RtServiceImpl implements IRtService {
@@ -46,14 +50,14 @@ public class RtServiceImpl implements IRtService {
private final RedisUtil redisUtil;
private final ChannelObjectUtil channelObjectUtil;
private final MqttPublisher publisher;
private final RedisSetUtil redisSetUtil;
@Override
public void analysis(AppAutoDataMessage appAutoDataMessage) {
List<CsDataArray> dataArrayList;
//监测点id
String lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
//用户Id
String userId = redisUtil.getObjectByKey("rtDataUserId:"+lineId).toString();
redisUtil.delete("cldRtDataOverTime:"+lineId);
//获取监测点基础信息
CsLinePO po = csLineFeignClient.getById(lineId).getData();
//获取数据集 dataSet
@@ -73,6 +77,8 @@ public class RtServiceImpl implements IRtService {
//fixme 这边先根据数据集的名称来返回对应实体，这边感觉不太合适，后期有好方案再调整
//基础数据
if (dataSet.getName().contains("Ds$Pqd$Rt$Basic$")) {
//用户Id
String userId = redisUtil.getObjectByKey("rtDataUserId:"+lineId).toString();
BaseRealDataSet baseRealDataSet = assembleData(dataArrayList,item,po.getConType());
baseRealDataSet.setUserId(userId);
baseRealDataSet.setLineId(lineId);
@@ -82,9 +88,28 @@ public class RtServiceImpl implements IRtService {
long timestamp = item.getDataTimeSec() - 8*3600;
baseRealDataSet.setDataTime(getTime(timestamp));
publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false);
} else if (dataSet.getName().contains("实时数据集合")) {
//用户Id
Object redisObject = redisUtil.getObjectByKey("rtDataUserId:"+lineId);
if (ObjectUtil.isNotNull(redisObject)) {
Set<String> userSet = redisSetUtil.convertToSet(redisObject);
userSet.forEach(userId->{
BaseRealDataSet baseRealDataSet = assembleData(dataArrayList,item,po.getConType());
baseRealDataSet.setUserId(userId);
baseRealDataSet.setLineId(lineId);
baseRealDataSet.setPt(po.getPtRatio().floatValue());
baseRealDataSet.setCt(po.getCtRatio().floatValue());
baseRealDataSet.setDataLevel(dataSet.getDataLevel());
long timestamp = item.getDataTimeSec();
baseRealDataSet.setDataTime(getTime(timestamp));
publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false);
});
}
}
//fixme 目前实时数据只有基础数据和谐波数据，后期拓展，这边需要再判断
else {
//用户Id
String userId = redisUtil.getObjectByKey("rtDataUserId:"+lineId).toString();
HarmRealDataSet harmRealDataSet = harmData(dataArrayList,item,dataSet.getDataLevel(),po.getCtRatio());
harmRealDataSet.setUserId(userId);
harmRealDataSet.setLineId(lineId);
@@ -322,4 +347,26 @@ public class RtServiceImpl implements IRtService {
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<>();
}
}
}

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

@@ -33,7 +33,6 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
@@ -85,7 +84,7 @@ public class StatServiceImpl implements IStatService {
List<CsEquipmentDeliveryPO> poList = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DEVICE_LIST),CsEquipmentDeliveryPO.class);
CsEquipmentDeliveryPO po = poList.stream().filter(item->Objects.equals(item.getNdid(),appAutoDataMessage.getId())).findFirst().orElse(null);
List<SysDicTreePO> dictTreeList = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DICT_TREE),SysDicTreePO.class);
String code = dictTreeList.stream().filter(item->Objects.equals(item.getId(),po.getDevType())).findFirst().orElse(null).getCode();
String code = Objects.requireNonNull(dictTreeList.stream().filter(item -> Objects.equals(item.getId(), po.getDevType())).findFirst().orElse(null)).getCode();
//便携式设备
if (Objects.equals(DicDataEnum.PORTABLE.getCode(),code)) {
@@ -93,13 +92,19 @@ public class StatServiceImpl implements IStatService {
}
//直连设备
else if (Objects.equals(DicDataEnum.CONNECT_DEV.getCode(),code)) {
if (Objects.equals(appAutoDataMessage.getDid(),1)){lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appAutoDataMessage.getId())), Map.class).get("0").toString();
} else if (Objects.equals(appAutoDataMessage.getDid(),2)){
if (Objects.equals(appAutoDataMessage.getDid(),1)) {
lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appAutoDataMessage.getId())), Map.class).get("0").toString();
} else if (Objects.equals(appAutoDataMessage.getDid(),2)) {
lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appAutoDataMessage.getId())), Map.class).get(appAutoDataMessage.getMsg().getClDid().toString()).toString();
}
}
//云前置设备
else if (Objects.equals(DicDataEnum.DEV_CLD.getCode(),code)) {
lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
}
//获取当前设备信息
if (CollectionUtil.isNotEmpty(list)){
if (CollectionUtil.isNotEmpty(list)) {
List<String> recordList = new ArrayList<>();
for (AppAutoDataMessage.DataArray item : list) {
switch (item.getDataAttr()) {

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

@@ -3,6 +3,7 @@ package com.njcn.zlevent.service.impl;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.IdUtil;
import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.csdevice.api.CsLineFeignClient;
import com.njcn.csdevice.api.EquipmentFeignClient;
@@ -92,7 +93,6 @@ public class EventServiceImpl implements IEventService {
//获取设备类型 true:治理设备 false:其他类型的设备
boolean devModel = equipmentFeignClient.judgeDevModel(appEventMessage.getId()).getData();
try {
// lineId = appEventMessage.getId() + appEventMessage.getMsg().getClDid();
if (devModel) {
if (Objects.equals(appEventMessage.getDid(),1)){
lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get("0").toString();
@@ -103,74 +103,81 @@ public class EventServiceImpl implements IEventService {
lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get(appEventMessage.getMsg().getClDid().toString()).toString();
}
//处理事件数据
List<AppEventMessage.DataArray> dataArray = appEventMessage.getMsg().getDataArray();
for (AppEventMessage.DataArray item : dataArray) {
eventTime = timeFormat(item.getDataTimeSec(),item.getDataTimeUSec());
id = IdUtil.fastSimpleUUID();
//事件入库
CsEventPO csEvent = new CsEventPO();
csEvent.setId(id);
csEvent.setDeviceId(po.getId());
csEvent.setProcess(po.getProcess());
csEvent.setCode(item.getCode());
eventTime = timeFormat(item.getDataTimeSec(),item.getDataTimeUSec());
csEvent.setStartTime(eventTime);
tag = item.getName();
csEvent.setTag(tag);
if (Objects.equals(item.getType(),"2")){
csEvent.setType(0);
} else if (Objects.equals(item.getType(),"3")){
csEvent.setType(1);
} else if (Objects.equals(item.getType(),"1")){
csEvent.setType(2);
}
csEvent.setLevel(Integer.parseInt(item.getType()));
csEvent.setClDid(appEventMessage.getMsg().getClDid());
csEvent.setLineId(lineId);
//参数入库
Map<String,String> map = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.ELE_EPD_PQD)), Map.class);
//判断是否有参数
List<AppEventMessage.Param> params = item.getParam();
if (CollectionUtil.isNotEmpty(params)) {
String tableName = map.get(item.getName());
Map<String, String> tags = new HashMap<>();
tags.put(InfluxDBTableConstant.UUID,id);
Map<String,Object> fields = new HashMap<>();
for (AppEventMessage.Param param : params) {
if (Objects.equals(param.getName(),ZlConstant.EVT_PARAM_TM)){
csEvent.setPersistTime(Double.parseDouble(param.getData().toString()));
}
fields.put(param.getName(),param.getData());
//判断事件是否存在，如果存在则不处理
LambdaQueryWrapper<CsEventPO> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(CsEventPO::getDeviceId,po.getId())
.eq(CsEventPO::getTag,tag)
.eq(CsEventPO::getStartTime,eventTime)
.eq(CsEventPO::getLineId,lineId);
List<CsEventPO> eventList = csEventService.list(queryWrapper);
if (CollectionUtil.isEmpty(eventList)) {
id = IdUtil.fastSimpleUUID();
//事件入库
CsEventPO csEvent = new CsEventPO();
csEvent.setId(id);
csEvent.setDeviceId(po.getId());
csEvent.setProcess(po.getProcess());
csEvent.setCode(item.getCode());
eventTime = timeFormat(item.getDataTimeSec(),item.getDataTimeUSec());
csEvent.setStartTime(eventTime);
tag = item.getName();
csEvent.setTag(tag);
if (Objects.equals(item.getType(),"2")){
csEvent.setType(0);
} else if (Objects.equals(item.getType(),"3")){
csEvent.setType(1);
} else if (Objects.equals(item.getType(),"1")){
csEvent.setType(2);
}
//只有治理型号的设备有监测位置
if (devModel) {
if (appEventMessage.getMsg().getClDid() == 1) {
fields.put("Evt_Param_Position","电网侧");
csEvent.setLocation("grid");
} else if (appEventMessage.getMsg().getClDid() == 2) {
fields.put("Evt_Param_Position","负载侧");
csEvent.setLocation("load");
csEvent.setLevel(Integer.parseInt(item.getType()));
csEvent.setClDid(appEventMessage.getMsg().getClDid());
csEvent.setLineId(lineId);
//参数入库
Map<String,String> map = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.ELE_EPD_PQD)), Map.class);
//判断是否有参数
List<AppEventMessage.Param> params = item.getParam();
if (CollectionUtil.isNotEmpty(params)) {
String tableName = map.get(item.getName());
Map<String, String> tags = new HashMap<>();
tags.put(InfluxDBTableConstant.UUID,id);
Map<String,Object> fields = new HashMap<>();
for (AppEventMessage.Param param : params) {
if (Objects.equals(param.getName(),ZlConstant.EVT_PARAM_TM)){
csEvent.setPersistTime(Double.parseDouble(param.getData().toString()));
}
fields.put(param.getName(),param.getData());
}
//只有治理型号的设备有监测位置
if (devModel) {
if (appEventMessage.getMsg().getClDid() == 1) {
fields.put("Evt_Param_Position","电网侧");
csEvent.setLocation("grid");
} else if (appEventMessage.getMsg().getClDid() == 2) {
fields.put("Evt_Param_Position","负载侧");
csEvent.setLocation("load");
}
}
//fixme 这边前置传递的应该是UTC时间,但是前置说是传递的北京时间,讨论了一下没太理解。这边暂时先这样处理,influx入库处理成北京时间,减去8小时。
Point point = influxDbUtils.pointBuilder(tableName, item.getDataTimeSec()-8*3600, TimeUnit.SECONDS, tags, fields);
BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();
batchPoints.point(point);
records.add(batchPoints.lineProtocol());
}
//fixme 这边前置传递的应该是UTC时间,但是前置说是传递的北京时间,讨论了一下没太理解。这边暂时先这样处理,influx入库处理成北京时间,减去8小时。
Point point = influxDbUtils.pointBuilder(tableName, item.getDataTimeSec()-8*3600, TimeUnit.SECONDS, tags, fields);
BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();
batchPoints.point(point);
records.add(batchPoints.lineProtocol());
list1.add(csEvent);
//事件处理日志库
CsEventLogs csEventLogs = new CsEventLogs();
csEventLogs.setLineId(lineId);
csEventLogs.setDeviceId(po.getId());
csEventLogs.setStartTime(timeFormat(item.getDataTimeSec(),item.getDataTimeUSec()));
csEventLogs.setTag(item.getName());
csEventLogs.setStatus(1);
csEventLogs.setTime(LocalDateTime.now());
csEventLogsService.save(csEventLogs);
}
list1.add(csEvent);
//事件处理日志库
CsEventLogs csEventLogs = new CsEventLogs();
csEventLogs.setLineId(lineId);
csEventLogs.setDeviceId(po.getId());
csEventLogs.setStartTime(timeFormat(item.getDataTimeSec(),item.getDataTimeUSec()));
csEventLogs.setTag(item.getName());
csEventLogs.setStatus(1);
csEventLogs.setTime(LocalDateTime.now());
csEventLogsService.save(csEventLogs);
}
//cs_event入库
if (CollectionUtil.isNotEmpty(list1)){

1
iot-message/message-boot/src/main/java/com/njcn/message/MessageBootApplication.java
View File

@@ -15,7 +15,6 @@ import org.springframework.context.annotation.DependsOn;
@Slf4j
@EnableFeignClients(basePackages = "com.njcn")
@SpringBootApplication(scanBasePackages = "com.njcn")
@DependsOn("proxyMapperRegister")
public class MessageBootApplication {
public static void main(String[] args) {

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

@@ -0,0 +1,135 @@
package com.njcn.message.consumer;
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.mq.message.AppAutoDataMessage;
import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.pojo.enums.RedisKeyEnum;
import com.njcn.redis.utils.RedisUtil;
import com.njcn.rt.api.RtFeignClient;
import com.njcn.system.api.RocketMqLogFeignClient;
import com.njcn.system.pojo.po.RocketmqMsgErrorLog;
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 xuyang
* @version 1.0.0
* @createTime 2023/8/11 15:32
*/
@Service
@RocketMQMessageListener(
topic = BusinessTopic.CLD_HANDLE_REAL_DATA_TOPIC,
consumerGroup = BusinessTopic.CLD_HANDLE_REAL_DATA_TOPIC,
selectorExpression = BusinessTopic.AppDataTag.RT_TAG,
consumeThreadNumber = 10,
enableMsgTrace = true
)
@Slf4j
public class RealDataConsumer extends EnhanceConsumerMessageHandler<AppAutoDataMessage> implements RocketMQListener<AppAutoDataMessage> {
@Resource
private RedisUtil redisUtil;
@Resource
private RocketMqLogFeignClient rocketMqLogFeignClient;
@Resource
private RtFeignClient rtFeignClient;
@Override
protected void handleMessage(AppAutoDataMessage appAutoDataMessage) {
log.info("分发至实时数据");
rtFeignClient.analysis(appAutoDataMessage);
}
/***
* 通过redis分布式锁判断当前消息所处状态
* 1、null 查不到该key的数据，属于第一次消费，放行
* 2、fail 上次消息消费时发生异常，放行
* 3、being processed 正在处理，打回去
* 4、success 最近72小时消费成功，避免重复消费，打回去
*/
@Override
public boolean filter(AppAutoDataMessage message) {
String keyStatus = redisUtil.getStringByKey(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()));
if (Objects.isNull(keyStatus) || keyStatus.equalsIgnoreCase(MessageStatus.FAIL)) {
redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.BEING_PROCESSED, 60L);
return false;
}
return true;
}
/**
* 消费成功，缓存到redis72小时，避免重复消费
*/
@Override
protected void consumeSuccess(AppAutoDataMessage message) {
redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
}
/**
* 发生异常时，进行错误信息入库保存
* 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存
*/
@Override
protected void saveExceptionMsgLog(AppAutoDataMessage message, String identity, Exception exception) {
redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_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(AppAutoDataMessage appAutoDataMessage) {
super.dispatchMessage(appAutoDataMessage);
}
}