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

refactor(access): 优化设备接入流程和日志记录

Browse Source 
- 移除不必要的日志模板依赖和相关变量
- 简化设备接入方法中的成功失败ID收集逻辑
- 优化模板信息获取的等待机制，替换固定延时为条件循环
- 更新设备状态查询条件，排除未注册状态的设备
- 重构消息处理器中的日志记录方式，统一操作描述格式
- 修复事务注解在特定方法上的配置问题
- 调整设备接入流程步骤顺序，优化异常处理机制
- 更新MQTT消息处理中的操作日志描述，增加系统端标识
- 优化设备数据询问功能的日志记录完整性
This commit is contained in:
xy
2026-06-12 14:43:39 +08:00
parent cfd22f3ea8
commit 9e9fbb360b
6 changed files with 130 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -57,7 +57,7 @@ public enum AccessResponseEnum {
WAVE_INFO_MISSING("A0307","波形参数缺失!"),
MODEL_MISS("A0308","询问模板信息超时,设备未响应!"),
MODEL_VERSION_ERROR("A0308","询问装置模板信息错误"),
MODEL_VERSION_ERROR("A0308","装置模板DevModInfo数据为空"),
UPLOAD_ERROR("A0308","平台上送文件异常"),
RELOAD_UPLOAD_ERROR("A0308","平台重新上送文件异常"),

62
iot-access/access-boot/src/main/java/com/njcn/access/handler/MqttMessageHandler.java
View File

@@ -105,15 +105,16 @@ public class MqttMessageHandler {
@MqttSubscribe(value = "/Dev/DevTopic/{edgeId}",qos = 1)
@Transactional(rollbackFor = Exception.class)
public void devTopic(String topic, MqttMessage message, @NamedValue("edgeId") String nDid, @Payload String payload){
//业务流程开始
Gson gson = new Gson();
ReqAndResParam.Res res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResParam.Res.class);
//日志记录
LogMessage logDto = new LogMessage();
logDto.setUserIndex("系统");
logDto.setLoginName("系统");
logDto.setOperate(nDid + "设备主题录入");
logDto.setOperate("系统端收到装置端"+nDid+"发送的主题信息，code = " + res.getCode());
logDto.setResult(1);
//业务流程开始
Gson gson = new Gson();
ReqAndResParam.Res res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResParam.Res.class);
logMessageTemplate.sendMember(logDto);
//检验传递的参数是否准确
Set<ConstraintViolation<ReqAndResParam.Res>> validate = validator.validate(res);
// validate.forEach(constraintViolation -> {
@@ -162,15 +163,16 @@ public class MqttMessageHandler {
@Transactional(rollbackFor = Exception.class)
public void devOperation(String topic, MqttMessage message, @NamedValue("edgeId") String nDid, @Payload String payload){
log.info("收到注册应答响应--->{}", nDid);
//业务处理
Gson gson = new Gson();
ReqAndResDto.Res res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResDto.Res.class);
//日志记录
LogMessage logDto = new LogMessage();
logDto.setUserIndex("系统");
logDto.setLoginName("系统");
logDto.setOperate("收到设备"+nDid+"注册应答响应");
logDto.setOperate("系统端收到装置端"+nDid+"注册应答响应，code = " + res.getCode());
logDto.setResult(1);
//业务处理
Gson gson = new Gson();
ReqAndResDto.Res res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResDto.Res.class);
logMessageTemplate.sendMember(logDto);
if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){
if (Objects.equals(res.getType(),Integer.parseInt(TypeEnum.TYPE_17.getCode()))){
//询问模板数据
@@ -182,6 +184,11 @@ public class MqttMessageHandler {
reqAndResParam.setExpire(-1);
String version = csTopicService.getVersion(nDid);
publisher.send("/Pfm/DevCmd/"+version+"/"+nDid,new Gson().toJson(reqAndResParam),1,false);
//记录日志
logDto.setUserIndex("系统");
logDto.setLoginName("系统");
logDto.setOperate("注册阶段：系统端向装置端"+nDid+"发送询问模板请求");
logDto.setResult(1);
logMessageTemplate.sendMember(logDto);
} else {
logDto.setResult(0);
@@ -207,7 +214,7 @@ public class MqttMessageHandler {
*/
@MqttSubscribe(value = "/Pfm/DevRsp/{version}/{edgeId}",qos = 1)
@Transactional(rollbackFor = Exception.class)
public void devAccessOperation(String topic, MqttMessage message, @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload){
public void devAccessOperation(String topic, MqttMessage message, @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload) throws InterruptedException {
//日志实体
LogMessage logDto = new LogMessage();
logDto.setUserIndex("系统");
@@ -226,7 +233,8 @@ public class MqttMessageHandler {
* 3.平台端需读取装置的DevMod来判断网关支持的设备模板（包含设备型号和模板版本），根据app提交的接入子设备DID匹配数据模板（型号及版本），生成DevCfg下发给网关，网关根据下发信息生成就地设备点表。
*/
case 4611:
logDto.setOperate("装置模板应答");
logDto.setOperate("系统端收到装置端"+nDid+"模板应答，code = " + res.getCode());
logMessageTemplate.sendMember(logDto);
//log.info("{},装置模板应答,应答code {}",nDid,res.getCode());
ModelDto modelDto = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ModelDto.class);
List<DevModInfoDto> list = modelDto.getMsg().getDevMod();
@@ -292,7 +300,9 @@ public class MqttMessageHandler {
logMessageTemplate.sendMember(logDto);
break;
case 4613:
logDto.setOperate(nDid + "设备接入");
logDto.setOperate("系统端收到装置端"+nDid+"接入应答，code = " + res.getCode());
logDto.setResult(1);
logMessageTemplate.sendMember(logDto);
//log.info("{},收到接入应答响应,应答code {}",nDid,res.getCode());
if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){
int mid = 1;
@@ -334,7 +344,8 @@ public class MqttMessageHandler {
if (!Objects.isNull(rspDataDto.getDataType())) {
switch (rspDataDto.getDataType()){
case 1:
logDto.setOperate(nDid + "更新设备软件信息");
logDto.setOperate("系统端收到装置端"+nDid+"更新设备软件信息报文，code = " + res.getCode());
logDto.setResult(1);
RspDataDto.SoftInfo softInfo = JSON.parseObject(JSON.toJSONString(rspDataDto.getDataArray()), RspDataDto.SoftInfo.class);
//记录设备软件信息
CsSoftInfoPO csSoftInfoPo = new CsSoftInfoPO();
@@ -365,7 +376,8 @@ public class MqttMessageHandler {
List<RspDataDto.LdevInfo> devInfo = JSON.parseArray(JSON.toJSONString(rspDataDto.getDataArray()), RspDataDto.LdevInfo.class);
if (CollectionUtil.isNotEmpty(devInfo)){
if (Objects.equals(res.getDid(),1)){
logDto.setOperate(nDid + "更新APF容量");
logDto.setOperate("系统端收到装置端"+nDid+"更新APF容量报文，code = " + res.getCode());
logDto.setResult(1);
List<CsDevCapacityPO> list3 = new ArrayList<>();
boolean hasZeroClDid = devInfo.stream().anyMatch(item -> item.getClDid() == 0);
//治理设备
@@ -394,7 +406,8 @@ public class MqttMessageHandler {
equipmentFeignClient.updateModuleNumber(nDid,(devInfo.size()-1));
}
} else if (Objects.equals(res.getDid(),2)) {
logDto.setOperate(nDid + "更新电网侧、负载侧监测点信息");
logDto.setOperate("系统端收到装置端"+nDid+"更新电网侧、负载侧监测点信息报文，code = " + res.getCode());
logDto.setResult(1);
//1.更新电网侧、负载侧监测点相关信息
devInfo.forEach(item->{
updateLineInfo(nDid,item);
@@ -404,7 +417,8 @@ public class MqttMessageHandler {
logMessageTemplate.sendMember(logDto);
break;
case 15:
logDto.setOperate(nDid + "更新设备软件信息");
logDto.setOperate("系统端收到装置端"+nDid+"更新设备软件信息报文，code = " + res.getCode());
logDto.setResult(1);
JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(res));
AppAutoDataMessage appAutoDataMessage = JSONObject.toJavaObject(jsonObject, AppAutoDataMessage.class);
appAutoDataMessage.setId(nDid);
@@ -412,8 +426,8 @@ public class MqttMessageHandler {
logMessageTemplate.sendMember(logDto);
break;
case 48:
logDto.setUserIndex("系统");
logDto.setOperate("监测点:" + (nDid + rspDataDto.getClDid()) + "询问项目列表");
logDto.setOperate("系统端收到装置端"+nDid+"询问项目列表报文，code = " + res.getCode());
logDto.setResult(1);
List<RspDataDto.ProjectInfo> projectInfoList = JSON.parseArray(JSON.toJSONString(rspDataDto.getDataArray()), RspDataDto.ProjectInfo.class);
String key3 = AppRedisKey.PROJECT_INFO + nDid + rspDataDto.getClDid();
redisUtil.saveByKeyWithExpire(key3,projectInfoList,60L);
@@ -433,6 +447,11 @@ public class MqttMessageHandler {
default:
break;
}
} else if (Objects.equals(res.getCode(),AccessEnum.START_CHANNEL.getCode())) {
logDto.setOperate(AccessEnum.START_CHANNEL.getMessage() + "，系统等待5s");
logDto.setResult(1);
logMessageTemplate.sendMember(logDto);
Thread.sleep(5000);
} else {
String result = getEnum(res.getCode());
//log.info(result);
@@ -690,6 +709,10 @@ public class MqttMessageHandler {
* 3：监测点pt/ct信息
*/
public void askDevData(String nDid,String version,Integer type,Integer mid){
LogMessage logDto = new LogMessage();
logDto.setUserIndex("系统");
logDto.setLoginName("系统");
logDto.setResult(1);
ReqAndResDto.Req reqAndResParam = new ReqAndResDto.Req();
reqAndResParam.setMid(mid);
reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
@@ -705,28 +728,33 @@ public class MqttMessageHandler {
reqAndResParam.setDid(0);
askDataDto.setCldid(0);
askDataDto.setDataType(1);
logDto.setOperate("系统端向装置端"+nDid+"询问软件信息");
break;
case 2:
reqAndResParam.setDid(1);
askDataDto.setCldid(-1);
askDataDto.setDataType(2);
logDto.setOperate("系统端向装置端"+nDid+"询问逻辑设备1信息");
break;
case 3:
reqAndResParam.setDid(2);
askDataDto.setCldid(-1);
askDataDto.setDataType(2);
logDto.setOperate("系统端向装置端"+nDid+"询问逻辑设备2信息");
break;
//询问工程信息
case 48:
reqAndResParam.setDid(1);
askDataDto.setCldid(1);
askDataDto.setDataType(48);
logDto.setOperate("系统端向装置端"+nDid+"询问工程信息");
break;
default:
break;
}
reqAndResParam.setMsg(askDataDto);
publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(reqAndResParam),1,false);
logMessageTemplate.sendMember(logDto);
}
public String getEnum(Integer code) {

23
iot-access/access-boot/src/main/java/com/njcn/access/runner/AccessApplicationRunner.java
View File

@@ -41,7 +41,6 @@ public class AccessApplicationRunner implements ApplicationRunner {
private final ICsEquipmentDeliveryService csEquipmentDeliveryService;
private final ICsTopicService csTopicService;
private final CsDeviceServiceImpl csDeviceService;
private final LogMessageTemplate logMessageTemplate;
private final ChannelObjectUtil channelObjectUtil;
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
@@ -101,8 +100,6 @@ public class AccessApplicationRunner implements ApplicationRunner {
public void accessDev(List<CsEquipmentDeliveryPO> list, Map<String, SysDicTreePO> dictTreeMap, Map<String,String> topicVersions) {
if (CollUtil.isNotEmpty(list)) {
List<String> successIds = new ArrayList<>();
List<String> failIds = new ArrayList<>();
try {
list.forEach(item->{
//System.out.println(Thread.currentThread().getName() + ": reboot : nDid : " + item.getNdid());
@@ -118,29 +115,9 @@ public class AccessApplicationRunner implements ApplicationRunner {
boolean result = csDeviceService.autoAccess(item.getNdid(),version,1);
if (result) {
redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1);
successIds.add(item.getNdid());
} else {
failIds.add(item.getNdid());
}
}
});
if (CollUtil.isNotEmpty(successIds)) {
LogMessage logDto = new LogMessage();
logDto.setUserIndex("系统首次启动");
logDto.setLoginName("系统首次启动");
logDto.setResult(1);
logDto.setOperate(String.join(",", successIds) + "装置接入");
logMessageTemplate.sendMember(logDto);
}
if (CollUtil.isNotEmpty(failIds)) {
LogMessage logDto = new LogMessage();
logDto.setUserIndex("系统首次启动");
logDto.setLoginName("系统首次启动");
logDto.setResult(0);
logDto.setOperate(String.join(",", failIds) + "装置接入");
logDto.setFailReason("装置不在线或者系统询问装置模板信息，装置未响应");
logMessageTemplate.sendMember(logDto);
}
} catch (Exception e) {
log.error(e.getMessage());
}

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

@@ -6,7 +6,6 @@ import com.njcn.access.service.ICsTopicService;
import com.njcn.access.service.impl.CsDeviceServiceImpl;
import com.njcn.access.utils.ChannelObjectUtil;
import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
import com.njcn.mq.message.LogMessage;
import com.njcn.mq.template.LogMessageTemplate;
import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.utils.RedisUtil;
@@ -40,7 +39,6 @@ public class AutoAccessTimer implements ApplicationRunner {
private final ICsTopicService csTopicService;
private final CsDeviceServiceImpl csDeviceService;
private final ChannelObjectUtil channelObjectUtil;
private final LogMessageTemplate logMessageTemplate;
@Override
public void run(ApplicationArguments args) {
@@ -140,8 +138,6 @@ public class AutoAccessTimer implements ApplicationRunner {
//安全的accessDev版本
private void accessDevSafely(List<CsEquipmentDeliveryPO> list, Map<String, SysDicTreePO> dictTreeMap, Map<String,String> topicVersions) {
if (CollUtil.isNotEmpty(list)) {
List<String> successIds = new ArrayList<>();
List<String> failIds = new ArrayList<>();
for (CsEquipmentDeliveryPO item : list) {
try {
if (Objects.equals(dictTreeMap.get(item.getDevType()).getCode(), DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(), 1)) {
@@ -156,10 +152,6 @@ public class AutoAccessTimer implements ApplicationRunner {
boolean success = csDeviceService.autoAccess2(item.getNdid(), version, 1);
if (success) {
redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1);
successIds.add(item.getNdid());
} else {
log.warn("设备 {} 接入失败", item.getNdid());
failIds.add(item.getNdid());
}
} catch (Exception e) {
log.error("设备 {} 接入异常: {}", item.getNdid(), e.getMessage());
@@ -169,23 +161,6 @@ public class AutoAccessTimer implements ApplicationRunner {
log.error("处理设备 {} 失败: {}", item.getNdid(), e.getMessage());
}
}
if (CollUtil.isNotEmpty(successIds)) {
LogMessage logDto = new LogMessage();
logDto.setUserIndex("系统首次启动");
logDto.setLoginName("系统首次启动");
logDto.setResult(1);
logDto.setOperate(String.join(",", successIds) + "装置接入");
logMessageTemplate.sendMember(logDto);
}
if (CollUtil.isNotEmpty(failIds)) {
LogMessage logDto = new LogMessage();
logDto.setUserIndex("系统首次启动");
logDto.setLoginName("系统首次启动");
logDto.setResult(0);
logDto.setOperate(String.join(",", failIds) + "装置接入");
logDto.setFailReason("装置不在线或者系统询问装置模板信息，装置未响应");
logMessageTemplate.sendMember(logDto);
}
}
}

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

@@ -17,7 +17,6 @@ import com.njcn.access.pojo.param.DeviceStatusParam;
import com.njcn.access.service.*;
import com.njcn.access.utils.ChannelObjectUtil;
import com.njcn.access.utils.MqttUtil;
import com.njcn.common.pojo.dto.DeviceLogDTO;
import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.csdevice.api.*;
@@ -39,7 +38,7 @@ import com.njcn.user.api.UserFeignClient;
import com.njcn.user.enums.AppRoleEnum;
import com.njcn.user.pojo.vo.UserVO;
import com.njcn.web.utils.RequestUtil;
import lombok.AllArgsConstructor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,12 +57,11 @@ import java.util.stream.Collectors;
* @createTime 2023/7/4 13:43
*/
@Service
@AllArgsConstructor
@RequiredArgsConstructor
@Slf4j
public class CsDeviceServiceImpl implements ICsDeviceService {
private static final Logger logger = LoggerFactory.getLogger(CsDeviceServiceImpl.class);
private final EquipmentFeignClient equipmentFeignClient;
private final ICsEquipmentDeliveryService csEquipmentDeliveryService;
private final DictTreeFeignClient dictTreeFeignClient;
@@ -92,6 +90,7 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
private final CsHarmonicPlanLineFeignClient csHarmonicPlanLineFeignClient;
private final LogMessageTemplate logMessageTemplate;
@Override
@Transactional(rollbackFor = {Exception.class})
public void devRegister(String nDid,Integer type) {
@@ -99,7 +98,6 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
LogMessage message = new LogMessage();
message.setUserIndex(RequestUtil.getUserNickname());
message.setLoginName(RequestUtil.getUsername());
message.setResult(1);
//1.判断客户端是否在线
String clientName = "NJCN-" + nDid.substring(nDid.length() - 6);
boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
@@ -170,8 +168,6 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
}
String devModel = dictData.getCode();
zhiLianRegister(nDid,devModel);
message.setOperate("直连装置"+nDid+"下发注册报文");
logMessageTemplate.sendMember(message);
}
@Override
@@ -180,13 +176,18 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
LogMessage logDto = new LogMessage();
logDto.setUserIndex(RequestUtil.getUserNickname());
logDto.setLoginName(RequestUtil.getUsername());
logDto.setOperate("获取"+nDid+"装置模板信息");
logDto.setOperate("注册阶段：获取"+nDid+"装置模板信息");
logDto.setResult(1);
Object model = null;
try {
Thread.sleep(3000);
String key = AppRedisKey.LINE + nDid;
model = redisUtil.getObjectByKey(key);
for (int i = 0; i < 3 ; i++) {
Thread.sleep(1000);
if (!Objects.isNull(model)){
break;
}
}
if (Objects.isNull(model)){
logDto.setResult(0);
logDto.setFailReason(AccessResponseEnum.MODEL_MISS.getMessage());
@@ -201,14 +202,14 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
}
@Override
@Transactional(rollbackFor = {Exception.class})
// @Transactional(rollbackFor = {Exception.class})
//fixme 这边事务不起作用，中途出错会导致数据部分录入，再次接入会报主键冲突，所以暂时加了个重置按钮，清空台账数据的
public void devAccess(DevAccessParam devAccessParam) {
//日志实体
LogMessage logDto = new LogMessage();
logDto.setUserIndex(RequestUtil.getUserNickname());
logDto.setLoginName(RequestUtil.getUsername());
logDto.setOperate("装置"+devAccessParam.getNDid()+"接入");
logDto.setOperate("系统端向装置端"+devAccessParam.getNDid()+"发送接入请求");
logDto.setResult(1);
try {
//获取版本
@@ -318,14 +319,10 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
po.setSubUserId(RequestUtil.getUserIndex());
po.setDeviceId(vo.getId());
csDeviceUserService.saveBatch(Collections.singletonList(po));
//6.修改装置状态;修改装置接入的工程、项目
csEquipmentDeliveryService.updateStatusBynDid(devAccessParam.getNDid(), AccessEnum.REGISTERED.getCode(),devAccessParam.getEngineeringId(), devAccessParam.getProjectId());
//7.发起自动接入请求
devAccessAskTemplate(devAccessParam.getNDid(),version,1);
//8.删除redis监测点模板信息
//6.删除redis监测点模板信息
redisUtil.delete(AppRedisKey.MODEL + devAccessParam.getNDid());
redisUtil.delete(AppRedisKey.LINE + devAccessParam.getNDid());
//9.存储装置调试日志表
//7.存储装置调试日志表
CsEquipmentProcessPO csEquipmentProcess = new CsEquipmentProcessPO();
csEquipmentProcess.setDevId(devAccessParam.getNDid());
csEquipmentProcess.setOperator(RequestUtil.getUserIndex());
@@ -336,7 +333,7 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
csEquipmentProcess.setStatus(1);
}
processFeignClient.add(csEquipmentProcess);
//11.这里会出现工程用户接入装置时，如果当前工程用户并没有关注，接入之后应该将用户和工程关联起来
//8.这里会出现工程用户接入装置时，如果当前工程用户并没有关注，接入之后应该将用户和工程关联起来
List<UserVO> users = userFeignClient.getUserVOByIdList(Collections.singletonList(RequestUtil.getUserIndex())).getData();
if (CollectionUtil.isNotEmpty(users)) {
UserVO userVO = users.get(0);
@@ -344,12 +341,23 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
csMarketDataFeignClient.insertData(userVO.getId(), devAccessParam.getEngineeringId());
}
}
//9.修改装置状态;修改装置接入的工程、项目
csEquipmentDeliveryService.updateStatusBynDid(devAccessParam.getNDid(), AccessEnum.REGISTERED.getCode(),devAccessParam.getEngineeringId(), devAccessParam.getProjectId());
//10.存储日志
logMessageTemplate.sendMember(logDto);
//11.发起自动接入请求
devAccessAskTemplate(devAccessParam.getNDid(),version,1);
} catch (Exception e) {
logDto.setResult(0);
logDto.setFailReason(e.getMessage());
logMessageTemplate.sendMember(logDto);
//这边如果发生异常，已经发送接入指令，设备端会响应，这里需要重置
try {
Thread.sleep(1000);
resetFactory(devAccessParam.getNDid());
} catch (InterruptedException ex) {
throw new RuntimeException(ex);
}
throw new BusinessException(CommonResponseEnum.FAIL);
}
}
@@ -359,24 +367,13 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
public void resetFactory(String nDid) {
CsEquipmentDeliveryVO csEquipmentDeliveryVO = csEquipmentDeliveryService.queryEquipmentBynDid(nDid);
String devId = csEquipmentDeliveryVO.getId();
//修改装置状态
DeviceStatusParam param = new DeviceStatusParam();
param.setNDid(nDid);
param.setStatus(1);
param.setRunStatus(1);
// boolean isConnectDev = DicDataEnum.CONNECT_DEV.getCode().equals(dictTreeFeignClient.queryById(csEquipmentDeliveryVO.getDevType()).getData().getCode());
// if (isConnectDev) {
// param.setProcess(2);
// } else {
// param.setProcess(4);
// }
param.setProcess(4);
csEquipmentDeliveryService.devResetFactory(param);
//清除关系表
QueryWrapper<CsLedger> csLedgerQueryWrapper = new QueryWrapper<>();
/**
* 删除装置
*/
csLedgerQueryWrapper.clear();
csLedgerQueryWrapper.eq("id",devId);
csLedgerService.remove(csLedgerQueryWrapper);
@@ -403,12 +400,10 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
appLineTopologyDiagramPOQueryWrapper.in("line_id",collect);
appLineTopologyDiagramService.remove(appLineTopologyDiagramPOQueryWrapper);
}
//删除topic表
csTopicService.deleteByNDid(nDid);
//清空缓存
redisUtil.deleteKeysByString(AppRedisKey.LINE_POSITION+nDid);
}
@Override
@Transactional(rollbackFor = Exception.class)
public boolean manualAccess(String nDid) {
@@ -424,7 +419,7 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
LogMessage logDto = new LogMessage();
logDto.setUserIndex(RequestUtil.getUserNickname());
logDto.setLoginName(RequestUtil.getUsername());
logDto.setOperate("便携式装置"+nDid+"注册、接入");
logDto.setOperate("系统端向便携式装置端"+nDid+"发送注册、接入请求");
logDto.setResult(1);
try {
// 装置状态判断
@@ -493,22 +488,23 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
csDevModelRelationAddParm.setModelId(modelList.get(0).getModelId());
csDevModelRelationAddParm.setDid(modelList.get(0).getDid());
csDevModelRelationService.addDevModelRelation(csDevModelRelationAddParm);
//5.发起自动接入请求
Thread.sleep(2000);
//先获取版本
//String version = csTopicService.getVersion(nDid);
String version = "V1";
devAccessAskTemplate(nDid,version,1);
//6.修改流程,便携式装置接入成功即为实际环境
csEquipmentDeliveryService.updateProcessBynDid(nDid,4);
//7.删除redis监测点模板信息
redisUtil.delete(AppRedisKey.MODEL + nDid);
redisUtil.delete(AppRedisKey.LINE + nDid);
//5.发起自动接入请求
Thread.sleep(2000);
String version = "V1";
devAccessAskTemplate(nDid,version,1);
//判断接入状态
Thread.sleep(5000);
Object object = redisUtil.getObjectByKey("online" + nDid);
if (Objects.nonNull(object)) {
result = "success";
for (int i = 0; i < 5; i++) {
Thread.sleep(1000);
if (Objects.nonNull(object)) {
result = "success";
break;
}
}
//8.存储日志
logMessageTemplate.sendMember(logDto);
@@ -516,7 +512,13 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
logDto.setResult(0);
logDto.setFailReason(e.getMessage());
logMessageTemplate.sendMember(logDto);
resetFactory(nDid);
//这边如果发生异常，已经发送接入指令，设备端会响应，这里需要重置
try {
Thread.sleep(1000);
resetFactory(nDid);
} catch (InterruptedException ex) {
throw new RuntimeException(ex);
}
throw new BusinessException(AccessResponseEnum.ACCESS_ERROR);
}
return result;
@@ -572,6 +574,7 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
logDto.setUserIndex(RequestUtil.getUserNickname());
logDto.setLoginName(RequestUtil.getUsername());
logDto.setOperate("监测装置"+nDid+"注册、接入");
logDto.setOperate("系统端向监测装置端"+nDid+"发送注册、接入请求");
logDto.setResult(1);
try {
// 装置状态判断
@@ -646,22 +649,24 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
po.setSubUserId(RequestUtil.getUserIndex());
po.setDeviceId(vo.getId());
csDeviceUserService.saveBatch(Collections.singletonList(po));
//发起自动接入请求
Thread.sleep(2000);
//先获取版本
String version = "V1";
devAccessAskTemplate(nDid,version,1);
//6.修改流程,接入成功即为实际环境
csEquipmentDeliveryService.updateProcessBynDid(nDid,4);
//7.删除redis监测点模板信息
redisUtil.delete(AppRedisKey.MODEL + nDid);
redisUtil.delete(AppRedisKey.LINE + nDid);
//发起自动接入请求
Thread.sleep(2000);
//先获取版本
String version = "V1";
devAccessAskTemplate(nDid,version,1);
//判断接入状态
Thread.sleep(5000);
Object object = redisUtil.getObjectByKey("online" + nDid);
if (Objects.nonNull(object)) {
result = "success";
for (int i = 0; i < 5; i++) {
Thread.sleep(1000);
if (Objects.nonNull(object)) {
result = "success";
break;
}
}
//8.存储日志
logMessageTemplate.sendMember(logDto);
@@ -669,7 +674,13 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
logDto.setResult(0);
logDto.setFailReason(e.getMessage());
logMessageTemplate.sendMember(logDto);
resetFactory(nDid);
//这边如果发生异常，已经发送接入指令，设备端会响应，这里需要重置
try {
Thread.sleep(1000);
resetFactory(nDid);
} catch (InterruptedException ex) {
throw new RuntimeException(ex);
}
throw new BusinessException(e.getMessage());
}
return result;
@@ -784,6 +795,9 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
}
//发起接入
publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(getJson(mid,TypeEnum.TYPE_5.getCode())),1,false);
logDto.setOperate("系统端向装置端"+nDid+"发送接入请求");
logDto.setResult(1);
logMessageTemplate.sendMember(logDto);
//录波任务倒计时
redisUtil.saveByKeyWithExpire("startFile:" + nDid,null,60L);
result = true;
@@ -963,6 +977,13 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
reqAndResParam.setExpire(-1);
logger.info("询问主题报文为:{}", new Gson().toJson(reqAndResParam));
publisher.send("/Pfm/DevTopic/"+nDid, new Gson().toJson(reqAndResParam),1,false);
//记录日志
LogMessage logDto = new LogMessage();
logDto.setUserIndex("系统");
logDto.setLoginName("系统");
logDto.setOperate("系统端向装置端"+nDid+"发送主题询问报文");
logDto.setResult(1);
logMessageTemplate.sendMember(logDto);
}
/**
@@ -983,5 +1004,12 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
reqAndResParam.setMsg(accessDto);
logger.info("注册报文为:{}", new Gson().toJson(reqAndResParam));
publisher.send("/Pfm/DevReg/"+nDid, new Gson().toJson(reqAndResParam),1,false);
//记录日志
LogMessage logDto = new LogMessage();
logDto.setUserIndex("系统");
logDto.setLoginName("系统");
logDto.setOperate("系统端向装置端"+nDid+"发送注册请求");
logDto.setResult(1);
logMessageTemplate.sendMember(logDto);
}
}

6
iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsEquipmentDeliveryServiceImpl.java
View File

@@ -16,7 +16,6 @@ import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
@@ -101,6 +100,7 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
List<CsEquipmentDeliveryPO> result = new ArrayList<>();
List<CsEquipmentDeliveryPO> list = this.lambdaQuery()
.ne(CsEquipmentDeliveryPO::getRunStatus,AccessEnum.DEL.getCode())
.ne(CsEquipmentDeliveryPO::getStatus,AccessEnum.UNREGISTERED.getCode())
.eq(CsEquipmentDeliveryPO::getUsageStatus,1)
.list();
if (CollUtil.isNotEmpty(list)) {
@@ -125,12 +125,12 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
}
@Override
public List<CsEquipmentDeliveryPO> getOfflineDev() {
public List<CsEquipmentDeliveryPO> getOfflineDev() {
List<CsEquipmentDeliveryPO> result = new ArrayList<>();
List<CsEquipmentDeliveryPO> list = this.lambdaQuery()
.eq(CsEquipmentDeliveryPO::getRunStatus,AccessEnum.OFFLINE.getCode())
.eq(CsEquipmentDeliveryPO::getUsageStatus,1)
.in(CsEquipmentDeliveryPO::getStatus, Arrays.asList(2,3))
.ne(CsEquipmentDeliveryPO::getStatus,AccessEnum.UNREGISTERED.getCode())
.isNull(CsEquipmentDeliveryPO::getNodeId)
.list();
if (CollUtil.isNotEmpty(list)) {