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

装置接入流程调整

Browse Source
This commit is contained in:
xy
2024-08-16 13:00:22 +08:00
parent ab9215f13e
commit 90dc22021f
8 changed files with 195 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
iot-access/access-boot/src/main/java/com/njcn/access/controller/CsDeviceController.java
View File

@@ -86,8 +86,12 @@ public class CsDeviceController extends BaseController {
@ApiImplicitParam(name = "nDid", value = "设备id", required = true) @ApiImplicitParam(name = "nDid", value = "设备id", required = true)
public HttpResult<String> manualAccess(@RequestParam String nDid){ public HttpResult<String> manualAccess(@RequestParam String nDid){
String methodDescribe = getMethodDescribe("manualAccess"); String methodDescribe = getMethodDescribe("manualAccess");
csDeviceService.manualAccess(nDid); boolean result = csDeviceService.manualAccess(nDid);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe); if (result) {
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, "成功", methodDescribe);
} else {
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.FAIL, "失败", methodDescribe);
}
} }
@OperateInfo(info = LogEnum.BUSINESS_COMMON) @OperateInfo(info = LogEnum.BUSINESS_COMMON)

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

@@ -88,7 +88,7 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
try { try {
String version = csTopicService.getVersion(nDid); String version = csTopicService.getVersion(nDid);
//装置没有心跳，则立马发起接入请求 //装置没有心跳，则立马发起接入请求
csDeviceService.devAccess(nDid,version); csDeviceService.devAccessAskTemplate(nDid,version,1);
logDto.setOperate("装置掉线3分钟发送接入请求"); logDto.setOperate("装置掉线3分钟发送接入请求");
csLogsFeignClient.addUserLog(logDto); csLogsFeignClient.addUserLog(logDto);
Integer status = csEquipmentDeliveryService.queryEquipmentBynDid(nDid).getRunStatus(); Integer status = csEquipmentDeliveryService.queryEquipmentBynDid(nDid).getRunStatus();
@@ -96,10 +96,10 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
throw new BusinessException(CommonResponseEnum.SUCCESS); throw new BusinessException(CommonResponseEnum.SUCCESS);
} }
//心跳断连立马发起接入失败后，1分钟再次发起请求，请求3次 //心跳断连立马发起接入失败后，1分钟再次发起请求，请求3次
for (int i = 1; i < 4; i++) { for (int i = 2; i < 5; i++) {
//接入再次失败，则定时发起接入请求 //接入再次失败，则定时发起接入请求
Thread.sleep(1000 * 60); Thread.sleep(1000 * 60);
csDeviceService.devAccess(nDid,version); csDeviceService.devAccessAskTemplate(nDid,version,i);
status = csEquipmentDeliveryService.queryEquipmentBynDid(nDid).getRunStatus(); status = csEquipmentDeliveryService.queryEquipmentBynDid(nDid).getRunStatus();
if (Objects.equals(status,AccessEnum.ONLINE.getCode())){ if (Objects.equals(status,AccessEnum.ONLINE.getCode())){
break; break;
@@ -108,10 +108,10 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
csLogsFeignClient.addUserLog(logDto); csLogsFeignClient.addUserLog(logDto);
} }
if (!Objects.isNull(status) && Objects.equals(status,AccessEnum.OFFLINE.getCode())){ if (!Objects.isNull(status) && Objects.equals(status,AccessEnum.OFFLINE.getCode())){
final int[] mid = {2}; final int[] mid = {5};
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(2); ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(2);
ScheduledFuture<?> runnableFuture = executor.scheduleAtFixedRate(() -> { ScheduledFuture<?> runnableFuture = executor.scheduleAtFixedRate(() -> {
csDeviceService.devAccessMid(nDid,version, mid[0]); csDeviceService.devAccessAskTemplate(nDid,version, mid[0]);
Integer status2 = csEquipmentDeliveryService.queryEquipmentBynDid(nDid).getRunStatus(); Integer status2 = csEquipmentDeliveryService.queryEquipmentBynDid(nDid).getRunStatus();
if (Objects.equals(status2,AccessEnum.ONLINE.getCode())){ if (Objects.equals(status2,AccessEnum.ONLINE.getCode())){
throw new BusinessException(CommonResponseEnum.SUCCESS); throw new BusinessException(CommonResponseEnum.SUCCESS);

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

@@ -1,47 +1,47 @@
package com.njcn.access.runner; //package com.njcn.access.runner;
//
import com.njcn.access.service.ICsEquipmentDeliveryService; //import com.njcn.access.service.ICsEquipmentDeliveryService;
import com.njcn.access.service.ICsTopicService; //import com.njcn.access.service.ICsTopicService;
import com.njcn.access.service.impl.CsDeviceServiceImpl; //import com.njcn.access.service.impl.CsDeviceServiceImpl;
import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO; //import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments; //import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner; //import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component; //import org.springframework.stereotype.Component;
//
import javax.annotation.Resource; //import javax.annotation.Resource;
import java.util.List; //import java.util.List;
import java.util.Objects; //import java.util.Objects;
//
/** ///**
* 类的介绍：用来重新发起设备的接入，存在程序意外停止了，缓存失效导致无法更新装置的状态，所以需要在程序启动时发起设备的接入 // * 类的介绍：用来重新发起设备的接入，存在程序意外停止了，缓存失效导致无法更新装置的状态，所以需要在程序启动时发起设备的接入
* // *
* @author xuyang // * @author xuyang
* @version 1.0.0 // * @version 1.0.0
* @createTime 2023/8/28 13:57 // * @createTime 2023/8/28 13:57
*/ // */
@Component //@Component
@Slf4j //@Slf4j
public class AccessApplicationRunner implements ApplicationRunner { //public class AccessApplicationRunner implements ApplicationRunner {
//
@Resource // @Resource
private CsDeviceServiceImpl csDeviceService; // private CsDeviceServiceImpl csDeviceService;
//
@Resource // @Resource
private ICsTopicService csTopicService; // private ICsTopicService csTopicService;
//
@Resource // @Resource
private ICsEquipmentDeliveryService csEquipmentDeliveryService; // private ICsEquipmentDeliveryService csEquipmentDeliveryService;
//
@Override // @Override
public void run(ApplicationArguments args){ // public void run(ApplicationArguments args){
List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getAll(); // List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getAll();
list.forEach(item->{ // list.forEach(item->{
String version = csTopicService.getVersion(item.getNdid()); // String version = csTopicService.getVersion(item.getNdid());
if (!Objects.isNull(version)){ // if (!Objects.isNull(version)){
csDeviceService.devAccess(item.getNdid(),version); // csDeviceService.devAccess(item.getNdid(),version,1);
} // }
}); // });
} // }
//
} //}

103
iot-access/access-boot/src/main/java/com/njcn/access/runner/AccessScheduledTask.java
View File

@@ -1,50 +1,53 @@
package com.njcn.access.runner; //package com.njcn.access.runner;
//
import com.njcn.access.service.ICsEquipmentDeliveryService; //import cn.hutool.core.collection.CollUtil;
import com.njcn.access.service.ICsTopicService; //import cn.hutool.core.collection.CollectionUtil;
import com.njcn.access.service.impl.CsDeviceServiceImpl; //import com.njcn.access.service.ICsEquipmentDeliveryService;
import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO; //import com.njcn.access.service.ICsTopicService;
import lombok.extern.slf4j.Slf4j; //import com.njcn.access.service.impl.CsDeviceServiceImpl;
import org.springframework.scheduling.annotation.Scheduled; //import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
import org.springframework.stereotype.Component; //import lombok.extern.slf4j.Slf4j;
//import org.springframework.scheduling.annotation.Scheduled;
import javax.annotation.Resource; //import org.springframework.stereotype.Component;
import java.util.List; //
import java.util.Objects; //import javax.annotation.Resource;
//import java.util.List;
/** //import java.util.Objects;
* 类的介绍：防止设备掉线 系统未能调整，做一个定时任务，每天凌晨将所有设备重新接入 //
* ///**
* @author xuyang // * 类的介绍：防止设备掉线 系统未能调整，做一个定时任务，每天凌晨将所有设备重新接入
* @version 1.0.0 // *
* @createTime 2023/8/28 14:21 // * @author xuyang
*/ // * @version 1.0.0
@Component // * @createTime 2023/8/28 14:21
@Slf4j // */
public class AccessScheduledTask { //@Component
//@Slf4j
@Resource //public class AccessScheduledTask {
private CsDeviceServiceImpl csDeviceService; //
// @Resource
@Resource // private CsDeviceServiceImpl csDeviceService;
private ICsTopicService csTopicService; //
// @Resource
@Resource // private ICsTopicService csTopicService;
private ICsEquipmentDeliveryService csEquipmentDeliveryService; //
// @Resource
/** // private ICsEquipmentDeliveryService csEquipmentDeliveryService;
* {秒数} {分钟} {小时} {日期} {月份} {星期} {年份(可为空)} //
*/ // /**
@Scheduled(cron = "0 0 0 * * ?") // * {秒数} {分钟} {小时} {日期} {月份} {星期} {年份(可为空)}
public void executeTask() { // */
log.info("每日凌晨定时任务执行"); // @Scheduled(cron = "0 0 0 * * ?")
List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getAll(); // public void executeTask() {
list.forEach(item->{ // log.info("每日凌晨定时任务执行");
String version = csTopicService.getVersion(item.getNdid()); // List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getAll();
if (!Objects.isNull(version)){ // if (CollUtil.isNotEmpty(list)) {
csDeviceService.devAccess(item.getNdid(),version); // for (int i = 0; i < list.size(); i++) {
} // String version = csTopicService.getVersion(list.get(i).getNdid());
}); // if (!Objects.isNull(version)){
} // csDeviceService.devAccess(list.get(i).getNdid(),version,i);
// }
} // }
// }
// }
//}

5
iot-access/access-boot/src/main/java/com/njcn/access/service/ICsDevModelRelationService.java
View File

@@ -27,5 +27,10 @@ public interface ICsDevModelRelationService extends IService<CsDevModelRelationP
*/ */
CsDevModelRelationPO addDevModelRelation(CsDevModelRelationAddParm addParm); CsDevModelRelationPO addDevModelRelation(CsDevModelRelationAddParm addParm);
/**
* 根据装置id删除模板信息
*/
void deleteByDeviceId(String deviceId);
} }

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

@@ -41,7 +41,7 @@ public interface ICsDeviceService {
* 手动发起接入 * 手动发起接入
* @param nDid * @param nDid
*/ */
void manualAccess(String nDid); boolean manualAccess(String nDid);
/** /**
* 直连设备注册 * 直连设备注册

9
iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDevModelRelationServiceImpl.java
View File

@@ -1,5 +1,6 @@
package com.njcn.access.service.impl; package com.njcn.access.service.impl;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.njcn.access.mapper.CsDevModelRelationMapper; import com.njcn.access.mapper.CsDevModelRelationMapper;
@@ -48,6 +49,14 @@ public class CsDevModelRelationServiceImpl extends ServiceImpl<CsDevModelRelatio
return csDevModelRelationPO; return csDevModelRelationPO;
} }
@Override
public void deleteByDeviceId(String deviceId) {
LambdaQueryWrapper<CsDevModelRelationPO> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(CsDevModelRelationPO::getDevId, deviceId)
.eq (CsDevModelRelationPO::getStatus, 1);
this.baseMapper.delete(wrapper);
}
public List<CsDevModelRelationVO> queryDevModelRelation(CsDevModelRelationQueryParm queryParm) { public List<CsDevModelRelationVO> queryDevModelRelation(CsDevModelRelationQueryParm queryParm) {
QueryWrapper<CsDevModelRelationPO> queryWrapper = new QueryWrapper<> (); QueryWrapper<CsDevModelRelationPO> queryWrapper = new QueryWrapper<> ();
queryWrapper.eq (StringUtils.isNotBlank (queryParm.getId ()),"id",queryParm.getId ()). queryWrapper.eq (StringUtils.isNotBlank (queryParm.getId ()),"id",queryParm.getId ()).

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

@@ -41,10 +41,7 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.util.ArrayList; import java.util.*;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors; import java.util.stream.Collectors;
/** /**
@@ -288,7 +285,7 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
//6.修改装置状态 //6.修改装置状态
csEquipmentDeliveryService.updateStatusBynDid(devAccessParam.getNDid(), AccessEnum.REGISTERED.getCode()); csEquipmentDeliveryService.updateStatusBynDid(devAccessParam.getNDid(), AccessEnum.REGISTERED.getCode());
//7.发起自动接入请求 //7.发起自动接入请求
devAccess(devAccessParam.getNDid(),version); devAccessAskTemplate(devAccessParam.getNDid(),version,1);
//8.删除redis监测点模板信息 //8.删除redis监测点模板信息
redisUtil.delete(AppRedisKey.MODEL + devAccessParam.getNDid()); redisUtil.delete(AppRedisKey.MODEL + devAccessParam.getNDid());
@@ -360,9 +357,10 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
} }
@Override @Override
public void manualAccess(String nDid) { @Transactional(rollbackFor = Exception.class)
public boolean manualAccess(String nDid) {
String version = csTopicService.getVersion(nDid); String version = csTopicService.getVersion(nDid);
devAccess(nDid,version); return devAccessAskTemplate(nDid,version,new Random().nextInt(10000));
} }
@@ -423,7 +421,7 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
//3.修改装置状态为注册状态 //3.修改装置状态为注册状态
csEquipmentDeliveryService.updateStatusBynDid(nDid, AccessEnum.REGISTERED.getCode()); csEquipmentDeliveryService.updateStatusBynDid(nDid, AccessEnum.REGISTERED.getCode());
//4.发起自动接入请求 //4.发起自动接入请求
devAccess(nDid,version); devAccessAskTemplate(nDid,version,1);
//5.存储日志 //5.存储日志
csLogsFeignClient.addUserLog(logDto); csLogsFeignClient.addUserLog(logDto);
//6.存储设备调试日志表 //6.存储设备调试日志表
@@ -457,7 +455,7 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
//获取版本 //获取版本
String version = csTopicService.getVersion(nDid); String version = csTopicService.getVersion(nDid);
//发起接入请求 //发起接入请求
this.devAccess(nDid,version); this.devAccessAskTemplate(nDid,version,1);
} }
private void checkDeviceStatus(String nDid) { private void checkDeviceStatus(String nDid) {
@@ -511,27 +509,70 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
throw new BusinessException(responseEnum); throw new BusinessException(responseEnum);
} }
public void devAccess(String nDid,String version) { /**
ReqAndResDto.Req reqAndResParam = new ReqAndResDto.Req(); * 装置重新接入系统，需要校验所用的模板
reqAndResParam.setMid(1); * @param nDid
reqAndResParam.setDid(0); * @param version
reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode()); */
reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_5.getCode())); @Transactional(rollbackFor = Exception.class)
reqAndResParam.setExpire(-1); public boolean devAccessAskTemplate(String nDid,String version,Integer mid) {
logger.info("设备接入报文为:" + new Gson().toJson(reqAndResParam)); boolean result = false;
publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(reqAndResParam),1,false); try {
} //询问装置当前所用模板
public void devAccessMid(String nDid,String version,Integer mid) {
ReqAndResDto.Req reqAndResParam = new ReqAndResDto.Req(); ReqAndResDto.Req reqAndResParam = new ReqAndResDto.Req();
reqAndResParam.setMid(mid); reqAndResParam.setMid(mid);
reqAndResParam.setDid(0); reqAndResParam.setDid(0);
reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode()); reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_5.getCode())); reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_3.getCode()));
reqAndResParam.setExpire(-1); reqAndResParam.setExpire(-1);
logger.info("设备接入报文为:" + new Gson().toJson(reqAndResParam)); publisher.send("/Pfm/DevCmd/"+version+"/"+nDid,new Gson().toJson(reqAndResParam),1,false);
publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(reqAndResParam),1,false); //接收到模板，判断模板是否存在，替换模板，发起接入
Thread.sleep(2000);
List<CsModelDto> modelId = objectToList(redisUtil.getObjectByKey(AppRedisKey.MODEL + nDid));
if (CollUtil.isNotEmpty(modelId)) {
CsEquipmentDeliveryVO vo = equipmentFeignClient.queryEquipmentByndid(nDid).getData();
//删除之前的模板信息
csDevModelRelationService.deleteByDeviceId(vo.getId());
//重新录入装置和模板关系信息
for (CsModelDto item : modelId) {
CsDevModelRelationAddParm csDevModelRelationAddParm = new CsDevModelRelationAddParm();
csDevModelRelationAddParm.setDevId(vo.getId());
csDevModelRelationAddParm.setModelId(item.getModelId());
csDevModelRelationAddParm.setDid(item.getDid());
csDevModelRelationService.addDevModelRelation(csDevModelRelationAddParm);
} }
//发起接入
reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_5.getCode()));
publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(reqAndResParam),1,false);
result = true;
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return result;
}
// public void devAccess(String nDid,String version) {
// ReqAndResDto.Req reqAndResParam = new ReqAndResDto.Req();
// reqAndResParam.setMid(1);
// reqAndResParam.setDid(0);
// reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
// reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_5.getCode()));
// reqAndResParam.setExpire(-1);
// logger.info("设备接入报文为:" + new Gson().toJson(reqAndResParam));
// publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(reqAndResParam),1,false);
// }
//
// public void devAccessMid(String nDid,String version,Integer mid) {
// ReqAndResDto.Req reqAndResParam = new ReqAndResDto.Req();
// reqAndResParam.setMid(mid);
// reqAndResParam.setDid(0);
// reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
// reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_5.getCode()));
// reqAndResParam.setExpire(-1);
// logger.info("设备接入报文为:" + new Gson().toJson(reqAndResParam));
// publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(reqAndResParam),1,false);
// }
/** /**
* 平台对设备发起主题询问命令 * 平台对设备发起主题询问命令