代码优化

异常处理

iot-access/access-api/src/main/java/com/njcn/access/pojo/dto/AutoDataDto.java

@@ -1,6 +1,7 @@
 package com.njcn.access.pojo.dto;
 
 import com.alibaba.nacos.shaded.com.google.gson.annotations.SerializedName;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
@@ -17,39 +18,49 @@ import java.util.List;
 public class AutoDataDto {
 
     @SerializedName("Mid")
+    @JsonProperty("Mid")
     private Integer mid;
 
     @SerializedName("Did")
+    @JsonProperty("Did")
     @ApiModelProperty("逻辑设备 治理逻辑设备为1 电能质量设备为2")
     private Integer did;
 
     @SerializedName("Pri")
+    @JsonProperty("Pri")
     private Integer pri;
 
     @SerializedName("Type")
+    @JsonProperty("Type")
     private Integer type;
 
     @SerializedName("Msg")
+    @JsonProperty("Msg")
     private Msg msg;
 
     @Data
     public static class Msg{
 
         @SerializedName("Cldid")
+        @JsonProperty("Cldid")
         @ApiModelProperty("逻辑子设备 治理逻辑设备为0 电能质量设备为1、2")
         private Integer clDid;
 
         @SerializedName("DataType")
+        @JsonProperty("DataType")
         private Integer dataType;
 
         @SerializedName("DataAttr")
+        @JsonProperty("DataAttr")
         @ApiModelProperty("数据属性：无-0、实时-1、统计-2")
         private Integer dataAttr;
 
         @SerializedName("DsNameIdx")
+        @JsonProperty("DsNameIdx")
         private Integer dsNameIdx;
 
         @SerializedName("DataArray")
+        @JsonProperty("DataArray")
         private List<DataArray> dataArray;
     }
 
@@ -57,95 +68,119 @@ public class AutoDataDto {
     public static class DataArray{
 
         @SerializedName("DataAttr")
+        @JsonProperty("DataAttr")
         @ApiModelProperty("数据属性 -1-无 0-Rt(实时) 1-Max 2-Min 3-Avg 4-Cp95")
         private Integer dataAttr;
 
         @SerializedName("DataTimeSec")
+        @JsonProperty("DataTimeSec")
         private Long dataTimeSec;
 
         @SerializedName("DataTimeUSec")
+        @JsonProperty("DataTimeUSec")
         private Integer dataTimeUSec;
 
         @SerializedName("DataTag")
+        @JsonProperty("DataTag")
         @ApiModelProperty("数据是否参与合格率统计")
         private Integer dataTag;
 
         @SerializedName("Code")
+        @JsonProperty("Code")
         @ApiModelProperty("事件码")
         private Integer code;
 
         @SerializedName("Data")
+        @JsonProperty("Data")
         private String data;
 
         @SerializedName("PrjName")
+        @JsonProperty("PrjName")
         @ApiModelProperty("工程名称")
         private String prjName;
 
         @SerializedName("PrjTimeStart")
+        @JsonProperty("PrjTimeStart")
         @ApiModelProperty("装置启动时间")
         private Long prjTimeStart;
 
         @SerializedName("PrjTimeEnd")
+        @JsonProperty("PrjTimeEnd")
         @ApiModelProperty("装置结束时间")
         private Long prjTimeEnd;
 
         @SerializedName("PrjDataPath")
+        @JsonProperty("PrjDataPath")
         @ApiModelProperty("装置数据路径")
         private String prjDataPath;
 
         @SerializedName("DevType")
+        @JsonProperty("DevType")
         @ApiModelProperty("装置型号")
         private String devType;
 
         @SerializedName("DevMac")
+        @JsonProperty("DevMac")
         @ApiModelProperty("装置mac地址")
         private String devMac;
 
         @SerializedName("AppVersion")
+        @JsonProperty("AppVersion")
         @ApiModelProperty("装置程序版本")
         private String appVersion;
 
         @SerializedName("Cldid")
+        @JsonProperty("Cldid")
         @ApiModelProperty("逻辑子设备id")
         private Integer clDid;
 
         @SerializedName("StatCycle")
+        @JsonProperty("StatCycle")
         @ApiModelProperty("统计间隔")
         private Integer statCycle;
 
         @SerializedName("VolGrade")
+        @JsonProperty("VolGrade")
         @ApiModelProperty("电压等级")
         private Float volGrade;
 
         @SerializedName("VolConType")
+        @JsonProperty("VolConType")
         @ApiModelProperty("电压接线方式（0-星型, 1-角型, 2-V型）")
         private Integer volConType;
 
         @SerializedName("CurConSel")
+        @JsonProperty("CurConSel")
         @ApiModelProperty("电流接线方式（0-正常, 1-合成IB, 2-合成IC）")
         private Integer curConSel;
 
         @SerializedName("PtRatio")
+        @JsonProperty("PtRatio")
         @ApiModelProperty("PT变比")
         private Integer ptRatio;
 
         @SerializedName("CtRatio")
+        @JsonProperty("CtRatio")
         @ApiModelProperty("ct变比")
         private Integer ctRatio;
 
         @SerializedName("CapacitySscb")
+        @JsonProperty("CapacitySscb")
         @ApiModelProperty("基准短路容量")
         private Float capacitySscb;
 
         @SerializedName("CapacitySscmin")
+        @JsonProperty("CapacitySscmin")
         @ApiModelProperty("最小短路容量")
         private Float capacitySscmin;
 
         @SerializedName("CapacitySt")
+        @JsonProperty("CapacitySt")
         @ApiModelProperty("供电设备容量")
         private Float capacitySt;
 
         @SerializedName("CapacitySi")
+        @JsonProperty("CapacitySi")
         @ApiModelProperty("用户协议容量")
         private Float capacitySi;
     }

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

@@ -328,17 +328,18 @@ public class MqttMessageHandler {
                                 onlineLogsService.updateById(record);
                             }
                         }
-                        //接入后系统重置装置心跳
-                        redisUtil.saveByKeyWithExpire("MQTT:" + nDid, Instant.now().toEpochMilli(),180L);
-                        //修改redis的mid
-                        redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,1);
-                        redisUtil.saveByKeyWithExpire("online" + nDid,"online",10L);
                         //询问设备软件信息
                         askDevData(nDid,version,1,mid);
                         //更新治理监测点信息和设备容量
                         askDevData(nDid,version,2,(res.getMid()+1));
                         //更新电网侧、负载侧监测点信息
                         askDevData(nDid,version,3,(res.getMid()+1));
