refactor(access): 重构设备接入任务和优化线程管理

- 移除废弃的日志消息相关依赖和代码
- 优化 AccessApplicationRunner 和 AutoAccessTimer 的线程池使用
- 采用共享线程池和信号量控制并发访问设备
- 批量处理设备接入任务并添加批次间隔控制
- 重构 autoAccess2 方法移除事务注解并改用异步处理
- 修复更新线路ID时参数传递错误的问题
- 统一错误处理和异常捕获机制
- 添加更详细的日志记录和错误追踪功能

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

@@ -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","平台重新上送文件异常"),
 
@@ -75,6 +75,7 @@ public enum AccessResponseEnum {
     FILE_CHECK_ERROR("A0312","文件校验码不一致!"),
 
     CLD_MODEL_EXIST("A0313","云前置模板已存在,请先删除再录入!"),
+    DEV_DATA_ERROR("A0313","系统端询问装置端监测点信息失败，无法生成监测点!"),
 
     /**
      * A3001 ~ A3099 用于zlevent模块的枚举

iot-access/access-api/src/main/java/com/njcn/access/utils/ChannelObjectUtil.java

@@ -5,7 +5,8 @@ import com.njcn.redis.utils.RedisUtil;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.List;
 
 /**
  * @author xy
@@ -51,22 +52,4 @@ public class ChannelObjectUtil {
     public Object getDeviceMid(String nDid) {
         return redisUtil.getObjectByKey(AppRedisKey.DEVICE_MID + nDid);
     }
-
-
-    public Map<String, List<String>> objectToMap(Object obj) {
-        // 创建并填充 Map
-        Map<String, List<String>> resultMap = new HashMap<>();
-        String json = obj.toString();
-        // 移除首尾的 {}
-        json = json.substring(1, json.length() - 1);
-        // 找到键和值的分隔符位置
-        int keyEndIndex = json.indexOf("=[");
-        String key = json.substring(0, keyEndIndex);
-        String valuesStr = json.substring(keyEndIndex + 2, json.length() - 1);
-        // 将值字符串分割成列表
-        String[] valuesArray = valuesStr.split(", ");
-        List<String> valuesList = Arrays.asList(valuesArray);
-        resultMap.put(key, valuesList);
-        return resultMap;
-    }
 }

iot-access/access-boot/src/main/java/com/njcn/access/config/MqttAccessSchedulerConfig.java

@@ -0,0 +1,52 @@
+package com.njcn.access.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.concurrent.*;
+
+/**
+ * MQTT接入任务共享资源池配置
+ */
+@Configuration
+@Slf4j
+public class MqttAccessSchedulerConfig {
+
+    /**
+     * 全局MQTT发送并发上限
+     * 建议设为 MqttConnectOptions.maxInflight 的 50%-70%
+     */
+    @Value("${mqtt.access.concurrency-limit:200}")
+    private int concurrencyLimit;
+
+    @Value("${mqtt.access.worker-threads:5}")
+    private int workerThreads;
+
+    @Bean(destroyMethod = "shutdownNow")
+    public ScheduledExecutorService sharedScheduler() {
+        return Executors.newSingleThreadScheduledExecutor(r -> {
+            Thread t = new Thread(r, "mqtt-access-scheduler");
+            t.setDaemon(true);
+            return t;
+        });
+    }
+
+    @Bean(destroyMethod = "shutdownNow")
+    public ExecutorService sharedWorkerPool() {
+        return new ThreadPoolExecutor(
+                workerThreads, workerThreads,
+                0L, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(200),
+                r -> new Thread(r, "mqtt-access-worker"),
+                new ThreadPoolExecutor.CallerRunsPolicy() // 队列满时调用者线程执行，天然背压
+        );
+    }
+
+    @Bean
+    public Semaphore mqttSemaphore() {
+        log.info("初始化MQTT接入信号量, 并发上限={}", concurrencyLimit);
+        return new Semaphore(concurrencyLimit);
+    }
+}

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

@@ -159,5 +159,15 @@ public class CsDeviceController extends BaseController {
         return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, result, methodDescribe);
     }
 
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/accessByUpdateMac")
+    @ApiOperation("修改mac后手动接入")
+    @ApiImplicitParam(name = "nDid", value = "设备识别码", required = true)
+    public HttpResult<String> accessByUpdateMac(@RequestParam("nDid") String nDid){
+        String methodDescribe = getMethodDescribe("accessByUpdateMac");
+        String result = csDeviceService.accessByUpdateMac(nDid);
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, result, methodDescribe);
+    }
+
 
 }

iot-access/access-boot/src/main/java/com/njcn/access/controller/CsHeartController.java

@@ -17,7 +17,8 @@ import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
-import springfox.documentation.annotations.ApiIgnore;
+
+import java.util.List;
 
 /**
  * 类的介绍：
@@ -31,7 +32,6 @@ import springfox.documentation.annotations.ApiIgnore;
 @RequestMapping("/heartbeat")
 @Api(tags = "心跳")
 @AllArgsConstructor
-@ApiIgnore
 public class CsHeartController extends BaseController {
 
     private final ICsHeartService csHeartService;
@@ -45,4 +45,22 @@ public class CsHeartController extends BaseController {
         csHeartService.handleHeartbeat(message);
         return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
     }
+
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/getMissHeartbeat")
+    @ApiOperation("获取物联心跳丢失状态在线的设备")
+    public HttpResult<List<String>> getMissHeartbeat(){
+        String methodDescribe = getMethodDescribe("getMissHeartbeat");
+        List<String> result = csHeartService.getMissHeartbeat();
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, result, methodDescribe);
+    }
+
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/updateDevStatus")
+    @ApiOperation("手动调整状态")
+    public HttpResult<String> updateDevStatus(){
+        String methodDescribe = getMethodDescribe("updateDevStatus");
+        csHeartService.updateDevStatus();
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
+    }
 }

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

@@ -22,9 +22,11 @@ import com.njcn.access.pojo.dto.file.FileRedisDto;
 import com.njcn.access.pojo.param.ReqAndResParam;
 import com.njcn.access.pojo.po.CsLineModel;
 import com.njcn.access.pojo.po.CsTopic;
-import com.njcn.access.service.*;
+import com.njcn.access.service.ICsEquipmentDeliveryService;
+import com.njcn.access.service.ICsLineModelService;
+import com.njcn.access.service.ICsTopicService;
+import com.njcn.access.service.IHeartbeatService;
 import com.njcn.access.utils.ChannelObjectUtil;
-import com.njcn.common.pojo.dto.DeviceLogDTO;
 import com.njcn.common.pojo.exception.BusinessException;
 import com.njcn.csdevice.api.*;
 import com.njcn.csdevice.pojo.dto.PqsCommunicateDto;
@@ -36,10 +38,8 @@ import com.njcn.device.biz.utils.COverlimitUtil;
 import com.njcn.mq.message.AppAutoDataMessage;
 import com.njcn.mq.message.AppEventMessage;
 import com.njcn.mq.message.AppFileMessage;
-import com.njcn.mq.template.AppAutoDataMessageTemplate;
+import com.njcn.mq.message.LogMessage;
-import com.njcn.mq.template.AppEventMessageTemplate;
+import com.njcn.mq.template.*;
-import com.njcn.mq.template.AppFileMessageTemplate;
-import com.njcn.mq.template.AppFileStreamMessageTemplate;
 import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
 import com.njcn.rt.api.RtFeignClient;
@@ -86,10 +86,8 @@ public class MqttMessageHandler {
     private final DataSetFeignClient dataSetFeignClient;
     private final AppAutoDataMessageTemplate appAutoDataMessageTemplate;
     private final AppEventMessageTemplate appEventMessageTemplate;
-    private final CsLogsFeignClient csLogsFeignClient;
     private final AppFileMessageTemplate appFileMessageTemplate;
     private final AppFileStreamMessageTemplate appFileStreamMessageTemplate;
-    private final ICsDeviceOnlineLogsService onlineLogsService;
     private final CsSoftInfoFeignClient csSoftInfoFeignClient;
     private final CsLineFeignClient csLineFeignClient;
     private final DevCapacityFeignClient devCapacityFeignClient;
@@ -100,31 +98,29 @@ public class MqttMessageHandler {
     private final RtFeignClient rtFeignClient;
     private final CsCommunicateFeignClient csCommunicateFeignClient;
     private final IHeartbeatService heartbeatService;
+    private final LogMessageTemplate logMessageTemplate;
+    private final CsDeviceRegistryFeignClient csDeviceRegistryFeignClient;
     @Autowired
     Validator validator;
 
     @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){
-        //日志记录
-        DeviceLogDTO logDto = new DeviceLogDTO();
-        try{
-            logDto.setUserName("运维管理员");
-            logDto.setLoginName("njcnyw");
-        } catch (Exception e) {
-            logDto.setUserName("运维管理员");
-            logDto.setLoginName("njcnyw");
-        }
-        logDto.setOperate(nDid + "设备主题录入");
-        logDto.setResult(1);
         //业务流程开始
         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+"发送的主题信息，code = " + res.getCode());
+        logDto.setResult(1);
+        logMessageTemplate.sendMember(logDto);
         //检验传递的参数是否准确
         Set<ConstraintViolation<ReqAndResParam.Res>> validate = validator.validate(res);
-        validate.forEach(constraintViolation -> {
+//        validate.forEach(constraintViolation -> {
-            System.out.println(constraintViolation.getMessage());
+//            System.out.println(constraintViolation.getMessage());
-        });
+//        });
         if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){
             if (Objects.equals(res.getType(), Integer.parseInt(TypeEnum.TYPE_16.getCode()))){
                 List<CsTopic> list = new ArrayList<>();
@@ -141,18 +137,20 @@ public class MqttMessageHandler {
                     list.add(csTopic);
                 });
                 csTopicService.addTopic(nDid,list);
-                csLogsFeignClient.addUserLog(logDto);
+                String version = list.stream().map(CsTopic::getVersion).filter(Objects::nonNull).findFirst().orElse("V1");
+                redisUtil.saveByKeyWithExpire(nDid +":version",version,30L);
+                logMessageTemplate.sendMember(logDto);
             } else {
                 logDto.setResult(0);
                 logDto.setFailReason(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage());
-                csLogsFeignClient.addUserLog(logDto);
+                logMessageTemplate.sendMember(logDto);
-                log.info(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage());
+                //log.info(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage());
             }
         } else {
             logDto.setResult(0);
             logDto.setFailReason(AccessResponseEnum.RESPONSE_ERROR.getMessage());
-            csLogsFeignClient.addUserLog(logDto);
+            logMessageTemplate.sendMember(logDto);
-            log.info(AccessResponseEnum.RESPONSE_ERROR.getMessage());
+            //log.info(AccessResponseEnum.RESPONSE_ERROR.getMessage());
         }
     }
 
@@ -168,20 +166,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);
-        //日志记录
-        DeviceLogDTO logDto = new DeviceLogDTO();
-        try{
-            logDto.setUserName("运维管理员");
-            logDto.setLoginName("njcnyw");
-        } catch (Exception e) {
-            logDto.setUserName("运维管理员");
-            logDto.setLoginName("njcnyw");
-        }
-        logDto.setOperate("收到设备"+nDid+"注册应答响应");
-        logDto.setResult(1);
         //业务处理
         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+"注册应答响应，code = " + res.getCode());
+        logDto.setResult(1);
+        logMessageTemplate.sendMember(logDto);
         if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){
             if (Objects.equals(res.getType(),Integer.parseInt(TypeEnum.TYPE_17.getCode()))){
                 //询问模板数据
@@ -193,18 +187,23 @@ public class MqttMessageHandler {
                 reqAndResParam.setExpire(-1);
                 String version = csTopicService.getVersion(nDid);
                 publisher.send("/Pfm/DevCmd/"+version+"/"+nDid,new Gson().toJson(reqAndResParam),1,false);
-                csLogsFeignClient.addUserLog(logDto);
+                //记录日志
+                logDto.setUserIndex("系统");
+                logDto.setLoginName("系统");
+                logDto.setOperate("注册阶段：系统端向装置端"+nDid+"发送询问模板请求");
+                logDto.setResult(1);
+                logMessageTemplate.sendMember(logDto);
             } else {
                 logDto.setResult(0);
                 logDto.setFailReason(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage());
-                csLogsFeignClient.addUserLog(logDto);
+                logMessageTemplate.sendMember(logDto);
-                log.info(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage());
+                //log.info(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage());
             }
         } else {
             logDto.setResult(0);
             logDto.setFailReason(AccessResponseEnum.REGISTER_RESPONSE_ERROR.getMessage());
-            csLogsFeignClient.addUserLog(logDto);
+            logMessageTemplate.sendMember(logDto);
-            log.info(AccessResponseEnum.REGISTER_RESPONSE_ERROR.getMessage());
+            //log.info(AccessResponseEnum.REGISTER_RESPONSE_ERROR.getMessage());
         }
     }
 
@@ -218,22 +217,17 @@ 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 {
         //日志实体
-        DeviceLogDTO logDto = new DeviceLogDTO();
+        LogMessage logDto = new LogMessage();
-        try{
+        logDto.setUserIndex("系统");
-            logDto.setUserName("运维管理员");
+        logDto.setLoginName("系统");
-            logDto.setLoginName("njcnyw");
-        } catch (Exception e) {
-            logDto.setUserName("运维管理员");
-            logDto.setLoginName("njcnyw");
-        }
         logDto.setResult(1);
         //业务处理
         Gson gson = new Gson();
         ReqAndResDto.Res res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResDto.Res.class);
-        redisUtil.saveByKeyWithExpire("devResponse",res.getCode(),5L);
+        //redisUtil.saveByKeyWithExpire("devResponse",res.getCode(),5L);
-        if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){
+        if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())) {
             switch (res.getType()){
                 /**
                  * 装置类型模板应答
@@ -242,17 +236,21 @@ public class MqttMessageHandler {
                  * 3.平台端需读取装置的DevMod来判断网关支持的设备模板（包含设备型号和模板版本），根据app提交的接入子设备DID匹配数据模板（型号及版本），生成DevCfg下发给网关，网关根据下发信息生成就地设备点表。
                  */
                 case 4611:
-                    log.info("{},装置模板应答,应答code {}",nDid,res.getCode());
+                    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();
                     List<DevCfgDto> list2 = modelDto.getMsg().getDevCfg();
-                    if (CollectionUtils.isEmpty(list)){
+                    if (CollectionUtils.isEmpty(list)) {
-                        log.error(AccessResponseEnum.MODEL_VERSION_ERROR.getMessage());
+                        //log.error(AccessResponseEnum.MODEL_VERSION_ERROR.getMessage());
+                        logDto.setOperate("查看装置端模板报文数据");
                         logDto.setResult(0);
                         logDto.setFailReason(AccessResponseEnum.MODEL_VERSION_ERROR.getMessage());
-                        csLogsFeignClient.addUserLog(logDto);
+                        logMessageTemplate.sendMember(logDto);
                         //有异常删除缓存的模板信息
                         redisUtil.delete(AppRedisKey.MODEL + nDid);
+                        log.error("{}{}", nDid, AccessResponseEnum.MODEL_VERSION_ERROR.getMessage());
                         throw new BusinessException(AccessResponseEnum.MODEL_VERSION_ERROR);
                     }
                     //校验前置传递的装置模板库中是否存在
@@ -267,11 +265,11 @@ public class MqttMessageHandler {
                         CsModelDto csModelDto = new CsModelDto();
                         CsDevModelPO po = devModelFeignClient.findModel(item.getDevType(),item.getVersionNo(),item.getVersionDate()).getData();
                         if (Objects.isNull(po)){
-                            log.error(AccessResponseEnum.MODEL_NO_FIND.getMessage());
+                            //log.error(AccessResponseEnum.MODEL_NO_FIND.getMessage());
-                            logDto.setOperate(nDid + "模板缺失");
+                            logDto.setOperate(nDid + "查询系统中是否存在模板");
                             logDto.setResult(0);
                             logDto.setFailReason(AccessResponseEnum.MODEL_NO_FIND.getMessage());
-                            csLogsFeignClient.addUserLog(logDto);
+                            logMessageTemplate.sendMember(logDto);
                             //有异常删除缓存的模板信息
                             redisUtil.delete(AppRedisKey.MODEL + nDid);
                             throw new BusinessException(AccessResponseEnum.MODEL_NO_FIND);
@@ -279,9 +277,10 @@ public class MqttMessageHandler {
                         if (Objects.equals(po.getType(),0)){
                             List<CsDataSet> dataSetList = dataSetFeignClient.getModuleDataSet(po.getId()).getData();
                             if (CollectionUtils.isEmpty(dataSetList)){
+                                logDto.setOperate("查看APF模块个数");
                                 logDto.setResult(0);
                                 logDto.setFailReason(AccessResponseEnum.MODULE_NUMBER_IS_NULL.getMessage());
-                                csLogsFeignClient.addUserLog(logDto);
+                                logMessageTemplate.sendMember(logDto);
                                 //有异常删除缓存的模板信息
                                 redisUtil.delete(AppRedisKey.MODEL + nDid);
                                 throw new BusinessException(AccessResponseEnum.MODULE_NUMBER_IS_NULL);
@@ -296,16 +295,19 @@ public class MqttMessageHandler {
                     });
                     //存储模板id
                     String key2 = AppRedisKey.MODEL + nDid;
-                    redisUtil.saveByKeyWithExpire(key2,modelList,600L);
+                    redisUtil.saveByKey(key2,modelList);
                     //存储监测点模板信息，用于界面回显
                     List<String> modelId = modelList.stream().map(CsModelDto::getModelId).collect(Collectors.toList());
                     List<CsLineModel> lineList = csLineModelService.getMonitorNumByModelId(modelId);
                     String key = AppRedisKey.LINE + nDid;
                     redisUtil.saveByKeyWithExpire(key,lineList,600L);
+                    logMessageTemplate.sendMember(logDto);
                     break;
                 case 4613:
-                    logDto.setOperate(nDid + "设备接入");
+                    logDto.setOperate("系统端收到装置端"+nDid+"接入应答，code = " + res.getCode());
-                    log.info("{},收到接入应答响应,应答code {}",nDid,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;
                         //修改装置状态
@@ -333,20 +335,21 @@ public class MqttMessageHandler {
                         //录波任务倒计时
                         redisUtil.saveByKeyWithExpire("startFile:" + nDid,null,60L);
                     } else {
-                        log.info(AccessResponseEnum.ACCESS_RESPONSE_ERROR.getMessage());
+                        //log.info(AccessResponseEnum.ACCESS_RESPONSE_ERROR.getMessage());
                         logDto.setResult(0);
                         logDto.setFailReason(AccessResponseEnum.ACCESS_RESPONSE_ERROR.getMessage());
-                        csLogsFeignClient.addUserLog(logDto);
+                        logMessageTemplate.sendMember(logDto);
                         throw new BusinessException(AccessResponseEnum.ACCESS_RESPONSE_ERROR);
                     }
-                    csLogsFeignClient.addUserLog(logDto);
+                    logMessageTemplate.sendMember(logDto);
                     break;
                 case 4614:
                     RspDataDto rspDataDto = JSON.parseObject(JSON.toJSONString(res.getMsg()), RspDataDto.class);
                     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();
@@ -354,11 +357,14 @@ public class MqttMessageHandler {
                                 String id = IdUtil.fastSimpleUUID();
                                 csSoftInfoPo.setId(id);
                                 DateTimeFormatter formatter = new DateTimeFormatterBuilder()
-                                        .appendPattern("yyyy-MM-dd[[HH][:mm][:ss]]")
+                                        // 优先尝试紧凑格式
+                                        .optionalStart().appendPattern("yyyyMMdd").optionalEnd()
+                                        // 再尝试带横线格式
+                                        .optionalStart().appendPattern("yyyy-MM-dd").optionalEnd()
+                                        // 默认时间部分
                                         .parseDefaulting(ChronoField.HOUR_OF_DAY, 0)
                                         .parseDefaulting(ChronoField.MINUTE_OF_HOUR, 0)
                                         .parseDefaulting(ChronoField.SECOND_OF_MINUTE, 0)
-                                        .parseDefaulting(ChronoField.MILLI_OF_SECOND, 0)
                                         .toFormatter();
                                 LocalDateTime localDateTime = LocalDateTime.parse(softInfo.getAppDate(), formatter);
                                 assertThat(localDateTime).isNotNull();
@@ -371,22 +377,41 @@ public class MqttMessageHandler {
                                     csSoftInfoFeignClient.removeSoftInfo(soft);
                                 }
                                 equipmentFeignClient.updateSoftInfo(nDid,csSoftInfoPo.getId());
+                                logMessageTemplate.sendMember(logDto);
                                 break;
                             case 2:
                                 List<RspDataDto.LdevInfo> devInfo = JSON.parseArray(JSON.toJSONString(rspDataDto.getDataArray()), RspDataDto.LdevInfo.class);
                                 if (CollectionUtil.isNotEmpty(devInfo)){
                                     if (Objects.equals(res.getDid(),1)){
+                                        redisUtil.saveByKeyWithExpire("lineInfo:"+nDid,devInfo,30L);
                                         List<CsDevCapacityPO> list3 = new ArrayList<>();
                                         boolean hasZeroClDid = devInfo.stream().anyMatch(item -> item.getClDid() == 0);
                                         //治理设备
                                         if (hasZeroClDid) {
+                                            logDto.setOperate("系统端收到装置端"+nDid+"更新APF容量报文，code = " + res.getCode());
+                                            logDto.setResult(1);
                                             devInfo.forEach(item->{
                                                 if (Objects.equals(item.getClDid(),0)){
                                                     updateLineInfo(nDid,item);
                                                 }
                                                 //2.录入各个模块设备容量
+                                                CsDeviceRegistry csDeviceRegistry = csDeviceRegistryFeignClient.queryByCurrentNdidAndClDid(nDid, 0).getData();
+                                                String lineId;
+                                                if (Objects.isNull(csDeviceRegistry)) {
+                                                    List<CsLinePO> lines = csLineFeignClient.findByNdid(nDid).getData();
+                                                    if (CollectionUtil.isEmpty(lines)) {
+                                                        throw new BusinessException("通过装置NDID获取监测点信息失败");
+                                                    }
+                                                    CsLinePO line = lines.stream().filter(item2->Objects.equals(item2.getClDid(),0)).findFirst().orElse(null);
+                                                    if (Objects.isNull(line)) {
+                                                        throw new BusinessException("通过逻辑子设备ID获取监测点信息失败");
+                                                    }
+                                                    lineId = line.getLineId();
+                                                } else {
+                                                    lineId = csDeviceRegistry.getId();
+                                                }
                                                 CsDevCapacityPO csDevCapacity = new CsDevCapacityPO();
-                                                csDevCapacity.setLineId(nDid.concat("0"));
+                                                csDevCapacity.setLineId(lineId);
                                                 csDevCapacity.setCldid(item.getClDid());
                                                 csDevCapacity.setCapacity(Objects.isNull(item.getCapacityA())?0.0:item.getCapacityA());
                                                 list3.add(csDevCapacity);
@@ -394,6 +419,8 @@ public class MqttMessageHandler {
                                         }
                                         //其余设备
                                         else {
+                                            logDto.setOperate("系统端收到装置端"+nDid+"更新监测点台账报文，code = " + res.getCode());
+                                            logDto.setResult(1);
                                             devInfo.forEach(item->{
                                                 updateLineInfo(nDid,item);
                                             });
@@ -404,26 +431,47 @@ 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);
                                         });
                                     }
                                 }
+                                logMessageTemplate.sendMember(logDto);
                                 break;
                             case 15:
+                                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);
                                 rtFeignClient.apfRtAnalysis(appAutoDataMessage);
+                                logMessageTemplate.sendMember(logDto);
                                 break;
                             case 48:
-                                logDto.setUserName("运维管理员");
+                                logDto.setOperate("系统端收到装置端"+nDid+"询问项目列表报文，code = " + res.getCode());
-                                logDto.setOperate("监测点:" + (nDid + rspDataDto.getClDid()) + "询问项目列表");
+                                logDto.setResult(1);
                                 List<RspDataDto.ProjectInfo> projectInfoList = JSON.parseArray(JSON.toJSONString(rspDataDto.getDataArray()), RspDataDto.ProjectInfo.class);
-                                String key3 = AppRedisKey.PROJECT_INFO + nDid + rspDataDto.getClDid();
+                                CsDeviceRegistry csDeviceRegistry = csDeviceRegistryFeignClient.queryByCurrentNdidAndClDid(nDid, rspDataDto.getClDid()).getData();
+                                String lineId;
+                                if (Objects.isNull(csDeviceRegistry)) {
+                                    List<CsLinePO> lines = csLineFeignClient.findByNdid(nDid).getData();
+                                    if (CollectionUtil.isEmpty(lines)) {
+                                        throw new BusinessException("通过装置NDID获取监测点信息失败");
+                                    }
+                                    CsLinePO line = lines.stream().filter(item->Objects.equals(item.getClDid(),rspDataDto.getClDid())).findFirst().orElse(null);
+                                    if (Objects.isNull(line)) {
+                                        throw new BusinessException("通过逻辑子设备ID获取监测点信息失败");
+                                    }
+                                    lineId = line.getLineId();
+                                } else {
+                                    lineId = csDeviceRegistry.getId();
+                                }
+                                String key3 = AppRedisKey.PROJECT_INFO + lineId;
                                 redisUtil.saveByKeyWithExpire(key3,projectInfoList,60L);
+                                logMessageTemplate.sendMember(logDto);
                                 break;
                             default:
                                 break;
@@ -439,19 +487,40 @@ 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);
+            //log.info(result);
+            logDto.setOperate("装置响应");
             logDto.setResult(0);
             logDto.setFailReason(result);
-            csLogsFeignClient.addUserLog(logDto);
+            logMessageTemplate.sendMember(logDto);
             throw new BusinessException(result);
         }
     }
 
     public void updateLineInfo(String nDid,RspDataDto.LdevInfo item) {
+        CsDeviceRegistry csDeviceRegistry = csDeviceRegistryFeignClient.queryByCurrentNdidAndClDid(nDid, item.getClDid()).getData();
+        String lineId;
+        if (Objects.isNull(csDeviceRegistry)) {
+            List<CsLinePO> lines = csLineFeignClient.findByNdid(nDid).getData();
+            if (CollectionUtil.isEmpty(lines)) {
+                throw new BusinessException("通过装置NDID获取监测点信息失败");
+            }
+            CsLinePO line = lines.stream().filter(item2->Objects.equals(item2.getClDid(),item.getClDid())).findFirst().orElse(null);
+            if (Objects.isNull(line)) {
+                throw new BusinessException("通过逻辑子设备ID获取监测点信息失败");
+            }
+            lineId = line.getLineId();
+        } else {
+            lineId = csDeviceRegistry.getId();
+        }
         CsLineParam csLineParam = new CsLineParam();
-        csLineParam.setLineId(nDid.concat(item.getClDid().toString()));
+        csLineParam.setLineId(lineId);
         csLineParam.setVolGrade(item.getVolGrade());
         csLineParam.setPtRatio(item.getPtRatio());
         csLineParam.setCtRatio(item.getCtRatio());
@@ -460,8 +529,8 @@ public class MqttMessageHandler {
         csLineFeignClient.updateLine(csLineParam);
         //生成监测点限值
         Overlimit overlimit = COverlimitUtil.globalAssemble(item.getVolGrade().floatValue(),10f,10f,10f,0,0);
-        overlimit.setId(nDid.concat(item.getClDid().toString()));
+        overlimit.setId(lineId);
-        overLimitWlMapper.deleteById(nDid.concat(item.getClDid().toString()));
+        overLimitWlMapper.deleteById(lineId);
         overLimitWlMapper.insert(overlimit);
     }
 
@@ -694,44 +763,57 @@ public class MqttMessageHandler {
      * 2：模块信息
      * 3：监测点pt/ct信息
      */
-    public void askDevData(String nDid,String version,Integer type,Integer mid){
+    public void askDevData(String nDid,String version,Integer type,Integer mid) {
-        ReqAndResDto.Req reqAndResParam = new ReqAndResDto.Req();
+        //获取文件模板信息
-        reqAndResParam.setMid(mid);
+        List<CsModelDto> modelList = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.MODEL + nDid),CsModelDto.class);
-        reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
+        for (CsModelDto model : modelList) {
-        reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_6.getCode()));
+            LogMessage logDto = new LogMessage();
-        reqAndResParam.setExpire(-1);
+            logDto.setUserIndex("系统");
-        AskDataDto askDataDto = new AskDataDto();
+            logDto.setLoginName("系统");
-        askDataDto.setDataAttr(0);
+            logDto.setResult(1);
-        askDataDto.setOperate(1);
+            ReqAndResDto.Req reqAndResParam = new ReqAndResDto.Req();
-        askDataDto.setStartTime(-1);
+            reqAndResParam.setMid(mid);
-        askDataDto.setEndTime(-1);
+            reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
-        switch (type) {
+            reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_6.getCode()));
-            case 1:
+            reqAndResParam.setExpire(-1);
-                reqAndResParam.setDid(0);
+            AskDataDto askDataDto = new AskDataDto();
-                askDataDto.setCldid(0);
+            askDataDto.setDataAttr(0);
-                askDataDto.setDataType(1);
+            askDataDto.setOperate(1);
-                break;
+            askDataDto.setStartTime(-1);
-            case 2:
+            askDataDto.setEndTime(-1);
-                reqAndResParam.setDid(1);
+            switch (type) {
-                askDataDto.setCldid(-1);
+                case 1:
-                askDataDto.setDataType(2);
+                    reqAndResParam.setDid(model.getDid());
-                break;
+                    askDataDto.setCldid(0);
-            case 3:
+                    askDataDto.setDataType(1);
-                reqAndResParam.setDid(2);
+                    logDto.setOperate("系统端向装置端"+nDid+"询问软件信息");
-                askDataDto.setCldid(-1);
+                    break;
-                askDataDto.setDataType(2);
+                case 2:
-                break;
+                    reqAndResParam.setDid(model.getDid());
-            //询问工程信息
+                    askDataDto.setCldid(-1);
-            case 48:
+                    askDataDto.setDataType(2);
-                reqAndResParam.setDid(1);
+                    logDto.setOperate("系统端向装置端"+nDid+"询问逻辑设备1信息");
-                askDataDto.setCldid(1);
+                    break;
-                askDataDto.setDataType(48);
+                case 3:
-                break;
+                    reqAndResParam.setDid(model.getDid());
-            default:
+                    askDataDto.setCldid(-1);
-                break;
+                    askDataDto.setDataType(2);
+                    logDto.setOperate("系统端向装置端"+nDid+"询问逻辑设备2信息");
+                    break;
+                //询问工程信息
+                case 48:
+                    reqAndResParam.setDid(model.getDid());
+                    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);
         }
-        reqAndResParam.setMsg(askDataDto);
-        publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(reqAndResParam),1,false);
     }
 
     public String getEnum(Integer code) {

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

@@ -3,10 +3,7 @@ package com.njcn.access.listener;
 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.service.ICsEquipmentDeliveryService;
 import com.njcn.access.utils.RedisSetUtil;
-import com.njcn.access.utils.SendMessageUtil;
-import com.njcn.csdevice.api.*;
 import com.njcn.redis.utils.RedisUtil;
 import com.njcn.rt.pojo.dto.BaseRealDataSet;
 import lombok.extern.slf4j.Slf4j;
@@ -27,27 +24,12 @@ import java.util.Set;
 @Slf4j
 @Component
 public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
-
-    @Resource
-    private ICsEquipmentDeliveryService csEquipmentDeliveryService;
-    @Resource
-    private CsLogsFeignClient csLogsFeignClient;
-    @Resource
-    private CsLedgerFeignClient csLedgerFeignclient;
-    @Resource
-    private EquipmentFeignClient equipmentFeignClient;
     @Resource
     private RedisUtil redisUtil;
     @Resource
-    private SendMessageUtil sendMessageUtil;
-    @Resource
-    private CsCommunicateFeignClient csCommunicateFeignClient;
-    @Resource
     private MqttPublisher publisher;
     @Resource
     private RedisSetUtil redisSetUtil;
-    @Resource
-    private DeviceMessageFeignClient deviceMessageFeignClient;
 
     public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
         super(listenerContainer);

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

@@ -4,111 +4,253 @@ import cn.hutool.core.collection.CollUtil;
 import com.njcn.access.service.ICsEquipmentDeliveryService;
 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.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.pojo.po.SysDicTreePO;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.ApplicationRunner;
 import org.springframework.stereotype.Component;
 
-import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
-/**
- * 类的介绍：用来重新发起设备的接入，存在程序意外停止了，缓存失效导致无法更新装置的状态，所以需要在程序启动时发起设备的接入
- *
- * @author xuyang
- * @version 1.0.0
- * @createTime 2023/8/28 13:57
- */
 @Component
 @Slf4j
 @RequiredArgsConstructor
 public class AccessApplicationRunner implements ApplicationRunner {
 
+    private static final long STARTUP_DELAY = 60L;
+    private static final int BATCH_SIZE = 20;
+    private static final long BATCH_INTERVAL_MS = 500L;
+
     private final RedisUtil redisUtil;
     private final ICsEquipmentDeliveryService csEquipmentDeliveryService;
     private final ICsTopicService csTopicService;
     private final CsDeviceServiceImpl csDeviceService;
-    private final DictTreeFeignClient dictTreeFeignClient;
+    private final ChannelObjectUtil channelObjectUtil;
 
-    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+    // ✅ 注入公共资源
-    private static final long ACCESS_TIME = 60L;
+    private final ScheduledExecutorService sharedScheduler;
+    private final ExecutorService sharedWorkerPool;
+    private final Semaphore mqttSemaphore;
 
     @Override
     public void run(ApplicationArguments args) {
-        //项目启动60s后发起自动接入
+        sharedScheduler.schedule(() -> {
-        Runnable task = () -> {
+            try {
-            log.info("系统重启,所有符合条件的装置发起接入!");
+                executeStartupRecovery();
-            List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getOnlineDev();
+            } catch (Throwable t) {
-            if (CollUtil.isNotEmpty(list)) {
+                log.error("启动恢复任务异常", t);
-                ExecutorService executor = Executors.newFixedThreadPool(10);
-                // 将任务平均分配给10个子列表
-                List<List<CsEquipmentDeliveryPO>> subLists = new ArrayList<>();
-                int partitionSize = list.size() / 10;
-                for (int i = 0; i < 10; i++) {
-                    int start = i * partitionSize;
-                    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++) {
-                    int index = i;
-                    futures.add(executor.submit(new Callable<Void>() {
-                        @Override
-                        public Void call() {
-                            accessDev(subLists.get(index));
-                            return null;
-                        }
-                    }));
-                }
-                // 等待所有任务完成
-                for (Future<Void> future : futures) {
-                    try {
-                        future.get();
-                    } catch (InterruptedException | ExecutionException e) {
-                        throw new RuntimeException(e);
-                    }
-                }
-                // 关闭ExecutorService
-                executor.shutdown();
             }
-        };
+        }, STARTUP_DELAY, TimeUnit.SECONDS);
-        scheduler.schedule(task,ACCESS_TIME,TimeUnit.SECONDS);
+
-        scheduler.shutdown();
+        log.info("启动恢复任务已调度, 延迟={}s", STARTUP_DELAY);
     }
 
-    public void accessDev(List<CsEquipmentDeliveryPO> list) {
+    private void executeStartupRecovery() {
-        if (CollUtil.isNotEmpty(list)) {
+        List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getOnlineDev();
-            try {
+        if (CollUtil.isEmpty(list)) {
-                list.forEach(item->{
+            log.info("启动恢复: 无需恢复的设备");
-                    System.out.println(Thread.currentThread().getName() + ": reboot : nDid : " + item.getNdid());
+            return;
-                    //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入
+        }
-                    String code = dictTreeFeignClient.queryById(item.getDevType()).getData().getCode();
+        log.info("启动恢复: 待恢复设备数={}", list.size());
-                    if (Objects.equals(code, DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(),1)) {
+
-                        //csDeviceService.wlDevRegister(item.getNdid());
+        List<SysDicTreePO> dictTreeKey = channelObjectUtil.objectToList(
-                        log.info("请先手动注册、接入");
+                redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class);
-                    } else {
+        Map<String, SysDicTreePO> dictTreeMap = dictTreeKey.stream()
-                        String version = csTopicService.getVersion(item.getNdid());
+                .collect(Collectors.toMap(SysDicTreePO::getId, item -> item, (a, b) -> a));
-                        if (Objects.isNull(version)) {
+
-                            version = "V1";
+        List<String> ndidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList());
-                        }
+        Map<String, String> topicVersions = csTopicService.getVersion(ndidIds);
-                        csDeviceService.autoAccess(item.getNdid(),version,1);
+
-                    }
+        List<List<CsEquipmentDeliveryPO>> batches = CollUtil.split(list, BATCH_SIZE);
-                    redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(),1);
+
-                });
+        int batchCount = 0;
-            } catch (Exception e) {
+        for (List<CsEquipmentDeliveryPO> batch : batches) {
-                log.error(e.getMessage());
+            sharedWorkerPool.submit(() -> accessDevSafely(batch, dictTreeMap, topicVersions));
+            // 非最后一批时等待
+            if (++batchCount < batches.size()) {
+                try {
+                    Thread.sleep(BATCH_INTERVAL_MS);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    log.warn("批次间隔被中断，停止后续调度");
+                    return;
+                }
             }
         }
     }
 
+    private void accessDevSafely(List<CsEquipmentDeliveryPO> list,
+                                 Map<String, SysDicTreePO> dictTreeMap,
+                                 Map<String, String> topicVersions) {
+        for (CsEquipmentDeliveryPO item : list) {
+            try {
+                SysDicTreePO dicItem = dictTreeMap.get(item.getDevType());
+                if (dicItem != null
+                        && Objects.equals(dicItem.getCode(), DicDataEnum.PORTABLE.getCode())
+                        && Objects.equals(item.getStatus(), 1)) {
+                    log.debug("设备 {} 需手动注册接入，跳过", item.getNdid());
+                    continue;
+                }
+
+                String version = topicVersions.getOrDefault(item.getNdid(), "V1");
+
+                mqttSemaphore.acquire();
+                try {
+                    boolean result = csDeviceService.autoAccess(item.getNdid(), version, 1);
+                    if (result) {
+                        redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1);
+                    }
+                } finally {
+                    mqttSemaphore.release();
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.warn("启动恢复: 设备 {} 处理被中断", item.getNdid());
+                return;
+            } catch (Exception e) {
+                log.error("启动恢复: 设备 {} 接入异常: {}", item.getNdid(), e.getMessage());
+            }
+        }
+    }
 }
+
+//package com.njcn.access.runner;
+//
+//import cn.hutool.core.collection.CollUtil;
+//import com.njcn.access.service.ICsEquipmentDeliveryService;
+//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;
+//import com.njcn.system.enums.DicDataEnum;
+//import com.njcn.system.pojo.po.SysDicTreePO;
+//import lombok.RequiredArgsConstructor;
+//import lombok.extern.slf4j.Slf4j;
+//import org.springframework.boot.ApplicationArguments;
+//import org.springframework.boot.ApplicationRunner;
+//import org.springframework.stereotype.Component;
+//
+//import java.util.ArrayList;
+//import java.util.List;
+//import java.util.Map;
+//import java.util.Objects;
+//import java.util.concurrent.*;
+//import java.util.stream.Collectors;
+//
+///**
+// * 类的介绍：用来重新发起设备的接入，存在程序意外停止了，缓存失效导致无法更新装置的状态，所以需要在程序启动时发起设备的接入
+// *
+// * @author xuyang
+// * @version 1.0.0
+// * @createTime 2023/8/28 13:57
+// */
+//@Component
+//@Slf4j
+//@RequiredArgsConstructor
+//public class AccessApplicationRunner implements ApplicationRunner {
+//
+//    private final RedisUtil redisUtil;
+//    private final ICsEquipmentDeliveryService csEquipmentDeliveryService;
+//    private final ICsTopicService csTopicService;
+//    private final CsDeviceServiceImpl csDeviceService;
+//    private final ChannelObjectUtil channelObjectUtil;
+//
+//    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+//    private static final long ACCESS_TIME = 60L;
+//
+//    @Override
+//    public void run(ApplicationArguments args) {
+//        //项目启动60s后发起自动接入
+//        Runnable task = () -> {
+//            log.info("系统重启,所有符合条件的装置发起接入!");
+//            List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getOnlineDev();
+//            if (CollUtil.isNotEmpty(list)) {
+//                //获取字典数据
+//                List<SysDicTreePO> dictTreeKey = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class);
+//                Map<String, SysDicTreePO> dictTreeMap = dictTreeKey.stream().collect(Collectors.toMap(SysDicTreePO::getId, item -> item));
+//                //获取主题版本信息
+//                List<String> nDidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList());
+//                Map<String,String> topicVersions = csTopicService.getVersion(nDidIds);
+//
+//                ExecutorService executor = Executors.newFixedThreadPool(10);
+//                // 将任务平均分配给10个子列表
+//                List<List<CsEquipmentDeliveryPO>> subLists = new ArrayList<>();
+//                int partitionSize = list.size() / 10;
+//                for (int i = 0; i < 10; i++) {
+//                    int start = i * partitionSize;
+//                    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++) {
+//                    int index = i;
+//                    futures.add(executor.submit(new Callable<Void>() {
+//                        @Override
+//                        public Void call() {
+//                            accessDev(subLists.get(index), dictTreeMap, topicVersions);
+//                            return null;
+//                        }
+//                    }));
+//                }
+//                // 等待所有任务完成
+//                for (Future<Void> future : futures) {
+//                    try {
+//                        future.get();
+//                    } catch (InterruptedException | ExecutionException e) {
+//                        throw new RuntimeException(e);
+//                    }
+//                }
+//                // 关闭ExecutorService
+//                executor.shutdown();
+//            }
+//        };
+//        scheduler.schedule(task,ACCESS_TIME,TimeUnit.SECONDS);
+//        scheduler.shutdown();
+//    }
+//
+//    public void accessDev(List<CsEquipmentDeliveryPO> list, Map<String, SysDicTreePO> dictTreeMap,  Map<String,String> topicVersions) {
+//        if (CollUtil.isNotEmpty(list)) {
+//            try {
+//                list.forEach(item->{
+//                    //System.out.println(Thread.currentThread().getName() + ": reboot : nDid : " + item.getNdid());
+//                    //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入
+//                    if (Objects.equals(dictTreeMap.get(item.getDevType()).getCode(), DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(),1)) {
+//                        //csDeviceService.wlDevRegister(item.getNdid());
+//                        log.info("请先手动注册、接入");
+//                    } else {
+//                        String version = topicVersions.get(item.getNdid());
+//                        if (Objects.isNull(version)) {
+//                            version = "V1";
+//                        }
+//                        boolean result = csDeviceService.autoAccess(item.getNdid(),version,1);
+//                        if (result) {
+//                            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

@@ -4,161 +4,292 @@ import cn.hutool.core.collection.CollUtil;
 import com.njcn.access.service.ICsEquipmentDeliveryService;
 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.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.pojo.po.SysDicTreePO;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.ApplicationRunner;
 import org.springframework.stereotype.Component;
 
-import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
-/**
- * @author xy
- * 定时轮询离线设备接入
- */
 @Component
 @Slf4j
 @RequiredArgsConstructor
 public class AutoAccessTimer implements ApplicationRunner {
-    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+
     private static final long AUTO_TIME = 120L;
+    private static final int BATCH_SIZE = 20;
+    private static final long BATCH_INTERVAL_MS = 500L;
+
     private final RedisUtil redisUtil;
     private final ICsEquipmentDeliveryService csEquipmentDeliveryService;
     private final ICsTopicService csTopicService;
     private final CsDeviceServiceImpl csDeviceService;
-    private final DictTreeFeignClient dictTreeFeignClient;
+    private final ChannelObjectUtil channelObjectUtil;
+
+    // ✅ 注入公共资源
+    private final ScheduledExecutorService sharedScheduler;
+    private final ExecutorService sharedWorkerPool;
+    private final Semaphore mqttSemaphore;
 
     @Override
     public void run(ApplicationArguments args) {
-        if (scheduler.isShutdown() || scheduler.isTerminated()) {
+        sharedScheduler.scheduleWithFixedDelay(() -> {
-            scheduler = Executors.newScheduledThreadPool(1);
-        }
-        Runnable task = () -> {
             try {
                 executeScheduledTask();
+            } catch (Throwable t) {
+                log.error("定时轮询任务异常", t);
             }
-            // 捕获所有Throwable，包括Error
+        }, AUTO_TIME, AUTO_TIME, TimeUnit.SECONDS);
-            catch (Throwable t) {
-                log.error("定时任务发生严重异常，尝试恢复", t);
-                // 可以添加重启逻辑或告警
-            }
-        };
-        ScheduledFuture<?> future = scheduler.scheduleWithFixedDelay(task, AUTO_TIME, AUTO_TIME, TimeUnit.SECONDS);
-        // 添加监控，如果任务被取消则重新调度
-        monitorScheduledTask(future);
-    }
 
-    //10分钟检查一下调度任务
+        log.info("自动接入定时任务已启动, 周期={}s", AUTO_TIME);
-    private void monitorScheduledTask(ScheduledFuture<?> future) {
-        Thread monitorThread = new Thread(() -> {
-            while (!Thread.currentThread().isInterrupted()) {
-                try {
-                    //每10分钟检查一次
-                    Thread.sleep(600000);
-                    if (future.isCancelled() || future.isDone()) {
-                        log.warn("定时任务被取消或完成，重新调度...");
-                        // 重新启动任务
-                        run(null);
-                        break;
-                    }
-                } catch (InterruptedException e) {
-                    Thread.currentThread().interrupt();
-                    log.warn("监控线程被中断");
-                    break;
-                } catch (Exception e) {
-                    log.error("监控任务异常", e);
-                }
-            }
-        }, "Schedule-Monitor-Thread");
-
-        monitorThread.setDaemon(true);
-        monitorThread.start();
     }
 
     private void executeScheduledTask() {
-        log.info("轮询定时任务执行中!");
         List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getOfflineDev();
-        if (CollUtil.isNotEmpty(list)) {
+        if (CollUtil.isEmpty(list)) {
-            ExecutorService executor = Executors.newFixedThreadPool(10);
+            return;
-            try {
+        }
-                List<List<CsEquipmentDeliveryPO>> subLists = CollUtil.split(list, 10);
+        log.info("轮询定时任务执行中, 待处理设备数={}", list.size());
-                List<Future<Void>> futures = new ArrayList<>();
+
-                for (List<CsEquipmentDeliveryPO> subList : subLists) {
+        List<SysDicTreePO> dictTreeKey = channelObjectUtil.objectToList(
-                    futures.add(executor.submit(() -> {
+                redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class);
-                        try {
+        Map<String, SysDicTreePO> dictTreeMap = dictTreeKey.stream()
-                            accessDevSafely(subList); // 使用安全版本
+                .collect(Collectors.toMap(SysDicTreePO::getId, item -> item, (a, b) -> a));
-                        } catch (Exception e) {
+
-                            log.error("处理设备子列表异常", e);
+        List<String> ndidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList());
-                        }
+        Map<String, String> topicVersions = csTopicService.getVersion(ndidIds);
-                        return null;
+
-                    }));
+        List<List<CsEquipmentDeliveryPO>> batches = CollUtil.split(list, BATCH_SIZE);
-                }
+
-                for (Future<Void> future : futures) {
+        int batchCount = 0;
-                    try {
+        for (List<CsEquipmentDeliveryPO> batch : batches) {
-                        future.get(5, TimeUnit.MINUTES);
+            sharedWorkerPool.submit(() -> accessDevSafely(batch, dictTreeMap, topicVersions));
-                    } catch (TimeoutException e) {
+
-                        log.error("任务执行超时", e);
+            // 非最后一批时等待
-                    } catch (Exception e) {
+            if (++batchCount < batches.size()) {
-                        log.error("任务执行异常", e);
-                    }
-                }
-            } finally {
-                executor.shutdown();
                 try {
-                    if (!executor.awaitTermination(1, TimeUnit.MINUTES)) {
+                    Thread.sleep(BATCH_INTERVAL_MS);
-                        executor.shutdownNow();
-                    }
                 } catch (InterruptedException e) {
-                    executor.shutdownNow();
                     Thread.currentThread().interrupt();
+                    log.warn("批次间隔被中断，停止后续调度");
+                    return;
                 }
             }
         }
     }
 
-    //安全的accessDev版本
+    private void accessDevSafely(List<CsEquipmentDeliveryPO> list,
-    private void accessDevSafely(List<CsEquipmentDeliveryPO> list) {
+                                 Map<String, SysDicTreePO> dictTreeMap,
-        if (CollUtil.isNotEmpty(list)) {
+                                 Map<String, String> topicVersions) {
-            for (CsEquipmentDeliveryPO item : list) {
+        for (CsEquipmentDeliveryPO item : list) {
-                try {
-                    processSingleDevice(item);
-                } catch (Exception e) {
-                    log.error("处理设备 {} 失败: {}", item.getNdid(), e.getMessage());
-                }
-            }
-        }
-    }
-
-    private void processSingleDevice(CsEquipmentDeliveryPO item) {
-        System.out.println(Thread.currentThread().getName() + ": auto : nDid : " + item.getNdid());
-        String code = dictTreeFeignClient.queryById(item.getDevType()).getData().getCode();
-        if (Objects.equals(code, DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(), 1)) {
-            log.info("设备 {} 需要手动注册、接入", item.getNdid());
-        } else {
-            String version = csTopicService.getVersion(item.getNdid());
-            if (Objects.isNull(version)) {
-                version = "V1";
-            }
-            // 使用try-catch确保单个设备失败不影响其他设备
             try {
-                boolean success = csDeviceService.autoAccess2(item.getNdid(), version, 1);
+                SysDicTreePO dicItem = dictTreeMap.get(item.getDevType());
-                if (success) {
+                if (dicItem != null
-                    redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1);
+                        && Objects.equals(dicItem.getCode(), DicDataEnum.PORTABLE.getCode())
-                } else {
+                        && Objects.equals(item.getStatus(), 1)) {
-                    log.warn("设备 {} 接入失败", item.getNdid());
+                    log.debug("设备 {} 需手动注册接入，跳过", item.getNdid());
+                    continue;
                 }
+
+                String version = topicVersions.getOrDefault(item.getNdid(), "V1");
+
+                mqttSemaphore.acquire();
+                try {
+                    boolean success = csDeviceService.autoAccess2(item.getNdid(), version, 1);
+                    if (success) {
+                        redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1);
+                    }
+                } finally {
+                    mqttSemaphore.release();
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.warn("设备 {} 处理被中断", item.getNdid());
+                return;
             } catch (Exception e) {
                 log.error("设备 {} 接入异常: {}", item.getNdid(), e.getMessage());
             }
         }
     }
-
 }
+
+//package com.njcn.access.runner;
+//
+//import cn.hutool.core.collection.CollUtil;
+//import com.njcn.access.service.ICsEquipmentDeliveryService;
+//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.redis.pojo.enums.AppRedisKey;
+//import com.njcn.redis.utils.RedisUtil;
+//import com.njcn.system.enums.DicDataEnum;
+//import com.njcn.system.pojo.po.SysDicTreePO;
+//import lombok.RequiredArgsConstructor;
+//import lombok.extern.slf4j.Slf4j;
+//import org.springframework.boot.ApplicationArguments;
+//import org.springframework.boot.ApplicationRunner;
+//import org.springframework.stereotype.Component;
+//
+//import java.util.ArrayList;
+//import java.util.List;
+//import java.util.Map;
+//import java.util.Objects;
+//import java.util.concurrent.*;
+//        import java.util.stream.Collectors;
+//
+///**
+// * @author xy
+// * 定时轮询离线设备接入
+// */
+//@Component
+//@Slf4j
+//@RequiredArgsConstructor
+//public class AutoAccessTimer implements ApplicationRunner {
+//    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+//    private static final long AUTO_TIME = 120L;
+//    private final RedisUtil redisUtil;
+//    private final ICsEquipmentDeliveryService csEquipmentDeliveryService;
+//    private final ICsTopicService csTopicService;
+//    private final CsDeviceServiceImpl csDeviceService;
+//    private final ChannelObjectUtil channelObjectUtil;
+//
+//    @Override
+//    public void run(ApplicationArguments args) {
+//        if (scheduler.isShutdown() || scheduler.isTerminated()) {
+//            scheduler = Executors.newScheduledThreadPool(1);
+//        }
+//        Runnable task = () -> {
+//            try {
+//                executeScheduledTask();
+//            }
+//            // 捕获所有Throwable，包括Error
+//            catch (Throwable t) {
+//                log.error("定时任务发生严重异常，尝试恢复", t);
+//                // 可以添加重启逻辑或告警
+//            }
+//        };
+//        ScheduledFuture<?> future = scheduler.scheduleWithFixedDelay(task, AUTO_TIME, AUTO_TIME, TimeUnit.SECONDS);
+//        // 添加监控，如果任务被取消则重新调度
+//        monitorScheduledTask(future);
+//    }
+//
+//    //10分钟检查一下调度任务
+//    private void monitorScheduledTask(ScheduledFuture<?> future) {
+//        Thread monitorThread = new Thread(() -> {
+//            while (!Thread.currentThread().isInterrupted()) {
+//                try {
+//                    //每10分钟检查一次
+//                    Thread.sleep(600000);
+//                    if (future.isCancelled() || future.isDone()) {
+//                        log.warn("定时任务被取消或完成，重新调度...");
+//                        // 重新启动任务
+//                        run(null);
+//                        break;
+//                    }
+//                } catch (InterruptedException e) {
+//                    Thread.currentThread().interrupt();
+//                    log.warn("监控线程被中断");
+//                    break;
+//                } catch (Exception e) {
+//                    log.error("监控任务异常", e);
+//                }
+//            }
+//        }, "Schedule-Monitor-Thread");
+//
+//        monitorThread.setDaemon(true);
+//        monitorThread.start();
+//    }
+//
+//    private void executeScheduledTask() {
+//        log.info("轮询定时任务执行中!");
+//        List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getOfflineDev();
+//        if (CollUtil.isNotEmpty(list)) {
+//            //获取字典数据
+//            List<SysDicTreePO> dictTreeKey = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class);
+//            Map<String, SysDicTreePO> dictTreeMap = dictTreeKey.stream().collect(Collectors.toMap(SysDicTreePO::getId, item -> item));
+//            //获取主题版本信息
+//            List<String> nDidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList());
+//            Map<String,String> topicVersions = csTopicService.getVersion(nDidIds);
+//
+//            ExecutorService executor = Executors.newFixedThreadPool(10);
+//            try {
+//                List<List<CsEquipmentDeliveryPO>> subLists = CollUtil.split(list, 10);
+//                List<Future<Void>> futures = new ArrayList<>();
+//                for (List<CsEquipmentDeliveryPO> subList : subLists) {
+//                    futures.add(executor.submit(() -> {
+//                        try {
+//                            accessDevSafely(subList,dictTreeMap,topicVersions);
+//                        } catch (Exception e) {
+//                            log.error("处理设备子列表异常", e);
+//                        }
+//                        return null;
+//                    }));
+//                }
+//                for (Future<Void> future : futures) {
+//                    try {
+//                        future.get(5, TimeUnit.MINUTES);
+//                    } catch (TimeoutException e) {
+//                        log.error("任务执行超时", e);
+//                    } catch (Exception e) {
+//                        log.error("任务执行异常", e);
+//                    }
+//                }
+//            } finally {
+//                executor.shutdown();
+//                try {
+//                    if (!executor.awaitTermination(1, TimeUnit.MINUTES)) {
+//                        executor.shutdownNow();
+//                    }
+//                } catch (InterruptedException e) {
+//                    executor.shutdownNow();
+//                    Thread.currentThread().interrupt();
+//                }
+//            }
+//        }
+//    }
+//
+//    //安全的accessDev版本
+//    private void accessDevSafely(List<CsEquipmentDeliveryPO> list, Map<String, SysDicTreePO> dictTreeMap, Map<String,String> topicVersions) {
+//        if (CollUtil.isNotEmpty(list)) {
+//            for (CsEquipmentDeliveryPO item : list) {
+//                try {
+//                    if (Objects.equals(dictTreeMap.get(item.getDevType()).getCode(), DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(), 1)) {
+//                        log.info("设备 {} 需要手动注册、接入", item.getNdid());
+//                    } else {
+//                        String version = topicVersions.get(item.getNdid());
+//                        if (Objects.isNull(version)) {
+//                            version = "V1";
+//                        }
+//                        // 使用try-catch确保单个设备失败不影响其他设备
+//                        try {
+//                            boolean success = csDeviceService.autoAccess2(item.getNdid(), version, 1);
+//                            if (success) {
+//                                redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1);
+//                            }
+//                        } catch (Exception e) {
+//                            log.error("设备 {} 接入异常: {}", item.getNdid(), e.getMessage());
+//                        }
+//                    }
+//                } catch (Exception e) {
+//                    log.error("处理设备 {} 失败: {}", item.getNdid(), e.getMessage());
+//                }
+//            }
+//        }
+//
+//    }
+//}

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

@@ -0,0 +1,26 @@
+package com.njcn.access.runner;
+
+import com.njcn.access.service.ICsHeartService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+
+/**
+ * @author xy
+ * 定时轮询离线设备接入
+ */
+@Component
+@Slf4j
+@RequiredArgsConstructor
+public class UpdateDevStatusTimer {
+
+    private final ICsHeartService csHeartService;
+
+    @Scheduled(cron = "0 0/30 * * * ?")
+    public void timer() {
+        csHeartService.updateDevStatus();
+    }
+
+}

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

@@ -32,4 +32,6 @@ public interface ICsDataSetService extends IService<CsDataSet> {
      */
     List<CsDataSet> getDataSetData(String modelId);
 
+    List<CsDataSet> getDataSetData2(String modelId);
+
 }

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

@@ -58,4 +58,11 @@ public interface ICsDeviceService {
     String autoPortableLedger();
 
     String onlineRegister(String projectId,String nDid);
+
+    /**
+     * 设备修改mac之后，需要手动重新接入，因为已经存在台账了，只要询问主题、发起接入请求即可
+     * @param nDid
+     * @return
+     */
+    String accessByUpdateMac(String nDid);
 }

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

@@ -24,6 +24,11 @@ public interface ICsEquipmentDeliveryService extends IService<CsEquipmentDeliver
      */
     void updateStatusBynDid(String nDid,Integer status,String engineeringId, String projectId);
 
+    /**
+     * 批量修改设备通讯状态和接入状态
+     */
+    void updateStatusByList(List<String> list);
+
     /**
      * 根据网关id修改软件信息
      * @param nDid 网关id
@@ -61,7 +66,7 @@ public interface ICsEquipmentDeliveryService extends IService<CsEquipmentDeliver
     void devResetFactory(DeviceStatusParam param);
 
     /**
-     * 获取启用并且客户端在线的装置
+     * 获取启用并且未删除的装置
      */
     List<CsEquipmentDeliveryPO> getOnlineDev();
 

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

@@ -2,6 +2,8 @@ package com.njcn.access.service;
 
 import com.njcn.mq.message.HeartbeatTimeoutMessage;
 
+import java.util.List;
+
 /**
  * @author xy
  */
@@ -9,4 +11,8 @@ public interface ICsHeartService {
 
     void handleHeartbeat(HeartbeatTimeoutMessage message);
 
+    List<String> getMissHeartbeat();
+
+    void updateDevStatus();
+
 }

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

@@ -4,6 +4,7 @@ import com.baomidou.mybatisplus.extension.service.IService;
 import com.njcn.access.pojo.po.CsTopic;
 
 import java.util.List;
+import java.util.Map;
 
 /**
  * <p>
@@ -28,5 +29,7 @@ public interface ICsTopicService extends IService<CsTopic> {
      */
     String getVersion(String nDid);
 
+    Map<String,String> getVersion(List<String> list);
+
     void deleteByNDid(String nDid);
 }

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

@@ -10,7 +10,9 @@ import com.njcn.access.pojo.dto.ControlDto;
 import com.njcn.access.pojo.dto.ReqAndResDto;
 import com.njcn.access.service.AskDeviceDataService;
 import com.njcn.common.pojo.exception.BusinessException;
+import com.njcn.csdevice.api.CsLineFeignClient;
 import com.njcn.csdevice.enums.AlgorithmResponseEnum;
+import com.njcn.csdevice.pojo.po.CsLinePO;
 import com.njcn.mq.message.RealDataMessage;
 import com.njcn.mq.template.RealDataMessageTemplate;
 import com.njcn.redis.pojo.enums.AppRedisKey;
@@ -33,6 +35,7 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
     private static final Logger log = LoggerFactory.getLogger(AskDeviceDataServiceImpl.class);
     private final MqttPublisher publisher;
     private final CsTopicFeignClient csTopicFeignClient;
+    private final CsLineFeignClient csLineFeignClient;
     private final RedisUtil redisUtil;
     private final RealDataMessageTemplate realDataMessageTemplate;
     private static Integer mid = 1;
@@ -218,8 +221,8 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
     public void askCldRealData(String devId, String lineId, String nodeId, Integer idx) {
         RealDataMessage realDataMessage = new RealDataMessage();
         realDataMessage.setDevSeries(devId);
-        int lastDigit = Character.getNumericValue(lineId.charAt(lineId.length() - 1));
+        CsLinePO po = csLineFeignClient.getById(lineId).getData();
-        realDataMessage.setLine(lastDigit);
+        realDataMessage.setLine(po.getLineNo());
         realDataMessage.setRealData(true);
         realDataMessage.setSoeData(true);
         realDataMessage.setLimit(20);

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

@@ -6,6 +6,7 @@ import com.njcn.access.service.ICsDataSetService;
 import com.njcn.csdevice.pojo.po.CsDataSet;
 import org.springframework.stereotype.Service;
 
+import java.util.Collections;
 import java.util.List;
 
 /**
@@ -37,4 +38,13 @@ public class CsDataSetServiceImpl extends ServiceImpl<CsDataSetMapper, CsDataSet
                 .and(item->item.eq(CsDataSet::getDataType,"Stat").or().isNull(CsDataSet::getDataType))
                 .list();
     }
+
+    @Override
+    public List<CsDataSet> getDataSetData2(String modelId) {
+        return this.lambdaQuery()
+                .eq(CsDataSet::getPid, modelId)
+                .and(item->item.eq(CsDataSet::getDataType,"Stat").or().isNull(CsDataSet::getDataType))
+                .eq(CsDataSet::getStoreFlag,1)
+                .list();
+    }
 }

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

@@ -20,15 +20,19 @@ import com.njcn.access.pojo.po.CsLineModel;
 import com.njcn.access.service.*;
 import com.njcn.access.utils.CRC32Utils;
 import com.njcn.access.utils.JsonUtil;
-import com.njcn.common.pojo.dto.DeviceLogDTO;
 import com.njcn.common.pojo.exception.BusinessException;
-import com.njcn.csdevice.api.*;
+import com.njcn.csdevice.api.CsLineFeignClient;
+import com.njcn.csdevice.api.DevModelFeignClient;
+import com.njcn.csdevice.api.DevModelRelationFeignClient;
+import com.njcn.csdevice.api.EquipmentFeignClient;
 import com.njcn.csdevice.enums.AlgorithmResponseEnum;
 import com.njcn.csdevice.pojo.param.CsDevModelAddParm;
 import com.njcn.csdevice.pojo.po.CsDataArray;
 import com.njcn.csdevice.pojo.po.CsDataSet;
 import com.njcn.csdevice.pojo.po.CsDevModelPO;
 import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
+import com.njcn.mq.message.LogMessage;
+import com.njcn.mq.template.LogMessageTemplate;
 import com.njcn.oss.constant.OssPath;
 import com.njcn.oss.utils.FileStorageUtil;
 import com.njcn.redis.pojo.enums.AppRedisKey;
@@ -43,6 +47,7 @@ import com.njcn.system.pojo.vo.DictTreeVO;
 import com.njcn.web.utils.RequestUtil;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.web.multipart.MultipartFile;
@@ -74,7 +79,6 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
     private final ICsLineModelService csLineModelService;
     private final ICsGroupService csGroupService;
     private final ICsGroArrService csGroArrService;
-    private final CsLogsFeignClient csLogsFeignClient;
     private final EleWaveFeignClient waveFeignClient;
     private final DictTreeFeignClient dictTreeFeignClient;
     private final MqttPublisher publisher;
@@ -83,13 +87,15 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
     private final EquipmentFeignClient eequipmentFeignClient;
     private final DevModelRelationFeignClient devModelRelationFeignClient;
     private final CsLineFeignClient csLineFeignClient;
+    private final LogMessageTemplate logMessageTemplate;
+    private final StringRedisTemplate stringRedisTemplate;
 
     @Override
     @Transactional(rollbackFor = {Exception.class})
     public void addModel(MultipartFile file) {
         //日志实体
-        DeviceLogDTO logDto = new DeviceLogDTO();
+        LogMessage logDto = new LogMessage();
-        logDto.setUserName(RequestUtil.getUserNickname());
+        logDto.setUserIndex(RequestUtil.getUserNickname());
         logDto.setLoginName(RequestUtil.getUsername());
         logDto.setResult(1);
         String json = null;
@@ -128,21 +134,24 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                             .filter(item -> Objects.equals(item.getDevAccessMethod(),"CLD"))
                             .map(CsEquipmentDeliveryPO::getId)
                             .collect(Collectors.toList());
-                    devModelRelationFeignClient.updateDataByList(devList,csDevModelPo.getId());
+                    if (ObjectUtil.isNotEmpty(devList)) {
-                    Object object = redisUtil.getObjectByKey("setId:" + csDevModelPo.getId());
+                        devModelRelationFeignClient.updateDataByList(devList,csDevModelPo.getId());
-                    if (ObjectUtil.isNotNull(object)) {
+                        Object object = redisUtil.getObjectByKey("setId:" + csDevModelPo.getId());
-                        csLineFeignClient.updateDataByList(devList,csDevModelPo.getId(),object.toString());
+                        if (ObjectUtil.isNotNull(object)) {
+                            csLineFeignClient.updateDataByList(devList,csDevModelPo.getId(),object.toString());
+                        }
                     }
-
                 }
             }
             //5.清空模板缓存
-            redisUtil.deleteKeysByString(AppRedisKey.DEV_MODEL);
+            //redisUtil.deleteKeysByString(AppRedisKey.DEV_MODEL);
-            csLogsFeignClient.addUserLog(logDto);
+            //6.通知其他服务节点清理本地缓存
+            stringRedisTemplate.convertAndSend("model_cache_clear", "clear");
+            logMessageTemplate.sendMember(logDto);
         } catch (Exception e) {
             logDto.setResult(0);
             logDto.setFailReason(AccessResponseEnum.MODEL_ANALYSIS_ERROR.getMessage());
-            csLogsFeignClient.addUserLog(logDto);
+            logMessageTemplate.sendMember(logDto);
             throw new BusinessException(e.getMessage());
         }
     }
@@ -151,8 +160,8 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
     @Transactional(rollbackFor = {Exception.class})
     public void addDict(MultipartFile file) {
         //日志实体
-        DeviceLogDTO logDto = new DeviceLogDTO();
+        LogMessage logDto = new LogMessage();
-        logDto.setUserName(RequestUtil.getUserNickname());
+        logDto.setUserIndex(RequestUtil.getUserNickname());
         logDto.setLoginName(RequestUtil.getUsername());
         logDto.setResult(1);
         String json = null;
@@ -162,11 +171,11 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
             TemplateDto templateDto = gson.fromJson(json, TemplateDto.class);
             logDto.setOperate(templateDto.getDevType() + "录入通用字典");
             analysisDict(templateDto);
-            csLogsFeignClient.addUserLog(logDto);
+            logMessageTemplate.sendMember(logDto);
         } catch (Exception e) {
             logDto.setResult(0);
             logDto.setFailReason(AccessResponseEnum.DICT_ANALYSIS_ERROR.getMessage());
-            csLogsFeignClient.addUserLog(logDto);
+            logMessageTemplate.sendMember(logDto);
             throw new BusinessException(AccessResponseEnum.DICT_ANALYSIS_ERROR);
         }
     }
@@ -200,7 +209,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                     //发送数据给前端
                     String json = "{fileName:"+file.getOriginalFilename()+",allStep:"+times+",nowStep:"+i+"}";
                     publisher.send("/Web/Progress/" + id, new Gson().toJson(json), 1, false);
-                    DeviceLogDTO logDto = new DeviceLogDTO();
+                    LogMessage logDto = new LogMessage();
                     byte[] lsBytes;
                     if (length > 50*1024) {
                         lsBytes = Arrays.copyOfRange(bytes, (i - 1) * cap, i * cap);
@@ -226,25 +235,26 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                         //判断是否重发
                         sendNextStep(logDto,path,file,bytes.length,lsBytes,(i-1)*cap,version,id,i,hexString,true);
                     }
-                    csLogsFeignClient.addUserLog(logDto);
+                    logMessageTemplate.sendMember(logDto);
                 }
             } else {
                 String json = "{fileName:"+file.getOriginalFilename()+",allStep:\""+1+"\",nowStep:"+1+"}";
                 publisher.send("/Web/Progress", new Gson().toJson(json), 1, false);
                 ReqAndResDto.Req req = getPojo(1,path,file,length,bytes,0,hexString);
                 publisher.send("/Pfm/DevFileCmd/" + version + "/" + id, new Gson().toJson(req), 1, false);
-                DeviceLogDTO logDto = new DeviceLogDTO();
+                LogMessage logDto = new LogMessage();
                 logDto.setOperate(id + "系统上送文件,当前文件只有1帧");
                 logDto.setResult(1);
-                csLogsFeignClient.addUserLog(logDto);
+                logMessageTemplate.sendMember(logDto);
                 //判断是否重发
                 sendNextStep(logDto,path,file,length,bytes,0,version,id,1,hexString,false);
             }
         } catch (Exception e) {
-            DeviceLogDTO logDto = new DeviceLogDTO();
+            LogMessage logDto = new LogMessage();
             logDto.setResult(0);
-            logDto.setFailReason(AccessResponseEnum.UPLOAD_ERROR.getMessage());
+            logDto.setOperate("系统上传文件");
-            csLogsFeignClient.addUserLog(logDto);
+            logDto.setFailReason(e.getMessage());
+            logMessageTemplate.sendMember(logDto);
             throw new BusinessException(AccessResponseEnum.UPLOAD_ERROR);
         }
     }
@@ -278,7 +288,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
     /**
      * 根据装置响应来判断发送的内容
      */
-    public void sendNextStep(DeviceLogDTO logDto, String path, MultipartFile file, int length, byte[] bytes, Integer offset, String version, String id, int mid, String fileCheck, boolean result) {
+    public void sendNextStep(LogMessage logDto, String path, MultipartFile file, int length, byte[] bytes, Integer offset, String version, String id, int mid, String fileCheck, boolean result) {
         try {
             for (int i = 0; i < 30; i++) {
                 if (result) {
@@ -302,16 +312,17 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                         publisher.send("/Pfm/DevFileCmd/" + version + "/" + id, new Gson().toJson(req), 1, false);
                         logDto.setOperate(id + "系统上送文件,装置响应失败，重新发送，这是第" + (i+1) + "次");
                         logDto.setResult(1);
-                        csLogsFeignClient.addUserLog(logDto);
+                        logMessageTemplate.sendMember(logDto);
                     }
                 }
             }
         } catch (InterruptedException e) {
             assert logDto != null;
             logDto.setResult(0);
-            logDto.setFailReason(AccessResponseEnum.RELOAD_UPLOAD_ERROR.getMessage());
+            logDto.setOperate("平台重新上送文件");
-            csLogsFeignClient.addUserLog(logDto);
+            logDto.setFailReason(e.getMessage());
-            throw new RuntimeException(e);
+            logMessageTemplate.sendMember(logDto);
+            throw new BusinessException(AccessResponseEnum.RELOAD_UPLOAD_ERROR);
         }
     }
 
@@ -321,8 +332,8 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
      */
     private CsDevModelPO addCsDevModel(TemplateDto templateDto, String filePath){
         //日志实体
-        DeviceLogDTO logDto = new DeviceLogDTO();
+        LogMessage logDto = new LogMessage();
-        logDto.setUserName(RequestUtil.getUserNickname());
+        logDto.setUserIndex(RequestUtil.getUserNickname());
         logDto.setLoginName(RequestUtil.getUsername());
         logDto.setOperate("新增"+templateDto.getDevType()+"模板数据");
         logDto.setResult(1);
@@ -330,23 +341,9 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
         if (!Objects.isNull(po)){
             logDto.setResult(0);
             logDto.setFailReason(AccessResponseEnum.MODEL_REPEAT.getMessage());
-            csLogsFeignClient.addUserLog(logDto);
+            logMessageTemplate.sendMember(logDto);
             throw new BusinessException(AccessResponseEnum.MODEL_REPEAT);
         }
-//        CsDevModelPO model = new CsDevModelPO();
-//        model.setDevTypeName(templateDto.getDevType());
-//        model.setName(templateDto.getDevType());
-//        model.setVersionNo(templateDto.getVersion());
-//        model.setVersionDate(Date.valueOf(templateDto.getTime()));
-//        model.setFilePath(filePath);
-//        model.setStatus ("1");
-//        //fixme 先用数据类型来区分模板的类型
-//        if (templateDto.getDataList().contains("Apf") || templateDto.getDataList().contains("Dvr")){
-//            model.setType(0);
-//        } else {
-//            model.setType(1);
-//        }
-//        csDevModelMapper.insert(model);
         CsDevModelAddParm csDevModelAddParm = new CsDevModelAddParm();
         csDevModelAddParm.setDevTypeName(templateDto.getDevType());
         csDevModelAddParm.setName(templateDto.getDevType());
@@ -360,7 +357,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
             csDevModelAddParm.setType(1);
         }
         CsDevModelPO model = devModelFeignClient.addDevModel(csDevModelAddParm).getData();
-        csLogsFeignClient.addUserLog(logDto);
+        logMessageTemplate.sendMember(logDto);
         return model;
     }
 

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

@@ -9,15 +9,15 @@ import com.njcn.access.mapper.CsEquipmentDeliveryMapper;
 import com.njcn.access.pojo.param.DeviceStatusParam;
 import com.njcn.access.service.ICsEquipmentDeliveryService;
 import com.njcn.access.utils.MqttUtil;
-import com.njcn.common.pojo.dto.DeviceLogDTO;
-import com.njcn.csdevice.api.CsLogsFeignClient;
 import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
 import com.njcn.csdevice.pojo.vo.CsEquipmentDeliveryVO;
 import lombok.RequiredArgsConstructor;
 import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Service;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 /**
@@ -35,8 +35,6 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
 
     private final MqttUtil mqttUtil;
 
-    private final CsLogsFeignClient csLogsFeignClient;
-
     @Override
     public void updateStatusBynDid(String nDid,Integer status,String engineeringId, String projectId) {
         LambdaUpdateWrapper<CsEquipmentDeliveryPO> lambdaUpdateWrapper = new LambdaUpdateWrapper<>();
@@ -50,6 +48,16 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
         this.update(lambdaUpdateWrapper);
     }
 
+    @Override
+    public void updateStatusByList(List<String> list) {
+        LambdaUpdateWrapper<CsEquipmentDeliveryPO> lambdaUpdateWrapper = new LambdaUpdateWrapper<>();
+        lambdaUpdateWrapper
+                .set(CsEquipmentDeliveryPO::getRunStatus,AccessEnum.OFFLINE.getCode())
+                .set(CsEquipmentDeliveryPO::getStatus,AccessEnum.REGISTERED.getCode())
+                .in(CsEquipmentDeliveryPO::getNdid,list);
+        this.update(lambdaUpdateWrapper);
+    }
+
     @Override
     public void updateSoftInfoBynDid(String nDid, String id, Integer moduleNumber) {
         LambdaUpdateWrapper<CsEquipmentDeliveryPO> lambdaUpdateWrapper = new LambdaUpdateWrapper<>();
@@ -102,6 +110,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)) {
@@ -110,13 +119,6 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
                 boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
                 if (mqttClient) {
                     result.add(item);
-                } else {
-                    DeviceLogDTO logDto = new DeviceLogDTO();
-                    logDto.setUserName("运维管理员");
-                    logDto.setLoginName("njcnyw");
-                    logDto.setResult(1);
-                    logDto.setOperate(item.getNdid() + "接入失败，装置客户端不在线");
-                    csLogsFeignClient.addUserLog(logDto);
                 }
             });
         }
@@ -133,12 +135,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)) {
@@ -147,13 +149,6 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
                 boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
                 if (mqttClient) {
                     result.add(item);
-                } else {
-                    DeviceLogDTO logDto = new DeviceLogDTO();
-                    logDto.setUserName("运维管理员");
-                    logDto.setLoginName("njcnyw");
-                    logDto.setResult(1);
-                    logDto.setOperate(item.getNdid() + "接入失败，装置客户端不在线");
-                    csLogsFeignClient.addUserLog(logDto);
                 }
             });
         }

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

@@ -1,15 +1,19 @@
 package com.njcn.access.service.impl;
 
+import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.date.DatePattern;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.njcn.access.enums.AccessEnum;
 import com.njcn.access.pojo.dto.NoticeUserDto;
 import com.njcn.access.service.ICsEquipmentDeliveryService;
 import com.njcn.access.service.ICsHeartService;
 import com.njcn.access.service.IHeartbeatService;
 import com.njcn.access.utils.SendMessageUtil;
-import com.njcn.common.pojo.dto.DeviceLogDTO;
+import com.njcn.csdevice.api.CsCommunicateFeignClient;
-import com.njcn.csdevice.api.*;
+import com.njcn.csdevice.api.CsLedgerFeignClient;
+import com.njcn.csdevice.api.DeviceMessageFeignClient;
+import com.njcn.csdevice.api.EquipmentFeignClient;
 import com.njcn.csdevice.param.DeviceMessageParam;
 import com.njcn.csdevice.pojo.dto.DevDetailDTO;
 import com.njcn.csdevice.pojo.dto.PqsCommunicateDto;
@@ -17,8 +21,6 @@ import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
 import com.njcn.mq.message.HeartbeatTimeoutMessage;
 import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
-import com.njcn.user.api.AppUserFeignClient;
-import com.njcn.user.api.UserFeignClient;
 import com.njcn.user.pojo.po.User;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -26,8 +28,10 @@ import org.springframework.stereotype.Service;
 import javax.annotation.Resource;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 /**
@@ -45,20 +49,12 @@ public class CsHeartServiceImpl implements ICsHeartService {
     @Resource
     private ICsEquipmentDeliveryService csEquipmentDeliveryService;
     @Resource
-    private CsLogsFeignClient csLogsFeignClient;
-    @Resource
     private EquipmentFeignClient equipmentFeignClient;
     @Resource
     private SendMessageUtil sendMessageUtil;
     @Resource
     private CsLedgerFeignClient csLedgerFeignclient;
     @Resource
-    private AppUserFeignClient appUserFeignClient;
-    @Resource
-    private CsDeviceUserFeignClient csDeviceUserFeignClient;
-    @Resource
-    private UserFeignClient userFeignClient;
-    @Resource
     private IHeartbeatService heartbeatService;
     @Resource
     private CsCommunicateFeignClient csCommunicateFeignClient;
@@ -78,15 +74,54 @@ public class CsHeartServiceImpl implements ICsHeartService {
         handleDeviceOffline(nDid);
     }
 
+    @Override
+    public List<String> getMissHeartbeat() {
+        List<String> result = new ArrayList<>();
+        LambdaQueryWrapper<CsEquipmentDeliveryPO> queryWrapper = new LambdaQueryWrapper<>();
+        queryWrapper.eq(CsEquipmentDeliveryPO::getRunStatus, 2).eq(CsEquipmentDeliveryPO::getUsageStatus,1);
+        List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.list(queryWrapper);
+        //获取redis的数据
+        Set<String> keys = redisUtil.scanKeysByPattern("HEARTBEAT:*",100);
+        if (CollUtil.isNotEmpty(list)) {
+            list.forEach(item -> {
+                if (Objects.equals(item.getDevAccessMethod(),"CLD")) {
+                    return;
+                }
+                String nDid = item.getNdid();
+                boolean hasHeartbeat = keys.stream().anyMatch(key -> key.contains(nDid));
+                if (!hasHeartbeat) {
+                    result.add(nDid);
+                }
+            });
+        }
+        return result;
+    }
+
+    @Override
+    public void updateDevStatus() {
+        List<String> missHeartbeat = getMissHeartbeat();
+        if (CollUtil.isNotEmpty(missHeartbeat)) {
+            List<PqsCommunicateDto> list = new ArrayList<>();
+            //修改状态
+            csEquipmentDeliveryService.updateStatusByList(missHeartbeat);
+            //记录掉线
+            missHeartbeat.forEach(item->{
+                PqsCommunicateDto dto = new PqsCommunicateDto();
+                dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
+                dto.setDevId(item);
+                dto.setType(0);
+                dto.setDescription("通讯中断");
+                list.add(dto);
+            });
+            csCommunicateFeignClient.insertionList(list);
+        }
+    }
+
     private void handleDeviceOffline(String nDid) {
-        DeviceLogDTO logDto = new DeviceLogDTO();
-        logDto.setUserName("运维管理员");
-        logDto.setLoginName("njcnyw");
         //装置下线
         csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
         //装置调整为注册状态
         csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode(),null,null);
-        logDto.setOperate(nDid +"装置离线");
         sendMessage(nDid);
         //记录装置掉线时间
         PqsCommunicateDto dto = new PqsCommunicateDto();
@@ -95,7 +130,6 @@ public class CsHeartServiceImpl implements ICsHeartService {
         dto.setType(0);
         dto.setDescription("通讯中断");
         csCommunicateFeignClient.insertion(dto);
-        csLogsFeignClient.addUserLog(logDto);
         //清空缓存
         redisUtil.deleteKeysByString(AppRedisKey.LINE_POSITION+nDid);
     }
@@ -106,7 +140,6 @@ public class CsHeartServiceImpl implements ICsHeartService {
             NoticeUserDto dto = sendOffLine(nDid);
             if (CollectionUtil.isNotEmpty(dto.getPushClientId())) {
                 sendMessageUtil.sendEventToUser(dto);
-                addLogs(dto);
             }
         }
     }
@@ -135,12 +168,4 @@ public class CsHeartServiceImpl implements ICsHeartService {
         }
         return dto;
     }
-
-    private void addLogs(NoticeUserDto noticeUserDto) {
-        DeviceLogDTO dto = new DeviceLogDTO();
-        dto.setUserName("运维管理员");
-        dto.setLoginName("njcnyw");
-        dto.setOperate(noticeUserDto.getContent());
-        csLogsFeignClient.addUserLog(dto);
-    }
 }

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

@@ -8,7 +8,10 @@ import com.njcn.access.pojo.po.CsTopic;
 import com.njcn.access.service.ICsTopicService;
 import org.springframework.stereotype.Service;
 
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * <p>
@@ -42,6 +45,17 @@ public class CsTopicServiceImpl extends ServiceImpl<CsTopicMapper, CsTopic> impl
         return version;
     }
 
+    @Override
+    public Map<String, String> getVersion(List<String> list) {
+        Map<String, String> map = new HashMap<>();
+        List<CsTopic> topicList = this.lambdaQuery().in(CsTopic::getNDid,list).isNotNull(CsTopic::getVersion).list();
+        Map<String, List<CsTopic>> topicMap = topicList.stream().collect(Collectors.groupingBy(CsTopic::getNDid));
+        topicMap.forEach((key,value)->{
+            map.put(key,value.get(0).getVersion());
+        });
+        return map;
+    }
+
     @Override
     public void deleteByNDid(String nDid) {
         LambdaQueryWrapper<CsTopic> lambdaQueryWrapper = new LambdaQueryWrapper<>();

iot-access/access-boot/src/main/resources/bootstrap.yml

@@ -23,9 +23,13 @@ spring:
       discovery:
         ip: @service.server.url@
         server-addr: @nacos.url@
+        username: @nacos.username@
+        password: @nacos.password@
         namespace: @nacos.namespace@
       config:
         server-addr: @nacos.url@
+        username: @nacos.username@
+        password: @nacos.password@
         namespace: @nacos.namespace@
         file-extension: yaml
         shared-configs:
@@ -42,6 +46,7 @@ spring:
 
 #项目日志的配置
 logging:
+  #  config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&username=@nacos.username@&password=@nacos.password@&group=DEFAULT_GROUP&dataId=logback.xml
   config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
   level:
     root: info

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

@@ -8,14 +8,8 @@ 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;
+import com.njcn.csdevice.api.*;
-import com.njcn.csdevice.api.DataArrayFeignClient;
+import com.njcn.csdevice.pojo.po.*;
-import com.njcn.csdevice.api.DataSetFeignClient;
-import com.njcn.csdevice.api.EquipmentFeignClient;
-import com.njcn.csdevice.pojo.po.CsDataArray;
-import com.njcn.csdevice.pojo.po.CsDataSet;
-import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
-import com.njcn.csdevice.pojo.po.CsLinePO;
 import com.njcn.mq.message.AppAutoDataMessage;
 import com.njcn.redis.utils.RedisUtil;
 import com.njcn.rt.enums.RtResponseEnum;
@@ -53,12 +47,19 @@ public class RtServiceImpl implements IRtService {
     private final MqttPublisher publisher;
     private final RedisSetUtil redisSetUtil;
     private final EquipmentFeignClient equipmentFeignClient;
+    private final CsDeviceRegistryFeignClient csDeviceRegistryFeignClient;
 
     @Override
     public void analysis(AppAutoDataMessage appAutoDataMessage) {
         List<CsDataArray> dataArrayList;
         //监测点id
-        String lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
+        String lineId;
+        CsDeviceRegistry csDeviceRegistry = csDeviceRegistryFeignClient.queryByCurrentNdidAndClDid(appAutoDataMessage.getId(),appAutoDataMessage.getMsg().getClDid()).getData();
+        if (Objects.isNull(csDeviceRegistry)) {
+            lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
+        } else {
+            lineId = csDeviceRegistry.getId();
+        }
         redisUtil.delete("cldRtDataOverTime:"+lineId);
         //获取监测点基础信息
         CsLinePO po = csLineFeignClient.getById(lineId).getData();
@@ -96,6 +97,7 @@ 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);
+                redisUtil.saveByKeyWithExpire("devResponse:" + po1.getNdid() ,200,5L);
             }
         } else if (dataSet.getName().contains("实时数据") || dataSet.getName().contains("Ds$Pqd$Rt$01")) {
             //用户Id
@@ -112,6 +114,7 @@ public class RtServiceImpl implements IRtService {
                     long timestamp = item.getDataTimeSec();
                     baseRealDataSet.setDataTime(getTime(timestamp));
                     publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false);
+                    redisUtil.saveByKeyWithExpire("devResponse:" + lineId,200,5L);
                 });
             }
         }
@@ -135,6 +138,7 @@ public class RtServiceImpl implements IRtService {
                 }
                 harmRealDataSet.setDataTime(getTime(timestamp));
                 publisher.send("/Web/RealData/" + lineId, new Gson().toJson(harmRealDataSet), 1, false);
+                redisUtil.saveByKeyWithExpire("devResponse:" + lineId,200,5L);
             }
         }
     }
@@ -142,12 +146,17 @@ public class RtServiceImpl implements IRtService {
     @Override
     public void apfRtAnalysis(AppAutoDataMessage appAutoDataMessage) {
         List<CsDataArray> dataArrayList;
-        String lineId;
         //监测点id
-        if (appAutoDataMessage.getDid() == 1){
+        String lineId;
-            lineId = appAutoDataMessage.getId() + "0";
+        CsDeviceRegistry csDeviceRegistry = csDeviceRegistryFeignClient.queryByCurrentNdidAndClDid(appAutoDataMessage.getId(),appAutoDataMessage.getMsg().getClDid()).getData();
+        if (Objects.isNull(csDeviceRegistry)) {
+            if (appAutoDataMessage.getDid() == 1){
+                lineId = appAutoDataMessage.getId() + "0";
+            } else {
+                lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
+            }
         } else {
-            lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
+            lineId = csDeviceRegistry.getId();
         }
         //获取监测点基础信息
         CsLinePO po = csLineFeignClient.getById(lineId).getData();
@@ -165,7 +174,7 @@ public class RtServiceImpl implements IRtService {
         //根据dataArray解析数据
         AppAutoDataMessage.DataArray item = appAutoDataMessage.getMsg().getDataArray().get(0);
         Map<String,Float> map = getData(dataArrayList,item);
-        int data = Math.round(map.get("Apf_ModWorkingSts" + "M"));
+        int data = Math.round(map.get("Apf_ModWorkingSts" + "T"));
         redisUtil.saveByKeyWithExpire("ApfRtData:" + appAutoDataMessage.getMid(),data,10L);
     }
 

iot-analysis/analysis-rt/rt-boot/src/main/resources/bootstrap.yml

@@ -23,9 +23,13 @@ spring:
       discovery:
         ip: @service.server.url@
         server-addr: @nacos.url@
+        username: @nacos.username@
+        password: @nacos.password@
         namespace: @nacos.namespace@
       config:
         server-addr: @nacos.url@
+        username: @nacos.username@
+        password: @nacos.password@
         namespace: @nacos.namespace@
         file-extension: yaml
         shared-configs:
@@ -38,6 +42,7 @@ spring:
 
 #项目日志的配置
 logging:
+  #  config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&username=@nacos.username@&password=@nacos.password@&group=DEFAULT_GROUP&dataId=logback.xml
   config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
   level:
     root: info

iot-analysis/analysis-stat/stat-api/src/main/java/com/njcn/stat/dto/DataArrayLiteDto.java

@@ -0,0 +1,15 @@
+package com.njcn.stat.dto;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * @author 徐扬
+ */
+@Data
+public class DataArrayLiteDto implements Serializable {
+    private String name;
+    private String influxDbName;
+    private String phase;
+}

iot-analysis/analysis-stat/stat-api/src/main/java/com/njcn/stat/enums/StatResponseEnum.java

@@ -21,7 +21,7 @@ public enum StatResponseEnum {
     DICT_NULL("A10002","字典数据为空"),
     LINE_NULL("A10002","监测点为空"),
 
-    ARRAY_DATA_NOT_MATCH("A10003","上送数据与模板匹配失败"),
+    ARRAY_DATA_NOT_MATCH("A10003","上送数据个数与模板指标个数不一致"),
 
     ;
 

iot-analysis/analysis-stat/stat-boot/pom.xml

@@ -25,6 +25,11 @@
       <groupId>com.github.tocrhz</groupId>
       <artifactId>mqtt-spring-boot-starter</artifactId>
     </dependency>
+    <!-- caffeine 本地缓存 -->
+    <dependency>
+      <groupId>com.github.ben-manes.caffeine</groupId>
+      <artifactId>caffeine</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.njcn</groupId>
       <artifactId>common-web</artifactId>

iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/listener/ModelCacheClearListener.java

@@ -0,0 +1,40 @@
+package com.njcn.stat.listener;
+
+import com.njcn.stat.service.IStatService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.data.redis.listener.ChannelTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.Resource;
+
+/**
+ * @author 徐扬
+ */
+@Slf4j
+@Component
+public class ModelCacheClearListener implements MessageListener {
+
+    public static final String MODEL_CACHE_CLEAR_CHANNEL = "model_cache_clear";
+
+    @Resource
+    private RedisMessageListenerContainer redisMessageListenerContainer;
+
+    @Resource
+    private IStatService statService;
+
+    @PostConstruct
+    public void init() {
+        redisMessageListenerContainer.addMessageListener(this, new ChannelTopic(MODEL_CACHE_CLEAR_CHANNEL));
+    }
+
+    @Override
+    public void onMessage(Message message, byte[] pattern) {
+        log.info("收到模板缓存清理通知，channel={}", new String(message.getChannel()));
+        statService.clearModelCache();
+        log.info("本地模板缓存已清理");
+    }
+}

iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/listener/RedisKeyExpirationListener.java

@@ -1,67 +0,0 @@
-//package com.njcn.stat.listener;
-//
-//import cn.hutool.core.collection.CollectionUtil;
-//import com.njcn.common.pojo.exception.BusinessException;
-//import com.njcn.redis.pojo.enums.AppRedisKey;
-//import com.njcn.redis.utils.RedisUtil;
-//import com.njcn.stat.enums.StatResponseEnum;
-//import com.njcn.system.api.EpdFeignClient;
-//import com.njcn.system.pojo.dto.EpdDTO;
-//import lombok.extern.slf4j.Slf4j;
-//import org.apache.commons.lang3.StringUtils;
-//import org.springframework.core.annotation.Order;
-//import org.springframework.data.redis.connection.Message;
-//import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
-//import org.springframework.data.redis.listener.RedisMessageListenerContainer;
-//import org.springframework.stereotype.Component;
-//
-//import javax.annotation.Resource;
-//import java.util.HashMap;
-//import java.util.List;
-//import java.util.Map;
-//
-///**
-// * @author hongawen
-// * @version 1.0.0
-// * @date 2022年04月02日 14:31
-// */
-//@Slf4j
-//@Component
-//public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
-//
-//    @Resource
-//    private EpdFeignClient epdFeignClient;
-//
-//    @Resource
-//    private RedisUtil redisUtil;
-//
-//    public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
-//        super(listenerContainer);
-//    }
-//
-//
-//    /**
-//     * 针对redis数据失效事件，进行数据处理
-//     * 注意message.toString()可以获取失效的key
-//     */
-//    @Override
-//    @Order(0)
-//    public void onMessage(Message message, byte[] pattern) {
-//        if (StringUtils.isBlank(message.toString())) {
-//            return;
-//        }
-//        //判断失效的key
-//        String expiredKey = message.toString();
-//        if(expiredKey.equals(AppRedisKey.ELE_EPD_PQD)){
-//            Map<String,String> map = new HashMap<>();
-//            List<EpdDTO> list = epdFeignClient.findAll().getData();
-//            if (CollectionUtil.isEmpty(list)){
-//                throw new BusinessException(StatResponseEnum.DICT_NULL);
-//            }
-//            list.forEach(item->{
-//                map.put(item.getDictName(),item.getTableName());
-//            });
-//            redisUtil.saveByKeyWithExpire(AppRedisKey.ELE_EPD_PQD,map,3600L);
-//        }
-//    }
-//}

iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/IStatService.java

@@ -15,4 +15,5 @@ public interface IStatService {
 
     void electricityMeterAnalysis(AppAutoDataMessage appAutoDataMessage);
 
+    void clearModelCache();
 }

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

@@ -3,6 +3,8 @@ package com.njcn.stat.service.impl;
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.date.DatePattern;
 import com.alibaba.nacos.shaded.com.google.gson.Gson;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
 import com.njcn.access.api.CsDeviceFeignClient;
 import com.njcn.access.api.CsLineLatestDataFeignClient;
 import com.njcn.access.enums.AccessEnum;
@@ -23,6 +25,7 @@ import com.njcn.influx.utils.InfluxDbUtils;
 import com.njcn.mq.message.AppAutoDataMessage;
 import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
+import com.njcn.stat.dto.DataArrayLiteDto;
 import com.njcn.stat.enums.StatResponseEnum;
 import com.njcn.stat.service.IStatService;
 import com.njcn.system.enums.DicDataEnum;
@@ -63,6 +66,12 @@ public class StatServiceImpl implements IStatService {
     private final DeviceMessageFeignClient deviceMessageFeignClient;
     private final CsCommunicateFeignClient csCommunicateFeignClient;
 
+    // 本地缓存：模板数据变更极少，60分钟过期足够，最大缓存50000条
+    private final Cache<String, List<DataArrayLiteDto>> modelCache = Caffeine.newBuilder()
+            .expireAfterWrite(60, TimeUnit.MINUTES)
+            .softValues()
+            .build();
+
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void analysis(AppAutoDataMessage appAutoDataMessage) {
@@ -107,7 +116,7 @@ public class StatServiceImpl implements IStatService {
         }
         //云前置设备
         else if (Objects.equals(DicDataEnum.DEV_CLD.getCode(),code)) {
-            lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
+            lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appAutoDataMessage.getId())), Map.class).get(appAutoDataMessage.getMsg().getClDid().toString()).toString();
         }
 
         //获取当前设备信息
@@ -115,21 +124,22 @@ public class StatServiceImpl implements IStatService {
             Map<String,String> map = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.ELE_EPD_PQD)), Map.class);
             List<String> recordList = new ArrayList<>();
             for (AppAutoDataMessage.DataArray item : list) {
+                log.info("{}-->处理分钟数据", po.getNdid());
                 switch (item.getDataAttr()) {
                     case 1:
-                        log.info("{}-->处理最大值", po.getNdid() + appAutoDataMessage.getDid() + appAutoDataMessage.getMsg().getClDid());
+                        //log.info("{}-->处理最大值", po.getNdid() + appAutoDataMessage.getDid() + appAutoDataMessage.getMsg().getClDid());
                         dataArrayParam.setStatMethod("max");
                         break;
                     case 2:
-                        log.info("{}-->处理最小值", po.getNdid() + appAutoDataMessage.getDid() + appAutoDataMessage.getMsg().getClDid());
+                        //log.info("{}-->处理最小值", po.getNdid() + appAutoDataMessage.getDid() + appAutoDataMessage.getMsg().getClDid());
                         dataArrayParam.setStatMethod("min");
                         break;
                     case 3:
-                        log.info("{}-->处理avg", po.getNdid() + appAutoDataMessage.getDid() + appAutoDataMessage.getMsg().getClDid());
+                        //log.info("{}-->处理avg", po.getNdid() + appAutoDataMessage.getDid() + appAutoDataMessage.getMsg().getClDid());
                         dataArrayParam.setStatMethod("avg");
                         break;
                     case 4:
-                        log.info("{}-->处理cp95", po.getNdid() + appAutoDataMessage.getDid() + appAutoDataMessage.getMsg().getClDid());
+                        //log.info("{}-->处理cp95", po.getNdid() + appAutoDataMessage.getDid() + appAutoDataMessage.getMsg().getClDid());
                         dataArrayParam.setStatMethod("cp95");
                         break;
                     default:
@@ -137,15 +147,8 @@ public class StatServiceImpl implements IStatService {
                 }
                 boolean flag = Objects.equals(DicDataEnum.DEV_CLD.getCode(), code) && Objects.equals(po.getDevAccessMethod(), "CLD");
                 int clDid = flag?1:appAutoDataMessage.getMsg().getClDid();
-//                String key = AppRedisKey.DEV_MODEL.concat(dataArrayParam.getId() + dataArrayParam.getDid() + clDid + dataArrayParam.getStatMethod() + dataArrayParam.getIdx());
                 String key = AppRedisKey.DEV_MODEL.concat(dataArrayParam.getId() + dataArrayParam.getDid() + clDid + dataArrayParam.getIdx());
-                Object object = redisUtil.getObjectByKey(key);
+                List<DataArrayLiteDto> dataArrayList = getModelData(dataArrayParam, key);
-                List<CsDataArray> dataArrayList;
-                if (Objects.isNull(object)){
-                    dataArrayList = saveModelData(dataArrayParam,key);
-                } else {
-                    dataArrayList = objectToList(object);
-                }
                 List<String> result = assembleData(lineId,dataArrayList,item,appAutoDataMessage.getMsg().getClDid(),dataArrayParam.getStatMethod(),po.getProcess(),code,po.getDevAccessMethod(),map);
                 recordList.addAll(result);
                 //获取时间
@@ -176,7 +179,6 @@ public class StatServiceImpl implements IStatService {
                 csCommunicateFeignClient.insertion(dto);
             }
         }
-        System.gc();
     }
 
     @Override
@@ -217,7 +219,7 @@ public class StatServiceImpl implements IStatService {
         }
         //云前置设备
         else if (Objects.equals(DicDataEnum.DEV_CLD.getCode(),code)) {
-            lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
+            lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appAutoDataMessage.getId())), Map.class).get(appAutoDataMessage.getMsg().getClDid().toString()).toString();
         }
 
         //获取当前设备信息
@@ -229,13 +231,7 @@ public class StatServiceImpl implements IStatService {
                 boolean flag = Objects.equals(DicDataEnum.DEV_CLD.getCode(), code) && Objects.equals(po.getDevAccessMethod(), "CLD");
                 int clDid = flag?1:appAutoDataMessage.getMsg().getClDid();
                 String key = AppRedisKey.DEV_MODEL.concat(dataArrayParam.getId() + dataArrayParam.getDid() + clDid + dataArrayParam.getIdx());
-                Object object = redisUtil.getObjectByKey(key);
+                List<DataArrayLiteDto> dataArrayList = getModelData(dataArrayParam, key);
-                List<CsDataArray> dataArrayList;
-                if (Objects.isNull(object)){
-                    dataArrayList = saveModelData(dataArrayParam,key);
-                } else {
-                    dataArrayList = objectToList(object);
-                }
                 List<String> result = assembleDdData(lineId,dataArrayList,item,appAutoDataMessage.getMsg().getClDid(),code,po.getDevAccessMethod(),map);
                 recordList.addAll(result);
                 //获取时间
@@ -266,26 +262,20 @@ public class StatServiceImpl implements IStatService {
                 csCommunicateFeignClient.insertion(dto);
             }
         }
-        System.gc();
     }
 
     /**
-     * 缓存设备模板信息
+     * 清理本地模板缓存（由Redis Pub/Sub通知触发）
      */