+                        //接入后系统重置装置心跳
+                        redisUtil.saveByKeyWithExpire("MQTT:" + nDid, Instant.now().toEpochMilli(),180L);
+                        //修改redis的mid
+                        redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,1);
+                        //接入成功标识
+                        redisUtil.saveByKeyWithExpire("online" + nDid,"online",10L);
                         //录波任务倒计时
                         redisUtil.saveByKeyWithExpire("startFile:" + nDid,null,60L);
                     } else {

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

@@ -9,7 +9,6 @@ import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
 import com.njcn.system.api.DictTreeFeignClient;
 import com.njcn.system.enums.DicDataEnum;
-import com.njcn.system.enums.DicTreeEnum;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.ApplicationArguments;
@@ -88,20 +87,28 @@ public class AccessApplicationRunner implements ApplicationRunner {
     }
 
     public void accessDev(List<CsEquipmentDeliveryPO> list) {
+        if (CollUtil.isNotEmpty(list)) {
+            try {
                 list.forEach(item->{
                     System.out.println(Thread.currentThread().getName() + ": reboot : nDid : " + item.getNdid());
-            String version = csTopicService.getVersion(item.getNdid());
-            if (!Objects.isNull(version)){
                     //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入
                     String code = dictTreeFeignClient.queryById(item.getDevType()).getData().getCode();
                     if (Objects.equals(code, DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(),1)) {
-                    csDeviceService.wlDevRegister(item.getNdid());
+                        //csDeviceService.wlDevRegister(item.getNdid());
+                        log.info("请先手动注册、接入");
                     } else {
-                    csDeviceService.devAccessAskTemplate(item.getNdid(),version,1);
+                        String version = csTopicService.getVersion(item.getNdid());
+                        if (Objects.isNull(version)) {
+                            version = "V1";
+                        }
+                        csDeviceService.autoAccess(item.getNdid(),version,1);
                     }
                     redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(),1);
-            }
                 });
+            } catch (Exception e) {
+                log.error(e.getMessage());
+            }
+        }
     }
 
 }

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