-    public List<CsDataArray> saveModelData(DataArrayParam dataArrayParam,String key) {
+    @Override
-        List<CsDataArray> dataArrayList = dataArrayFeignClient.findListByParam(dataArrayParam).getData();
+    public void clearModelCache() {
-        if (CollectionUtil.isEmpty(dataArrayList)){
+        modelCache.invalidateAll();
-            throw new BusinessException(StatResponseEnum.DATA_ARRAY_NULL);
-        }
-        redisUtil.saveByKey(key,dataArrayList);
-        return dataArrayList;
     }
 
-
     /**
      * influxDB数据组装
      */
-    public List<String> assembleData(String lineId,List<CsDataArray> dataArrayList,AppAutoDataMessage.DataArray item,Integer clDid,String statMethod,Integer process,String devType,String accessMethod, Map<String,String> map) {
+    public List<String> assembleData(String lineId,List<DataArrayLiteDto> dataArrayList,AppAutoDataMessage.DataArray item,Integer clDid,String statMethod,Integer process,String devType,String accessMethod, Map<String,String> map) {
         List<String> records = new ArrayList<String>();
         List<Float> floats = PubUtils.byteArrayToFloatList(Base64.getDecoder().decode(item.getData()));
         if (CollectionUtil.isEmpty(floats)){
@@ -345,7 +335,7 @@ public class StatServiceImpl implements IStatService {
      * 电度influxDB数据组装
      * 电度数据15分钟入库一次
      */
-    public List<String> assembleDdData(String lineId,List<CsDataArray> dataArrayList,AppAutoDataMessage.DataArray item,Integer clDid,String devType,String accessMethod, Map<String,String> map) {
+    public List<String> assembleDdData(String lineId,List<DataArrayLiteDto> dataArrayList,AppAutoDataMessage.DataArray item,Integer clDid,String devType,String accessMethod, Map<String,String> map) {
         List<String> records = new ArrayList<String>();
         List<Float> floats = PubUtils.byteArrayToFloatList(Base64.getDecoder().decode(item.getData()));
         if (CollectionUtil.isEmpty(floats)){
@@ -386,16 +376,25 @@ public class StatServiceImpl implements IStatService {
         return records;
     }
 
-    public List<CsDataArray> objectToList(Object object) {
+    public List<DataArrayLiteDto> getModelData(DataArrayParam dataArrayParam, String key) {
-        List<CsDataArray> urlList = new ArrayList<>();
+        List<DataArrayLiteDto> result = new ArrayList<>();
-        if (object != null) {
+        List<DataArrayLiteDto> dataArrayList = modelCache.getIfPresent(key);
-            if (object instanceof ArrayList<?>) {
+        if (CollectionUtil.isEmpty(dataArrayList)) {
-                for (Object o : (List<?>) object) {
+            List<CsDataArray> data = dataArrayFeignClient.findListByParam(dataArrayParam).getData();
-                    urlList.add((CsDataArray) o);
+            if (CollectionUtil.isEmpty(data)){
-                }
+                throw new BusinessException(StatResponseEnum.DATA_ARRAY_NULL);
+            } else {
+                data.forEach(item->{
+                    DataArrayLiteDto dataArrayLiteDto = new DataArrayLiteDto();
+                    dataArrayLiteDto.setName(item.getName());
+                    dataArrayLiteDto.setInfluxDbName(item.getInfluxDbName());
+                    dataArrayLiteDto.setPhase(item.getPhase());
+                    result.add(dataArrayLiteDto);
+                });
             }
+            modelCache.put(key, result);
         }
-        return urlList;
+        return dataArrayList;
     }
 
 }

iot-analysis/analysis-stat/stat-boot/src/main/resources/bootstrap.yml

@@ -23,9 +23,13 @@ spring:
       discovery:
         ip: @service.server.url@
         server-addr: @nacos.url@
+        username: @nacos.username@
+        password: @nacos.password@
         namespace: @nacos.namespace@
       config:
         server-addr: @nacos.url@
+        username: @nacos.username@
+        password: @nacos.password@
         namespace: @nacos.namespace@
         file-extension: yaml
         shared-configs:
@@ -38,6 +42,7 @@ spring:
 
 #项目日志的配置
 logging:
+  #  config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&username=@nacos.username@&password=@nacos.password@&group=DEFAULT_GROUP&dataId=logback.xml
   config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
   level:
     root: info

iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/constant/ZlConstant.java

@@ -41,9 +41,29 @@ public interface ZlConstant {
      */
     String EVT_PARAM_TM = "Evt_Param_Tm";
 
+    /**
+     * 相别
+     */
+    String EVT_PARAM_PHASE = "Evt_Param_Phase";
+
     /**
      * 幅值
      */
     String EVT_PARAM_VVADEPTH = "Evt_Param_VVaDepth";
 
+    /**
+     * 暂降源与监测位置关系 0-未知、1-上游、2-下游
+     */
+    String SAG_SOURCE = "Evt_Param_QvvrPos";
+
+    /**
+     * 瞬态-有效值
+     */
+    String EVT_PARAM_RMS = "Evt_Param_Rms";
+
+    /**
+     * 瞬态-电压变化
+     */
+    String EVT_PARAM_UCHG = "Evt_Param_UCHG";
+
 }

iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/CsWaveAnalysisServiceImpl.java

@@ -5,10 +5,14 @@ import com.njcn.access.utils.ChannelObjectUtil;
 import com.njcn.access.utils.FileCommonUtils;
 import com.njcn.access.utils.MqttUtil;
 import com.njcn.common.pojo.exception.BusinessException;
+import com.njcn.csdevice.api.CsDeviceRegistryFeignClient;
+import com.njcn.csdevice.api.CsLineFeignClient;
 import com.njcn.csdevice.api.DeviceMessageFeignClient;
 import com.njcn.csdevice.api.EquipmentFeignClient;
 import com.njcn.csdevice.enums.AlgorithmResponseEnum;
 import com.njcn.csdevice.param.LineInfoParam;
+import com.njcn.csdevice.pojo.po.CsDeviceRegistry;
+import com.njcn.csdevice.pojo.po.CsLinePO;
 import com.njcn.mq.message.AppEventMessage;
 import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
@@ -45,6 +49,8 @@ public class CsWaveAnalysisServiceImpl implements ICsWaveAnalysisService {
     private final FileCommonUtils fileCommonUtils;
     private final DeviceMessageFeignClient deviceMessageFeignClient;
     private static Integer mid = 1;
+    private final CsDeviceRegistryFeignClient csDeviceRegistryFeignClient;
+    private final CsLineFeignClient csLineFeignClient;
 
     @Override
     public void analysis(AppEventMessage appEventMessage) {
@@ -58,6 +64,24 @@ public class CsWaveAnalysisServiceImpl implements ICsWaveAnalysisService {
             }
             //获取装置id
             String deviceId = equipmentFeignClient.findDevByNDid(appEventMessage.getId()).getData().getId();
+            //获取监测点id
+            String nDid = appEventMessage.getId();
+            Integer clDid = appEventMessage.getMsg().getClDid();
+            CsDeviceRegistry csDeviceRegistry = csDeviceRegistryFeignClient.queryByCurrentNdidAndClDid(nDid, clDid).getData();
+            String lineId;
+            if (Objects.isNull(csDeviceRegistry)) {
+                List<CsLinePO> lines = csLineFeignClient.findByNdid(nDid).getData();
+                if (CollectionUtil.isEmpty(lines)) {
+                    throw new BusinessException("通过装置NDID获取监测点信息失败");
+                }
+                CsLinePO line = lines.stream().filter(item->Objects.equals(item.getClDid(),clDid)).findFirst().orElse(null);
+                if (Objects.isNull(line)) {
+                    throw new BusinessException("通过逻辑子设备ID获取监测点信息失败");
+                }
+                lineId = line.getLineId();
+            } else {
+                lineId = csDeviceRegistry.getId();
+            }
             //获取波形文件名称
             List<AppEventMessage.DataArray> dataArrayList = appEventMessage.getMsg().getDataArray();
             if (CollectionUtil.isNotEmpty(dataArrayList)){
@@ -66,7 +90,6 @@ public class CsWaveAnalysisServiceImpl implements ICsWaveAnalysisService {
                     Object object = paramList.stream().filter(item2 -> ZlConstant.WAVE_NAME.equals(item2.getName())).findFirst().map(AppEventMessage.Param::getData).orElse(null);
                     Object object2 = paramList.stream().filter(item2 -> ZlConstant.WAVE_PARAM_RCDKEEPTIME.equals(item2.getName())).findFirst().map(AppEventMessage.Param::getData).orElse(null);
                     Object object3 = paramList.stream().filter(item2->ZlConstant.WAVE_POSITION.equals(item2.getName())).findFirst().map(AppEventMessage.Param::getData).orElse(null);
-                    String lineId = appEventMessage.getId() + appEventMessage.getMsg().getClDid();
                     String fileName = object.toString().replaceAll("\\[","").replaceAll("]","");
                     List<String> fileList = Arrays.stream(fileName.split(",")).collect(Collectors.toList());
                     //获取到录波文件，将文件信息存储起来

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

@@ -37,10 +37,8 @@ import com.njcn.system.enums.DicDataEnum;
 import com.njcn.system.pojo.dto.EpdDTO;
 import com.njcn.system.pojo.po.DictData;
 import com.njcn.zlevent.pojo.constant.ZlConstant;
-//import com.njcn.zlevent.service.AppNotificationService;
 import com.njcn.zlevent.service.ICsEventService;
 import com.njcn.zlevent.service.IEventService;
-//import com.njcn.zlevent.service.SmsNotificationService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.influxdb.InfluxDB;
@@ -51,6 +49,7 @@ import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.math.BigDecimal;
+import java.text.DecimalFormat;
 import java.text.SimpleDateFormat;
 import java.time.Instant;
 import java.time.LocalDateTime;
@@ -83,8 +82,6 @@ public class EventServiceImpl implements IEventService {
     private final WlRmpEventDetailMapper wlRmpEventDetailMapper;
     private final DictTreeFeignClient dictTreeFeignClient;
     private final DeviceMessageFeignClient deviceMessageFeignClient;
-//    private final AppNotificationService appNotificationService;
-//    private final SmsNotificationService smsNotificationService;
     private final EventAnalysisService eventAnalysisService;
     private final MsgSendFeignClient msgSendFeignClient;
 
@@ -176,12 +173,15 @@ public class EventServiceImpl implements IEventService {
                             if (Objects.equals(param.getName(),ZlConstant.EVT_PARAM_TM)){
                                 csEvent.setPersistTime(Double.parseDouble(param.getData().toString()));
                             }
-                            if (Objects.equals(param.getName(),ZlConstant.EVT_PARAM_VVADEPTH)) {
+                            if (Objects.equals(param.getName(),ZlConstant.EVT_PARAM_VVADEPTH) || Objects.equals(param.getName(),ZlConstant.EVT_PARAM_RMS)) {
                                 csEvent.setAmplitude(Double.parseDouble(param.getData().toString()));
                             }
-                            if (Objects.equals(param.getName(),"Evt_Param_Phase")) {
+                            if (Objects.equals(param.getName(),ZlConstant.EVT_PARAM_PHASE)) {
                                 csEvent.setPhase(param.getData().toString());
                             }
+                            if (Objects.equals(param.getName(),ZlConstant.SAG_SOURCE)) {
+                                csEvent.setSagSource(param.getData().toString());
+                            }
                             fields.put(param.getName(),param.getData());
                         }
                         //只有治理型号的设备有监测位置
@@ -194,7 +194,13 @@ public class EventServiceImpl implements IEventService {
                                 csEvent.setLocation("load");
                             }
                         }
-                        String dropZone = eventAnalysisService.determineDropZone(String.valueOf(csEvent.getAmplitude()),String.valueOf(csEvent.getPersistTime()));
+                        String dropZone = null;
+                        if (!Objects.isNull(csEvent.getAmplitude()) && !Objects.isNull(csEvent.getPersistTime())) {
+                            dropZone = eventAnalysisService.determineDropZone(String.valueOf(csEvent.getAmplitude()),String.valueOf(csEvent.getPersistTime()));
+                            if (Objects.equals(tag,"Evt_Sys_DipStr")) {
+                                csEvent.setSeverity(Double.valueOf(getYzd(csEvent.getPersistTime()*1000,csEvent.getAmplitude()/100)));
+                            }
+                        }
                         csEvent.setLandPoint(dropZone);
                         AppEventMessage.Param param = new AppEventMessage.Param();
                         param.setName("Evt_Param_DropZone");
@@ -252,7 +258,6 @@ public class EventServiceImpl implements IEventService {
                     msgSendParam.setAmplitude(amplitude);
                     msgSendParam.setPersistTime(persistTime);
                     msgSendParam.setDropZone(dropZone);
-//                    appNotificationService.sendAppNotification(1, item.getType(), po.getId(), item.getName(), eventTime, id, po.getNdid(),amplitude,persistTime,dropZone);
                     msgSendFeignClient.appMsgSend(msgSendParam);
                     //如果是暂降事件，则异步发送短信
                     if (Objects.equals(item.getName(), "Evt_Sys_DipStr")) {
@@ -263,7 +268,6 @@ public class EventServiceImpl implements IEventService {
                         msgSendParam2.setPersistTime(persistTime);
                         msgSendParam2.setDropZone(dropZone);
                         msgSendFeignClient.smsMsgSend(msgSendParam2);
-//                        smsNotificationService.sendSmsForDipEvent(po.getId(), eventTime,amplitude,persistTime,dropZone);
                     }
                 }
             }
@@ -272,6 +276,30 @@ public class EventServiceImpl implements IEventService {
         }
     }
 
+    /**
+     * 获取该事件的严重度
+     *
+     * @param persistTime 持续时间 ms单位
+     * @param eventValue  暂降、暂升幅值
+     */
+    public String getYzd(Double persistTime, Double eventValue) {
+        Double yzd;
+        // 格式化小数
+        DecimalFormat df = new DecimalFormat("0.000");
+        if (persistTime <= 20) {
+            yzd = 1.0 - eventValue;
+        } else if (persistTime > 20 && persistTime <= 200) {
+            yzd = 2.0 * (1 - eventValue);
+        } else if (persistTime > 200 && persistTime <= 500) {
+            yzd = 3.3 * (1 - eventValue);
+        } else if (persistTime > 500 && persistTime <= 10000) {
+            yzd = 5.0 * (1 - eventValue);
+        } else {
+            yzd = 10.0 * (1 - eventValue);
+        }
+        return df.format(yzd);
+    }
+
     public void insertEvent(CsEventPO item) {
         RmpEventDetailPO rmpEventDetailPo = new RmpEventDetailPO();
         rmpEventDetailPo.setEventId(item.getId());
@@ -284,9 +312,26 @@ public class EventServiceImpl implements IEventService {
         rmpEventDetailPo.setDealFlag(0);
         rmpEventDetailPo.setFileFlag(0);
         rmpEventDetailPo.setPhase(item.getPhase());
+        rmpEventDetailPo.setSagsource(getSagSource(item.getSagSource()));
+        rmpEventDetailPo.setSeverity(item.getSeverity());
         wlRmpEventDetailMapper.insert(rmpEventDetailPo);
     }
 
+    public String getSagSource(String tag) {
+        switch (tag) {
+            case "1":
+                tag = "Upper";
+                break;
+            case "2":
+                tag = "Lower";
+                break;
+            default:
+                tag = "Unknown";
+                break;
+        }
+        return tag;
+    }
+
     public String getEventType(String tag) {
         switch (tag) {
             case "Evt_Sys_DipStr":
@@ -392,7 +437,7 @@ public class EventServiceImpl implements IEventService {
         //前置告警
         if (Objects.equals(cldLogMessage.getLevel(),"process")) {
             //这边将前置服务器id当作设备id
-            po.setDeviceId(cldLogMessage.getNodeId());
+                po.setDeviceId(cldLogMessage.getNodeId());
             po.setClDid(Integer.valueOf(cldLogMessage.getProcessNo()));
             po.setType(4);
         }
@@ -401,9 +446,11 @@ public class EventServiceImpl implements IEventService {
             if (Objects.equals(cldLogMessage.getLevel(),"terminal")) {
                 po.setDeviceId(cldLogMessage.getBusinessId());
             } else {
-                CsLinePO line = csLineFeignClient.getById(cldLogMessage.getBusinessId()).getData();
+                if (!Objects.isNull(cldLogMessage.getBusinessId()) && !Objects.equals(cldLogMessage.getBusinessId(),"")) {
-                po.setDeviceId(line.getDeviceId());
+                    CsLinePO line = csLineFeignClient.getById(cldLogMessage.getBusinessId()).getData();
-                po.setLineId(cldLogMessage.getBusinessId());
+                    po.setDeviceId(line.getDeviceId());
+                    po.setLineId(cldLogMessage.getBusinessId());
+                }
             }
             po.setType(3);
         }

iot-analysis/analysis-zl-event/zl-event-boot/src/main/resources/bootstrap.yml

@@ -23,9 +23,13 @@ spring:
       discovery:
         ip: @service.server.url@
         server-addr: @nacos.url@
+        username: @nacos.username@
+        password: @nacos.password@
         namespace: @nacos.namespace@
       config:
         server-addr: @nacos.url@
+        username: @nacos.username@
+        password: @nacos.password@
         namespace: @nacos.namespace@
         file-extension: yaml
         shared-configs:
@@ -38,6 +42,7 @@ spring:
 
 #项目日志的配置
 logging:
+  #  config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&username=@nacos.username@&password=@nacos.password@&group=DEFAULT_GROUP&dataId=logback.xml
   config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
   level:
     root: info

iot-message/message-boot/pom.xml

@@ -20,6 +20,11 @@
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>com.njcn</groupId>
+            <artifactId>cs-system-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>com.njcn</groupId>
             <artifactId>common-web</artifactId>

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

@@ -0,0 +1,138 @@
+package com.njcn.message.consumer;
+
+import com.njcn.common.pojo.dto.DeviceLogDTO;
+import com.njcn.cssystem.api.CsLogsFeignClient;
+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.LogMessage;
+import com.njcn.redis.pojo.enums.AppRedisKey;
+import com.njcn.redis.pojo.enums.RedisKeyEnum;
+import com.njcn.redis.utils.RedisUtil;
+import com.njcn.system.api.RocketMqLogFeignClient;
+import com.njcn.system.pojo.po.RocketmqMsgErrorLog;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.beans.BeanUtils;
+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.ZL_LOG_TOPIC,
+        consumerGroup = BusinessTopic.ZL_LOG_TOPIC,
+        selectorExpression = BusinessTopic.LogTag.LOG_TAG,
+        consumeThreadNumber = 10,
+        enableMsgTrace = true
+)
+@Slf4j
+public class LogConsumer extends EnhanceConsumerMessageHandler<LogMessage> implements RocketMQListener<LogMessage> {
+
+    @Resource
+    private RedisUtil redisUtil;
+    @Resource
+    private RocketMqLogFeignClient rocketMqLogFeignClient;
+    @Resource
+    private CsLogsFeignClient csLogsFeignClient;
+
+    @Override
+    protected void handleMessage(LogMessage logMessage) {
+        DeviceLogDTO deviceLogDTO = new DeviceLogDTO();
+        BeanUtils.copyProperties(logMessage,deviceLogDTO);
+        csLogsFeignClient.addUserLog(deviceLogDTO);
+    }
+
+
+    /***
+     * 通过redis分布式锁判断当前消息所处状态
+     * 1、null 查不到该key的数据，属于第一次消费，放行
+     * 2、fail 上次消息消费时发生异常，放行
+     * 3、being processed 正在处理，打回去
+     * 4、success 最近72小时消费成功，避免重复消费，打回去
+     */
+    @Override
+    public boolean filter(LogMessage 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;
+    }
+
+    /**
+     * 消费成功，缓存到redis 5分钟，避免重复消费
+     */
+    @Override
+    protected void consumeSuccess(LogMessage message) {
+        redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, 5 * 60L);
+    }
+
+    /**
+     * 发生异常时，进行错误信息入库保存
+     * 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存
+     */
+    @Override
+    protected void saveExceptionMsgLog(LogMessage 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(LogMessage appAutoDataMessage) {
+        super.dispatchMessage(appAutoDataMessage);
+    }
+}

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

@@ -47,8 +47,6 @@ public class RealDataConsumer extends EnhanceConsumerMessageHandler<AppAutoDataM
     @Override
     protected void handleMessage(AppAutoDataMessage appAutoDataMessage) {
         log.info("分发至实时数据");
-        String lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
-        redisUtil.saveByKeyWithExpire("devResponse:" + lineId,200,5L);
         rtFeignClient.analysis(appAutoDataMessage);
     }
 

iot-message/message-boot/src/main/resources/bootstrap.yml

@@ -23,9 +23,13 @@ spring:
       discovery:
         ip: @service.server.url@
         server-addr: @nacos.url@
+        username: @nacos.username@
+        password: @nacos.password@
         namespace: @nacos.namespace@
       config:
         server-addr: @nacos.url@
+        username: @nacos.username@
+        password: @nacos.password@
         namespace: @nacos.namespace@
         file-extension: yaml
         shared-configs:
@@ -39,6 +43,7 @@ spring:
 
 #项目日志的配置
 logging:
+  #  config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&username=@nacos.username@&password=@nacos.password@&group=DEFAULT_GROUP&dataId=logback.xml
   config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
   level:
     root: info

pom.xml

@@ -32,6 +32,13 @@
 
   <properties>
 
+    <!--李凡测试环境-->
+<!--    <middle.server.url>192.168.2.134</middle.server.url>-->
+<!--    <service.server.url>192.168.2.134</service.server.url>-->
+<!--    <docker.server.url>192.168.2.134</docker.server.url>-->
+<!--    <nacos.url>${middle.server.url}:18848</nacos.url>-->
+<!--    <nacos.namespace></nacos.namespace>-->
+
     <!--103本地-->
 <!--    <middle.server.url>192.168.1.103</middle.server.url>-->
 <!--    <service.server.url>192.168.2.126</service.server.url>-->
@@ -52,6 +59,12 @@
     <nacos.url>${middle.server.url}:18848</nacos.url>
     <nacos.namespace></nacos.namespace>
 
+<!--    <middle.server.url>192.168.1.103</middle.server.url>-->
+<!--    <service.server.url>192.168.2.126</service.server.url>-->
+<!--    <docker.server.url>192.168.1.103</docker.server.url>-->
+<!--    <nacos.url>${middle.server.url}:18848</nacos.url>-->
+<!--    <nacos.namespace></nacos.namespace>-->
+
 <!--    <middle.server.url>192.168.1.22</middle.server.url>-->
 <!--    <service.server.url>192.168.1.126</service.server.url>-->
 <!--    <docker.server.url>192.168.1.22</docker.server.url>-->
@@ -149,6 +162,7 @@
     <minio.version>8.2.1</minio.version>
     <spring-cloud-huawei.version>1.7.0-Hoxton</spring-cloud-huawei.version>
     <kafka.version>2.9.6</kafka.version>
+    <caffeine.version>2.9.3</caffeine.version>
   </properties>
   <dependencyManagement>
     <dependencies>
@@ -396,6 +410,12 @@
         <artifactId>spring-kafka</artifactId>
         <version>${kafka.version}</version>
       </dependency>
+      <!-- caffeine 本地缓存 -->
+      <dependency>
+        <groupId>com.github.ben-manes.caffeine</groupId>
+        <artifactId>caffeine</artifactId>
+        <version>${caffeine.version}</version>
+      </dependency>
 
     </dependencies>
   </dependencyManagement>