@@ -9,7 +9,6 @@ import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
 import com.njcn.system.api.DictTreeFeignClient;
 import com.njcn.system.enums.DicDataEnum;
-import com.njcn.system.enums.DicTreeEnum;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.ApplicationArguments;
@@ -17,6 +16,7 @@ import org.springframework.boot.ApplicationRunner;
 import org.springframework.stereotype.Component;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.*;
@@ -45,57 +45,72 @@ public class AutoAccessTimer implements ApplicationRunner {
             if (CollUtil.isNotEmpty(list)) {
                 ExecutorService executor = Executors.newFixedThreadPool(10);
                 // 将任务平均分配给10个子列表
-                List<List<CsEquipmentDeliveryPO>> subLists = new ArrayList<>();
+                List<List<CsEquipmentDeliveryPO>> subLists = CollUtil.split(list, 10);
-                int partitionSize = list.size() / 10;
+//                List<List<CsEquipmentDeliveryPO>> subLists = new ArrayList<>();
-                for (int i = 0; i < 10; i++) {
+//                int partitionSize = list.size() / 10;
-                    int start = i * partitionSize;
+//                for (int i = 0; i < 10; i++) {
-                    int end = (i == 9) ? list.size() : start + partitionSize;
+//                    int start = i * partitionSize;
-                    subLists.add(list.subList(start, end));
+//                    int end = (i == 9) ? list.size() : start + partitionSize;
-                }
+//                    subLists.add(list.subList(start, end));
-
+//                }
                 // 创建一个ExecutorService来处理这些任务
                 List<Future<Void>> futures = new ArrayList<>();
                 // 提交任务给线程池执行
-                for (int i = 0; i < 10; i++) {
+//                for (int i = 0; i < 10; i++) {
-                    int index = i;
+//                    int index = i;
-                    futures.add(executor.submit(new Callable<Void>() {
+//                    futures.add(executor.submit(() -> {
-                        @Override
+//                        accessDev(subLists.get(index));
-                        public Void call() {
+//                        return null;
-                            accessDev(subLists.get(index));
+//                    }));
+//                }
+                for (List<CsEquipmentDeliveryPO> subList : subLists) {
+                    futures.add(executor.submit(() -> {
+                        accessDev(subList);
                         return null;
-                        }
                     }));
                 }
                 // 等待所有任务完成
                 for (Future<Void> future : futures) {
                     try {
                         future.get();
-                    } catch (InterruptedException | ExecutionException e) {
+                    } catch (InterruptedException e) {
-                        throw new RuntimeException(e);
+                        Thread.currentThread().interrupt();
+                        log.error("任务被中断", e);
+                    } catch (ExecutionException e) {
+                        log.error("任务执行异常", e.getCause());
                     }
                 }
+
                 // 关闭ExecutorService
                 executor.shutdown();
             }
         };
-        //第一次执行的时间为120s，然后每隔120s执行一次
+        //第一次执行的时间为120s，然后在前一个任务执行完毕后，等待120s再执行下一个任务
-        scheduler.scheduleAtFixedRate(task,AUTO_TIME,AUTO_TIME,TimeUnit.SECONDS);
+        scheduler.scheduleWithFixedDelay(task, AUTO_TIME, AUTO_TIME, TimeUnit.SECONDS);
     }
 
     public void accessDev(List<CsEquipmentDeliveryPO> list) {
+        if (CollUtil.isNotEmpty(list)) {
+            try {
                 list.forEach(item -> {
                     System.out.println(Thread.currentThread().getName() + ": auto : nDid : " + item.getNdid());
-            String version = csTopicService.getVersion(item.getNdid());
-            if (!Objects.isNull(version)){
                     //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入
                     String code = dictTreeFeignClient.queryById(item.getDevType()).getData().getCode();
                     if (Objects.equals(code, DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(), 1)) {
-                    csDeviceService.wlDevRegister(item.getNdid());
+                        //csDeviceService.wlDevRegister(item.getNdid());
+                        log.info("请先手动注册、接入");
                     } else {
-                    csDeviceService.devAccessAskTemplate(item.getNdid(),version,1);
+                        String version = csTopicService.getVersion(item.getNdid());
+                        if (Objects.isNull(version)) {
+                            version = "V1";
+                        }
+                        csDeviceService.autoAccess(item.getNdid(), version, 1);
                     }
                     redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1);
-            }
                 });
+            } catch (Exception e) {
+                log.error(e.getMessage());
+            }
+        }
     }
 }

iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDevModelServiceImpl.java

@@ -91,6 +91,12 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
             json = JsonUtil.convertStreamToString(file.getInputStream());
             Gson gson = new Gson();
             TemplateDto templateDto = gson.fromJson(json, TemplateDto.class);
+            //判断设备型号
+            String devType = templateDto.getDevType();
+            DictTreeVO dictTreeVO = dictTreeFeignClient.queryByCode(devType).getData();
+            if (Objects.isNull(dictTreeVO)){
+                throw new BusinessException(AccessResponseEnum.DEV_TYPE_NOT_FIND);
+            }
             logDto.setOperate("新增设备模板,模板名称:" + templateDto.getDevType());
             //模板文件存入文件服务器
             String filePath = fileStorageUtil.uploadMultipart(file, OssPath.DEV_MODEL + templateDto.getDevType() + "_");
@@ -838,9 +844,6 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
         List<DataSetDto> dataSetList = templateDto.getDataSet();
         String devType = templateDto.getDevType();
         DictTreeVO dictTreeVO = dictTreeFeignClient.queryByCode(devType).getData();
-        if (Objects.isNull(dictTreeVO)){
-            throw new BusinessException(AccessResponseEnum.DEV_TYPE_NOT_FIND);
-        }
         String code = dictTreeFeignClient.queryById(dictTreeVO.getPid()).getData().getCode();
         //逻辑设备录入
         if (CollectionUtil.isNotEmpty(dataSetList)){

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

@@ -38,6 +38,7 @@ import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -370,6 +371,7 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
     }
 
     @Override
+//    @Transactional(rollbackFor = Exception.class)
     public String wlDevRegister(String nDid) {
         String result = "fail";
         // 设备状态判断
@@ -444,15 +446,6 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
             csEquipmentDeliveryService.updateProcessBynDid(nDid,4);
             //7.存储日志
             csLogsFeignClient.addUserLog(logDto);
-            //8.存储设备调试日志表
-            CsEquipmentProcessPO csEquipmentProcess = new CsEquipmentProcessPO();
-            csEquipmentProcess.setDevId(nDid);
-            csEquipmentProcess.setOperator(RequestUtil.getUserIndex());
-            csEquipmentProcess.setStartTime(LocalDateTime.now());
-            csEquipmentProcess.setEndTime(LocalDateTime.now());
-            csEquipmentProcess.setProcess(4);
-            csEquipmentProcess.setStatus(1);
-            processFeignClient.add(csEquipmentProcess);
             //9.删除redis监测点模板信息
             redisUtil.delete(AppRedisKey.MODEL + nDid);
             redisUtil.delete(AppRedisKey.LINE + nDid);
@@ -604,6 +597,77 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
         return result;
     }
 
+    /**
+     * 装置重新接入系统，需要校验所用的模板
+     * @param nDid
+     * @param version
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public boolean autoAccess(String nDid,String version,Integer mid) {
+        String clientName = "NJCN-" + nDid.substring(nDid.length() - 6);
+        boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
+        if (!mqttClient) {
+            DeviceLogDTO logDto = new DeviceLogDTO();
+            logDto.setUserName("运维管理员");
+            logDto.setLoginName("njcnyw");
+            logDto.setResult(1);
+            logDto.setOperate(nDid + "接入失败，装置客户端不在线");
+            csLogsFeignClient.addUserLog(logDto);
+        }
+        boolean result = false;
+        Map<Integer,String> modelMap = new HashMap<>();
+        try {
+            //删除缓存数据
+            redisUtil.delete(AppRedisKey.MODEL + nDid);
+            redisUtil.deleteKeysByString(AppRedisKey.DEV_MODEL);
+            //询问装置当前所用模板
+            publisher.send("/Pfm/DevCmd/"+version+"/"+nDid,new Gson().toJson(getJson(mid,TypeEnum.TYPE_3.getCode())),1,false);
+            //接收到模板，判断模板是否存在，替换模板，发起接入
+            Thread.sleep(2000);
+            List<CsModelDto> modelId = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.MODEL + nDid),CsModelDto.class);
+            if (CollUtil.isNotEmpty(modelId)) {
+                CsEquipmentDeliveryVO vo = equipmentFeignClient.queryEquipmentByndid(nDid).getData();
+                //重新录入装置和模板关系信息
+                for (CsModelDto item : modelId) {
+                    CsDevModelRelationPO po = new CsDevModelRelationPO();
+                    po.setDevId(vo.getId());
+                    po.setModelId(item.getModelId());
+                    po.setDid(item.getDid());
+                    po.setUpdateTime(LocalDateTime.now());
+                    csDevModelRelationService.addRelation(po);
+                    modelMap.put(item.getType(),item.getModelId());
+                }
+                //修改监测点使用的模板和数据集
+                List<CsLinePO> lineList;
+                Object object = redisUtil.getObjectByKey("accessLineInfo:" + nDid);
+                if (Objects.isNull(object)) {
+                    lineList = csLineFeignClient.findByNdid(nDid).getData();
+                    for (CsLinePO item : lineList) {
+                        if (item.getClDid() == 0) {
+                            updateLineIds(modelMap.get(0),item.getClDid(),nDid);
+                        } else {
+                            updateLineIds(modelMap.get(1),item.getClDid(),nDid);
+                        }
+                    }
+                }
+                //发起接入
+                publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(getJson(mid,TypeEnum.TYPE_5.getCode())),1,false);
+                //录波任务倒计时
+                redisUtil.saveByKeyWithExpire("startFile:" + nDid,null,60L);
+                result = true;
+            }
+        } catch (Exception e) {
+            DeviceLogDTO logDto = new DeviceLogDTO();
+            logDto.setUserName("运维管理员");
+            logDto.setLoginName("njcnyw");
+            logDto.setResult(1);
+            logDto.setOperate(nDid + "装置接入失败");
+            csLogsFeignClient.addUserLog(logDto);
+            throw new BusinessException(e.getMessage());
+        }
+        return result;
+    }
+
     /**
      * 组装报文
      */

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

@@ -1,5 +1,6 @@
 package com.njcn.access.service.impl;
 
+import cn.hutool.core.collection.CollUtil;
 import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.njcn.access.enums.AccessEnum;
@@ -97,6 +98,7 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
                 .ne(CsEquipmentDeliveryPO::getRunStatus,AccessEnum.DEL.getCode())
                 .eq(CsEquipmentDeliveryPO::getUsageStatus,1)
                 .list();
+        if (CollUtil.isNotEmpty(list)) {
             list.forEach(item->{
                 String clientName = "NJCN-" + item.getNdid().substring(item.getNdid().length() - 6);
                 boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
@@ -111,6 +113,7 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
                     csLogsFeignClient.addUserLog(logDto);
                 }
             });
+        }
         return result;
     }
 
@@ -121,6 +124,7 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
                 .eq(CsEquipmentDeliveryPO::getRunStatus,AccessEnum.OFFLINE.getCode())
                 .eq(CsEquipmentDeliveryPO::getUsageStatus,1)
                 .list();
+        if (CollUtil.isNotEmpty(list)) {
             list.forEach(item->{
                 String clientName = "NJCN-" + item.getNdid().substring(item.getNdid().length() - 6);
                 boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
@@ -135,6 +139,7 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
                     csLogsFeignClient.addUserLog(logDto);
                 }
             });
+        }
         return result;
     }
 

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

@@ -59,7 +59,7 @@ public class AppAutoDataConsumer extends EnhanceConsumerMessageHandler<AppAutoDa
                 rtFeignClient.analysis(appAutoDataMessage);
                 break;
             case 2:
-                log.info(appAutoDataMessage.getKey() + "分发至统计数据");
+                log.info("{}分发至统计数据", appAutoDataMessage.getKey());
                 statFeignClient.analysis(appAutoDataMessage);
                 break;
             default: