微调

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

@@ -46,6 +46,10 @@ public class DataSetDto implements Serializable {
     @ApiModelProperty("Primary-一次值;Secondary-二次值")
     private String dataLevel;
 
+    @SerializedName("ConType")
+    @ApiModelProperty("接线方式 （0-星型,1-角型,2-V型）")
+    private Integer conType;
+
     @SerializedName("DataArray")
     @NotEmpty(message = "数据集合描述,不可为空")
     private List<DataArrayDto> dataArrayDtoList;

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

@@ -5,8 +5,7 @@ import com.njcn.redis.utils.RedisUtil;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
-import java.util.ArrayList;
+import java.util.*;
-import java.util.List;
 
 /**
  * @author xy
@@ -52,4 +51,22 @@ 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-api/src/main/java/com/njcn/access/utils/FileCommonUtils.java

@@ -0,0 +1,62 @@
+package com.njcn.access.utils;
+
+import com.alibaba.nacos.shaded.com.google.gson.Gson;
+import com.github.tocrhz.mqtt.publisher.MqttPublisher;
+import com.njcn.access.api.CsTopicFeignClient;
+import com.njcn.access.enums.AccessEnum;
+import com.njcn.access.enums.TypeEnum;
+import com.njcn.access.pojo.dto.ReqAndResDto;
+import com.njcn.redis.pojo.enums.AppRedisKey;
+import com.njcn.redis.utils.RedisUtil;
+import lombok.extern.slf4j.Slf4j;
+import net.sf.json.JSONObject;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * @author xy
+ */
+@Component
+@Slf4j
+public class FileCommonUtils {
+
+    @Resource
+    private CsTopicFeignClient csTopicFeignClient;
+    @Resource
+    private MqttPublisher publisher;
+    @Resource
+    private RedisUtil redisUtil;
+
+    /**
+     * 询问文件信息
+     */
+    public void askFileInfo(String nDid, Integer mid, String fileName) {
+        String version = csTopicFeignClient.find(nDid).getData();
+        ReqAndResDto.Req reqAndResParam = new ReqAndResDto.Req();
+        reqAndResParam.setMid(mid);
+        reqAndResParam.setDid(0);
+        reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
+        reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_8.getCode()));
+        reqAndResParam.setExpire(-1);
+        String json = "{Name:\""+fileName+"\"}";
+        JSONObject jsonObject = JSONObject.fromObject(json);
+        reqAndResParam.setMsg(jsonObject);
+        log.info("请求文件信息报文:" + new Gson().toJson(reqAndResParam));
+        publisher.send("/Pfm/DevFileCmd/"+version+"/"+nDid,new Gson().toJson(reqAndResParam),1,false);
+    }
+
+    /**
+     * 清空下载缓存
+     */
+    public void cleanRedisData(String nDid, String fileName) {
+        redisUtil.deleteKeysByString("downloadFilePath:"+ nDid);
+        redisUtil.delete("fileDowning:"+nDid);
+        redisUtil.delete("fileCheck" + nDid + fileName);
+        redisUtil.delete(AppRedisKey.FILE_PART_TIME.concat(fileName));
+        redisUtil.delete(AppRedisKey.FILE_PART.concat(fileName));
+        redisUtil.delete(AppRedisKey.FILE_PART_TIME.concat(fileName));
+        redisUtil.delete(AppRedisKey.FILE_DOWN_TIME.concat(fileName));
+    }
+
+}

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

@@ -267,6 +267,7 @@ public class MqttMessageHandler {
                         CsDevModelPO po = devModelFeignClient.findModel(item.getDevType(),item.getVersionNo(),item.getVersionDate()).getData();
                         if (Objects.isNull(po)){
                             log.error(AccessResponseEnum.MODEL_NO_FIND.getMessage());
+                            logDto.setOperate(nDid + "模板缺失");
                             logDto.setResult(0);
                             logDto.setFailReason(AccessResponseEnum.MODEL_NO_FIND.getMessage());
                             csLogsFeignClient.addUserLog(logDto);

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

@@ -74,6 +74,8 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
     private RedisUtil redisUtil;
     @Resource
     private SendMessageUtil sendMessageUtil;
+    @Resource
+    private CsDeviceServiceImpl csDeviceServiceImpl;
     private final Object lock = new Object();
 
     public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
@@ -106,7 +108,7 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
     //1.装置心跳断连
     //2.MQTT客户端不在线
     private void executeMainTask(ScheduledExecutorService scheduler, String nDid, String version) {
-        log.info("正在执行主任务...");
+        log.info("{}->装置离线", nDid);
         DeviceLogDTO logDto = new DeviceLogDTO();
         logDto.setUserName("运维管理员");
         logDto.setLoginName("njcnyw");
@@ -126,10 +128,22 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
                 } else {
                     //装置下线
                     csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
-                    startScheduledTask(scheduler,nDid,version);
+                    //装置调整为注册状态
-                    logDto.setOperate(nDid +"客户端离线进入定时任务");
+                    csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode());
+                    //startScheduledTask(scheduler,nDid,version);
+                    //logDto.setOperate(nDid +"客户端离线进入定时任务");
+                    logDto.setOperate(nDid +"装置离线");
+                    sendMessage(nDid);
+
+                    //记录装置掉线时间
+                    CsDeviceOnlineLogs record = onlineLogsService.findLastData(nDid);
+                    record.setOfflineTime(LocalDateTime.now());
+                    onlineLogsService.updateById(record);
+
+                    scheduler.shutdown();
                 }
             } catch (InterruptedException e) {
+                scheduler.shutdown();
                 throw new RuntimeException(e);
             }
             csLogsFeignClient.addUserLog(logDto);
@@ -138,18 +152,33 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
         else {
             //装置下线
             csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
-            logDto.setOperate(nDid +"主任务执行失败，装置下线，进入定时任务");
+            //装置调整为注册状态
+            csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode());
+            logDto.setOperate(nDid +"主任务执行失败，装置下线");
+            //logDto.setOperate(nDid +"主任务执行失败，装置下线，进入定时任务");
             csLogsFeignClient.addUserLog(logDto);
-            log.info("客户端离线进入定时任务...");
+            //log.info("客户端离线进入定时任务...");
-            startScheduledTask(scheduler,nDid,version);
+            //startScheduledTask(scheduler,nDid,version);
+            sendMessage(nDid);
+
+            //记录装置掉线时间
+            CsDeviceOnlineLogs record = onlineLogsService.findLastData(nDid);
+            record.setOfflineTime(LocalDateTime.now());
+            onlineLogsService.updateById(record);
+
+            scheduler.shutdown();
         }
     }
 
     private void startScheduledTask(ScheduledExecutorService scheduler, String nDid, String version) {
         synchronized (lock) {
-            NoticeUserDto dto = sendOffLine(nDid);
+            //判断是否推送消息
-            sendMessageUtil.sendEventToUser(dto);
+            boolean devModel = equipmentFeignClient.judgeDevModel(nDid).getData();
-            addLogs(dto);
+            if (devModel) {
+                NoticeUserDto dto = sendOffLine(nDid);
+                sendMessageUtil.sendEventToUser(dto);
+                addLogs(dto);
+            }
             String clientName = "NJCN-" + nDid.substring(nDid.length() - 6);
             ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> {
                 log.info(nDid + "执行重连定时任务...");
@@ -171,7 +200,7 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
                             logDto.setResult(0);
                             //一个小时未连接上，则推送告警消息
                             MAX_WARNING_TIMES++;
-                            if (MAX_WARNING_TIMES == 30) {
+                            if (MAX_WARNING_TIMES == 30 && devModel) {
                                 NoticeUserDto dto2 = sendConnectFail(nDid);
                                 sendMessageUtil.sendEventToUser(dto2);
                                 addLogs(dto2);
@@ -187,7 +216,7 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
                 } else {
                     //一个小时未连接上，则推送告警消息
                     MAX_WARNING_TIMES++;
-                    if (MAX_WARNING_TIMES == 30) {
+                    if (MAX_WARNING_TIMES == 30 && devModel) {
                         NoticeUserDto dto2 = sendConnectFail(nDid);
                         sendMessageUtil.sendEventToUser(dto2);
                         addLogs(dto2);
@@ -203,6 +232,16 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
         }
     }
 
+    //判断设备型号发送数据
+    private void sendMessage(String nDid) {
+        boolean devModel = equipmentFeignClient.judgeDevModel(nDid).getData();
+        if (devModel) {
+            NoticeUserDto dto = sendOffLine(nDid);
+            sendMessageUtil.sendEventToUser(dto);
+            addLogs(dto);
+        }
+    }
+
     //掉线通知
     private NoticeUserDto sendOffLine(String nDid) {
         NoticeUserDto dto = new NoticeUserDto();

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

@@ -1,22 +1,24 @@
 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.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.DicTreeEnum;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.ApplicationRunner;
 import org.springframework.stereotype.Component;
 
-import javax.annotation.Resource;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
-import java.util.concurrent.Executors;
+import java.util.concurrent.*;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
 
 /**
  * 类的介绍：用来重新发起设备的接入，存在程序意外停止了，缓存失效导致无法更新装置的状态，所以需要在程序启动时发起设备的接入
@@ -27,16 +29,15 @@ import java.util.concurrent.TimeUnit;
  */
 @Component
 @Slf4j
+@RequiredArgsConstructor
 public class AccessApplicationRunner implements ApplicationRunner {
 
-    @Resource
+    private final RedisUtil redisUtil;
-    private RedisUtil redisUtil;
+    private final ICsEquipmentDeliveryService csEquipmentDeliveryService;
-    @Resource
+    private final ICsTopicService csTopicService;
-    private ICsEquipmentDeliveryService csEquipmentDeliveryService;
+    private final CsDeviceServiceImpl csDeviceService;
-    @Resource
+    private final DictTreeFeignClient dictTreeFeignClient;
-    private ICsTopicService csTopicService;
+
-    @Resource
-    private CsDeviceServiceImpl csDeviceService;
     ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
     private static final long ACCESS_TIME = 60L;
 
@@ -44,17 +45,62 @@ public class AccessApplicationRunner implements ApplicationRunner {
     public void run(ApplicationArguments args) {
         //项目启动60s后发起自动接入
         Runnable task = () -> {
-            List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getAll();
+            log.info("系统重启,所有符合条件的装置全部接入!");
-            list.forEach(item->{
+            List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getOnlineDev();
-                String version = csTopicService.getVersion(item.getNdid());
+            if (CollUtil.isNotEmpty(list)) {
-                if (!Objects.isNull(version)){
+                ExecutorService executor = Executors.newFixedThreadPool(10);
-                    csDeviceService.devAccessAskTemplate(item.getNdid(),version,1);
+                // 将任务平均分配给10个子列表
-                    redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(),1);
+                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();
+            }
         };
-        scheduler.schedule(task, ACCESS_TIME, TimeUnit.SECONDS);
+        scheduler.schedule(task,ACCESS_TIME,TimeUnit.SECONDS);
-        // 关闭调度程序
         scheduler.shutdown();
     }
+
+    public void accessDev(List<CsEquipmentDeliveryPO> list) {
+        list.forEach(item->{
+            System.out.println(Thread.currentThread().getName() + ": reboot : nDid : " + item.getNdid());
+            String version = csTopicService.getVersion(item.getNdid());
+            if (!Objects.isNull(version)){
+                //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入
+                String code = dictTreeFeignClient.queryById(item.getDevModel()).getData().getCode();
+                if (Objects.equals(code,DicTreeEnum.PQV_520.getCode()) && Objects.equals(item.getStatus(),1)) {
+                    csDeviceService.wlDevRegister(item.getNdid());
+                } else {
+                    csDeviceService.devAccessAskTemplate(item.getNdid(),version,1);
+                }
+                redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(),1);
+            }
+        });
+    }
+
 }

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

@@ -0,0 +1,100 @@
+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.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.DicTreeEnum;
+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.Objects;
+import java.util.concurrent.*;
+
+/**
+ * @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 DictTreeFeignClient dictTreeFeignClient;
+
+    @Override
+    public void run(ApplicationArguments args) {
+        Runnable task = () -> {
+            log.info("轮询定时任务执行中!");
+            List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getOfflineDev();
+            if (CollUtil.isNotEmpty(list)) {
+                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();
+            }
+        };
+        //第一次执行的时间为120s，然后每隔120s执行一次
+        scheduler.scheduleAtFixedRate(task,AUTO_TIME,AUTO_TIME,TimeUnit.SECONDS);
+    }
+
+    public void accessDev(List<CsEquipmentDeliveryPO> list) {
+        list.forEach(item->{
+            System.out.println(Thread.currentThread().getName() + ": reboot : nDid : " + item.getNdid());
+            String version = csTopicService.getVersion(item.getNdid());
+            if (!Objects.isNull(version)){
+                //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入
+                String code = dictTreeFeignClient.queryById(item.getDevModel()).getData().getCode();
+                if (Objects.equals(code, DicTreeEnum.PQV_520.getCode()) && Objects.equals(item.getStatus(),1)) {
+                    csDeviceService.wlDevRegister(item.getNdid());
+                } else {
+                    csDeviceService.devAccessAskTemplate(item.getNdid(),version,1);
+                }
+                redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(),1);
+            }
+        });
+    }
+}

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

@@ -59,4 +59,14 @@ public interface ICsEquipmentDeliveryService extends IService<CsEquipmentDeliver
      * 恢复出厂设置
      */
     void devResetFactory(DeviceStatusParam param);
+
+    /**
+     * 获取启用并且客户端在线的装置
+     */
+    List<CsEquipmentDeliveryPO> getOnlineDev();
+
+    /**
+     * 获取离线、启用、客户端在线的装置
+     */
+    List<CsEquipmentDeliveryPO> getOfflineDev();
 }

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

@@ -84,8 +84,8 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
     public boolean downloadFile(String nDid, String name, Integer size, String fileCheck) {
         boolean result = true;
         try {
-            redisUtil.saveByKeyWithExpire("fileDowning:"+nDid,"fileDowning",120L);
+            redisUtil.saveByKeyWithExpire("fileDowning:"+nDid,"fileDowning",300L);
-            redisUtil.saveByKeyWithExpire("fileCheck"+nDid+name,fileCheck,120L);
+            redisUtil.saveByKey("fileCheck"+nDid+name,fileCheck);
             Object object = getDeviceMid(nDid);
             if (!Objects.isNull(object)) {
                 mid = (Integer) object;
@@ -100,6 +100,7 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
         } catch (Exception e) {
             redisUtil.delete("fileDowning:"+nDid);
             redisUtil.delete("fileCheck"+nDid+name);
+            redisUtil.delete("fileDownUserId"+nDid+name);
             throw new BusinessException(AlgorithmResponseEnum.FILE_DOWNLOAD_ERROR);
         }
         return result;

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

@@ -859,6 +859,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                 csDataSet.setType(0);
                 csDataSet.setClDev(0);
                 csDataSet.setDataLevel(item1.getDataLevel());
+                csDataSet.setConType(item1.getConType());
                 setList.add(csDataSet);
                 List<DataArrayDto> list = item1.getDataArrayDtoList();
                 if(CollectionUtil.isNotEmpty(list)) {
@@ -894,6 +895,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                         csDataSet.setDataList(String.join(",",item3.getDataList()));
                         csDataSet.setDataLevel(item4.getDataLevel());
                         csDataSet.setDataType(item4.getDataAttr());
+                        csDataSet.setConType(item4.getConType());
                         //fixme 先用数据类型来区分模板的类型
                         if (item3.getDataList().contains("Apf") || item3.getDataList().contains("Dvr")){
                             csDataSet.setType(1);

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

@@ -21,6 +21,7 @@ import com.njcn.common.pojo.dto.DeviceLogDTO;
 import com.njcn.common.pojo.enums.response.CommonResponseEnum;
 import com.njcn.common.pojo.exception.BusinessException;
 import com.njcn.csdevice.api.*;
+import com.njcn.csdevice.enums.AlgorithmResponseEnum;
 import com.njcn.csdevice.pojo.param.CsDevModelRelationAddParm;
 import com.njcn.csdevice.pojo.param.CsLedgerParam;
 import com.njcn.csdevice.pojo.param.CsLineParam;
@@ -538,6 +539,17 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
      */
     @Transactional(rollbackFor = Exception.class)
     public boolean devAccessAskTemplate(String nDid,String version,Integer mid) {
+        String clientName = "NJCN-" + nDid.substring(nDid.length() - 6);
+        boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
+        if (!mqttClient) {
+            DeviceLogDTO logDto = new DeviceLogDTO();
+            logDto.setUserName("运维管理员");
+            logDto.setLoginName("njcnyw");
+            logDto.setResult(1);
+            logDto.setOperate(nDid + "接入失败，装置客户端不在线");
+            csLogsFeignClient.addUserLog(logDto);
+            throw new BusinessException(AlgorithmResponseEnum.DEV_OFFLINE);
+        }
         boolean result = false;
         Map<Integer,String> modelMap = new HashMap<>();
         try {
@@ -580,8 +592,14 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
                 redisUtil.saveByKeyWithExpire("startFile:" + nDid,null,60L);
                 result = true;
             }
-        } catch (InterruptedException e) {
+        } catch (Exception e) {
-            throw new RuntimeException(e);
+            DeviceLogDTO logDto = new DeviceLogDTO();
+            logDto.setUserName("运维管理员");
+            logDto.setLoginName("njcnyw");
+            logDto.setResult(1);
+            logDto.setOperate(nDid + "装置接入失败");
+            csLogsFeignClient.addUserLog(logDto);
+            throw new BusinessException(e.getMessage());
         }
         return result;
     }

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

@@ -2,15 +2,20 @@ package com.njcn.access.service.impl;
 
 import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.njcn.access.enums.AccessEnum;
 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.ArrayList;
 import java.util.List;
 import java.util.Objects;
 
@@ -27,6 +32,10 @@ import java.util.Objects;
 @RequiredArgsConstructor
 public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliveryMapper, CsEquipmentDeliveryPO> implements ICsEquipmentDeliveryService {
 
+    private final MqttUtil mqttUtil;
+
+    private final CsLogsFeignClient csLogsFeignClient;
+
     @Override
     public void updateStatusBynDid(String nDid,Integer status) {
         LambdaUpdateWrapper<CsEquipmentDeliveryPO> lambdaUpdateWrapper = new LambdaUpdateWrapper<>();
@@ -81,4 +90,53 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
         this.update(lambdaUpdateWrapper);
     }
 
+    @Override
+    public List<CsEquipmentDeliveryPO> getOnlineDev() {
+        List<CsEquipmentDeliveryPO> result = new ArrayList<>();
+        List<CsEquipmentDeliveryPO> list = this.lambdaQuery()
+                .ne(CsEquipmentDeliveryPO::getRunStatus,AccessEnum.DEL.getCode())
+                .eq(CsEquipmentDeliveryPO::getUsageStatus,1)
+                .list();
+        list.forEach(item->{
+            String clientName = "NJCN-" + item.getNdid().substring(item.getNdid().length() - 6);
+            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);
+            }
+        });
+        return result;
+    }
+
+    @Override
+    public List<CsEquipmentDeliveryPO> getOfflineDev() {
+        List<CsEquipmentDeliveryPO> result = new ArrayList<>();
+        List<CsEquipmentDeliveryPO> list = this.lambdaQuery()
+                .eq(CsEquipmentDeliveryPO::getRunStatus,AccessEnum.OFFLINE.getCode())
+                .eq(CsEquipmentDeliveryPO::getUsageStatus,1)
+                .eq(CsEquipmentDeliveryPO::getStatus, AccessEnum.REGISTERED.getCode())
+                .list();
+        list.forEach(item->{
+            String clientName = "NJCN-" + item.getNdid().substring(item.getNdid().length() - 6);
+            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);
+            }
+        });
+        return result;
+    }
+
 }

iot-access/access-boot/src/test/java/com/njcn/BatchProcessing.java

@@ -0,0 +1,43 @@
+package com.njcn;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class BatchProcessing {
+
+    public static void main(String[] args) {
+        // 创建包含10000条数据的列表
+        List<Integer> dataList = new ArrayList<>();
+        for (int i = 1; i <= 20000; i++) {
+            dataList.add(i);
+        }
+
+        // 将数据分成10个子列表，每个子列表包含1000条数据
+        List<List<Integer>> batches = new ArrayList<>();
+        int batchSize = 1000;
+        int batchCount = dataList.size() / batchSize;
+        for (int i = 0; i < batchCount; i++) {
+            int fromIndex = i * batchSize;
+            int toIndex = fromIndex + batchSize;
+            List<Integer> batch = dataList.subList(fromIndex, toIndex);
+            batches.add(batch);
+        }
+
+        // 使用多线程并发处理每个子列表
+        ExecutorService executorService = Executors.newFixedThreadPool(10);
+        for (List<Integer> batch : batches) {
+            executorService.submit(() -> processBatch(batch));
+        }
+        executorService.shutdown();
+    }
+
+    private static void processBatch(List<Integer> batch) {
+        for (Integer data : batch) {
+            // 处理数据的逻辑
+            System.out.println(Thread.currentThread().getName() + ": processing data " + data);
+        }
+    }
+
+}

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

@@ -11,6 +11,9 @@ import java.io.Serializable;
 @Data
 public class BaseRealDataSet implements Serializable {
 
+    @ApiModelProperty("用户ID")
+    private String userId;
+
     @ApiModelProperty("监测点id")
     private String lineId;
 

iot-analysis/analysis-rt/rt-api/src/main/java/com/njcn/rt/pojo/dto/HarmRealDataSet.java

@@ -12,6 +12,9 @@ import java.io.Serializable;
 @Data
 public class HarmRealDataSet implements Serializable {
 
+    @ApiModelProperty("用户ID")
+    private String userId;
+
     @ApiModelProperty("监测点id")
     private String lineId;
 

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

@@ -52,6 +52,8 @@ public class RtServiceImpl implements IRtService {
         List<CsDataArray> dataArrayList;
         //监测点id
         String lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
+        //用户Id
+        String userId = redisUtil.getObjectByKey("rtDataUserId:"+lineId).toString();
         //获取监测点基础信息
         CsLinePO po = csLineFeignClient.getById(lineId).getData();
         //获取数据集 dataSet
@@ -72,6 +74,7 @@ public class RtServiceImpl implements IRtService {
         //基础数据
         if (dataSet.getName().contains("Ds$Pqd$Rt$Basic$")) {
             BaseRealDataSet baseRealDataSet = assembleData(dataArrayList,item,po.getConType());
+            baseRealDataSet.setUserId(userId);
             baseRealDataSet.setLineId(lineId);
             baseRealDataSet.setPt(po.getPtRatio().floatValue());
             baseRealDataSet.setCt(po.getCtRatio().floatValue());
@@ -83,6 +86,7 @@ public class RtServiceImpl implements IRtService {
         //fixme 目前实时数据只有基础数据和谐波数据，后期拓展，这边需要再判断
         else {
             HarmRealDataSet harmRealDataSet = harmData(dataArrayList,item);
+            harmRealDataSet.setUserId(userId);
             harmRealDataSet.setLineId(lineId);
             harmRealDataSet.setPt(po.getPtRatio().floatValue());
             harmRealDataSet.setCt(po.getCtRatio().floatValue());

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

@@ -65,6 +65,7 @@ public class StatServiceImpl implements IStatService {
         dataArrayParam.setId(appAutoDataMessage.getId());
         dataArrayParam.setDid(appAutoDataMessage.getDid());
         dataArrayParam.setCldId(appAutoDataMessage.getMsg().getClDid());
+        dataArrayParam.setIdx(appAutoDataMessage.getMsg().getDsNameIdx());
         List<AppAutoDataMessage.DataArray> list = appAutoDataMessage.getMsg().getDataArray();
         //获取监测点id
         String lineId = null;
@@ -113,7 +114,7 @@ public class StatServiceImpl implements IStatService {
                     default:
                         break;
                 }
-                String key = AppRedisKey.DEV_MODEL.concat(dataArrayParam.getId() + dataArrayParam.getDid() + dataArrayParam.getCldId() + dataArrayParam.getStatMethod());
+                String key = AppRedisKey.DEV_MODEL.concat(dataArrayParam.getId() + dataArrayParam.getDid() + dataArrayParam.getCldId() + dataArrayParam.getStatMethod() + dataArrayParam.getIdx());
                 Object object = redisUtil.getObjectByKey(key);
                 List<CsDataArray> dataArrayList;
                 if (Objects.isNull(object)){

iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/api/FileFeignClient.java

@@ -6,7 +6,7 @@ import com.njcn.mq.message.AppFileMessage;
 import com.njcn.zlevent.api.fallback.FileClientFallbackFactory;
 import org.springframework.cloud.openfeign.FeignClient;
 import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestParam;
 
 /**
  * @author xy
@@ -19,4 +19,7 @@ public interface FileFeignClient {
 
     @PostMapping("/fileStream")
     HttpResult<String> fileStream(AppFileMessage appFileMessage);
+
+    @PostMapping("/downloadMakeUpFile")
+    HttpResult<String> downloadMakeUpFile(@RequestParam("nDid") String nDid);
 }

iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/api/fallback/FileClientFallbackFactory.java

@@ -36,6 +36,12 @@ public class FileClientFallbackFactory implements FallbackFactory<FileFeignClien
                 log.error("{}异常，降级处理，异常为：{}","解析文件流",cause.toString());
                 throw new BusinessException(finalExceptionEnum);
             }
+
+            @Override
+            public HttpResult<String> downloadMakeUpFile(String nDid) {
+                log.error("{}异常，降级处理，异常为：{}","下载补召文件",cause.toString());
+                throw new BusinessException(finalExceptionEnum);
+            }
         };
     }
 }

iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/controller/FileController.java

@@ -13,10 +13,7 @@ import io.swagger.annotations.ApiImplicitParam;
 import io.swagger.annotations.ApiOperation;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.*;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
 
 /**
  * 类的介绍：
@@ -54,4 +51,14 @@ public class FileController extends BaseController {
         return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
     }
 
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/downloadMakeUpFile")
+    @ApiOperation("下载补召文件")
+    @ApiImplicitParam(name = "nDid", value = "nDid", required = true)
+    public HttpResult<String> downloadMakeUpFile(@RequestParam("nDid") String nDid){
+        String methodDescribe = getMethodDescribe("downloadMakeUpFile");
+        fileService.downloadMakeUpFile(nDid);
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
+    }
+
 }

iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/init/InitEventFiles.java

@@ -1,26 +0,0 @@
-//package com.njcn.zlevent.init;
-//
-//import com.njcn.redis.utils.RedisUtil;
-//import lombok.AllArgsConstructor;
-//import lombok.extern.slf4j.Slf4j;
-//import org.springframework.boot.CommandLineRunner;
-//import org.springframework.stereotype.Component;
-//
-//
-///**
-// * @author xy
-// *
-// * 程序重启设置任务，消费历史录波文件
-// */
-//@Slf4j
-//@Component
-//@AllArgsConstructor
-//public class InitEventFiles implements CommandLineRunner {
-//
-//    private final RedisUtil redisUtil;
-//
-//    @Override
-//    public void run(String... args) {
-//        redisUtil.saveByKeyWithExpire("startFile",null,120L);
-//    }
-//}

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

@@ -26,4 +26,8 @@ public interface IFileService {
      */
     void analysisFileStream(AppFileMessage appFileMessage);
 
+    /**
+     * 下载补召文件
+     */
+    void downloadMakeUpFile(String nDid);
 }

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

@@ -1,13 +1,10 @@
 package com.njcn.zlevent.service.impl;
 
 import cn.hutool.core.collection.CollectionUtil;
-import com.alibaba.nacos.shaded.com.google.gson.Gson;
 import com.github.tocrhz.mqtt.publisher.MqttPublisher;
 import com.njcn.access.api.CsTopicFeignClient;
-import com.njcn.access.enums.AccessEnum;
-import com.njcn.access.enums.TypeEnum;
-import com.njcn.access.pojo.dto.ReqAndResDto;
 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.CsLineFeignClient;
@@ -26,7 +23,6 @@ import com.njcn.zlevent.pojo.dto.WaveTimeDto;
 import com.njcn.zlevent.service.ICsWaveAnalysisService;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import net.sf.json.JSONObject;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 
@@ -54,6 +50,7 @@ public class CsWaveAnalysisServiceImpl implements ICsWaveAnalysisService {
     private final DicDataFeignClient dicDataFeignClient;
     private final ChannelObjectUtil channelObjectUtil;
     private final MqttUtil mqttUtil;
+    private final FileCommonUtils fileCommonUtils;
     private static Integer mid = 1;
 
     @Override
@@ -102,7 +99,6 @@ public class CsWaveAnalysisServiceImpl implements ICsWaveAnalysisService {
     }
 
     @Override
-    @Async("asyncExecutor")
     public void channelWave(String nDid) {
         //判断客户端是否在线，在线再处理文件
         String clientName = "NJCN-" + nDid.substring(nDid.length() - 6);
@@ -119,7 +115,7 @@ public class CsWaveAnalysisServiceImpl implements ICsWaveAnalysisService {
                     }
                     WaveTimeDto dto = list.get(0);
                     redisUtil.saveByKeyWithExpire("fileMid:" + nDid,mid + "concat" +dto.getFileName(),60L);
-                    askFileInfo(nDid,mid,dto.getFileName());
+                    fileCommonUtils.askFileInfo(nDid,mid,dto.getFileName());
                     mid = mid + 1;
                     if (mid > 10000) {
                         mid = 1;
@@ -134,25 +130,6 @@ public class CsWaveAnalysisServiceImpl implements ICsWaveAnalysisService {
         }
     }
 
-    /**
-     * 询问文件信息
-     */
-    public void askFileInfo(String nDid, Integer mid, String fileName) {
-        String version = csTopicFeignClient.find(nDid).getData();
-        ReqAndResDto.Req reqAndResParam = new ReqAndResDto.Req();
-        reqAndResParam.setMid(mid);
-        reqAndResParam.setDid(0);
-        reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
-        reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_8.getCode()));
-        reqAndResParam.setExpire(-1);
-        String json = "{Name:\""+fileName+"\"}";
-        JSONObject jsonObject = JSONObject.fromObject(json);
-        reqAndResParam.setMsg(jsonObject);
-        log.info("请求文件信息报文:" + new Gson().toJson(reqAndResParam));
-        publisher.send("/Pfm/DevFileCmd/"+version+"/"+nDid,new Gson().toJson(reqAndResParam),1,false);
-    }
-
-
     /**
      * 时间处理
      */

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

@@ -88,7 +88,10 @@ public class EventServiceImpl implements IEventService {
         }
         //获取装置id
         CsEquipmentDeliveryPO po = equipmentFeignClient.findDevByNDid(appEventMessage.getId()).getData();
+        //获取设备类型 true:治理设备 false:其他类型的设备
+        boolean devModel = equipmentFeignClient.judgeDevModel(appEventMessage.getId()).getData();
         try {
+            lineId = appEventMessage.getId() + appEventMessage.getMsg().getClDid();
             //处理事件数据
             List<AppEventMessage.DataArray> dataArray = appEventMessage.getMsg().getDataArray();
             for (AppEventMessage.DataArray item : dataArray) {
@@ -110,49 +113,41 @@ public class EventServiceImpl implements IEventService {
                     csEvent.setType(1);
                 }  else if (Objects.equals(item.getType(),"1")){
                     csEvent.setType(2);
-                    lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get("0").toString();
-                    csEvent.setClDid(appEventMessage.getMsg().getClDid());
                 }
                 csEvent.setLevel(Integer.parseInt(item.getType()));
+                csEvent.setClDid(appEventMessage.getMsg().getClDid());
+                csEvent.setLineId(lineId);
                 //参数入库
                 Map<String,String> map = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.ELE_EPD_PQD)), Map.class);
-                if (!Objects.isNull(item.getParam())){
+                //判断是否有参数
+                List<AppEventMessage.Param> params = item.getParam();
+                if (CollectionUtil.isNotEmpty(params)) {
                     String tableName = map.get(item.getName());
-                    List<AppEventMessage.Param> params = item.getParam();
+                    Map<String, String> tags = new HashMap<>();
+                    tags.put(InfluxDBTableConstant.UUID,id);
+                    Map<String,Object> fields = new HashMap<>();
                     for (AppEventMessage.Param param : params) {
-                        Map<String, String> tags = new HashMap<>();
+                        if (Objects.equals(param.getName(),ZlConstant.EVT_PARAM_TM)){
-                        tags.put(InfluxDBTableConstant.UUID,id);
+                            csEvent.setPersistTime(Double.parseDouble(param.getData().toString()));
-                        Map<String,Object> fields = new HashMap<>();
-                        //只有治理型号的设备有监测位置
-                        if (Objects.equals(ZlConstant.EVENT_POSITION,param.getName())){
-                            if (Objects.equals(param.getData(),ZlConstant.GRID)){
-                                fields.put(param.getName(),"电网侧");
-                                lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get("1").toString();
-                                csEvent.setClDid(1);
-                            } else if (Objects.equals(param.getData(),ZlConstant.LOAD)){
-                                fields.put(param.getName(),"负载侧");
-                                lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get("2").toString();
-                                csEvent.setClDid(2);
-                            }
-                            csEvent.setLocation(param.getData().toString());
-                        } else {
-                            if (Objects.equals(param.getName(),ZlConstant.EVT_PARAM_TM)){
-                                csEvent.setPersistTime(Double.parseDouble(param.getData().toString()));
-                            }
-                            lineId = appEventMessage.getId() + appEventMessage.getMsg().getClDid();
-                            fields.put(param.getName(),null);
-                            csEvent.setLocation(null);
-                            csEvent.setClDid(appEventMessage.getMsg().getClDid());
-                            fields.put(param.getName(),param.getData());
                         }
-                        //fixme 这边前置传递的应该是UTC时间,但是前置说是传递的北京时间,讨论了一下没太理解。这边暂时先这样处理,influx入库处理成北京时间,减去8小时。
+                        fields.put(param.getName(),param.getData());
-                        Point point = influxDbUtils.pointBuilder(tableName, item.getDataTimeSec()-8*3600, TimeUnit.SECONDS, tags, fields);
-                        BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();
-                        batchPoints.point(point);
-                        records.add(batchPoints.lineProtocol());
                     }
+                    //只有治理型号的设备有监测位置
+                    if (devModel) {
+                        if (appEventMessage.getMsg().getClDid() == 1) {
+                            fields.put("Evt_Param_Position","电网侧");
+                            csEvent.setLocation("grid");
+                        } else if (appEventMessage.getMsg().getClDid() == 2) {
+                            fields.put("Evt_Param_Position","负载侧");
+                            csEvent.setLocation("load");
+                        }
+                    }
+                    //fixme 这边前置传递的应该是UTC时间,但是前置说是传递的北京时间,讨论了一下没太理解。这边暂时先这样处理,influx入库处理成北京时间,减去8小时。
+                    Point point = influxDbUtils.pointBuilder(tableName, item.getDataTimeSec()-8*3600, TimeUnit.SECONDS, tags, fields);
+                    BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();
+                    batchPoints.point(point);
+                    records.add(batchPoints.lineProtocol());
                 }
-                csEvent.setLineId(lineId);
                 list1.add(csEvent);
                 //事件处理日志库
                 CsEventLogs csEventLogs = new CsEventLogs();

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

@@ -11,18 +11,27 @@ import com.njcn.access.enums.AccessEnum;
 import com.njcn.access.enums.AccessResponseEnum;
 import com.njcn.access.enums.TypeEnum;
 import com.njcn.access.pojo.dto.ReqAndResDto;
+import com.njcn.access.pojo.dto.file.FileDto;
 import com.njcn.access.utils.CRC32Utils;
 import com.njcn.access.utils.ChannelObjectUtil;
+import com.njcn.access.utils.FileCommonUtils;
+import com.njcn.access.utils.MqttUtil;
 import com.njcn.common.config.GeneralInfo;
 import com.njcn.common.pojo.exception.BusinessException;
+import com.njcn.csdevice.api.DeviceFtpFeignClient;
 import com.njcn.csdevice.api.EquipmentFeignClient;
+import com.njcn.csdevice.api.PortableOffLogFeignClient;
+import com.njcn.csdevice.enums.AlgorithmResponseEnum;
 import com.njcn.csharmonic.api.WavePicFeignClient;
+import com.njcn.csharmonic.enums.CsHarmonicResponseEnum;
+import com.njcn.csharmonic.pojo.dto.DownloadMakeUpDto;
 import com.njcn.mq.message.AppFileMessage;
 import com.njcn.oss.constant.GeneralConstant;
 import com.njcn.oss.constant.OssPath;
 import com.njcn.oss.utils.FileStorageUtil;
 import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
+import com.njcn.user.api.UserFeignClient;
 import com.njcn.zlevent.param.CsEventParam;
 import com.njcn.zlevent.pojo.dto.FileInfoDto;
 import com.njcn.zlevent.pojo.dto.FileStreamDto;
@@ -40,6 +49,7 @@ import net.sf.json.JSONObject;
 import org.springframework.stereotype.Service;
 
 import java.io.*;
+import java.nio.file.Files;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
@@ -69,6 +79,10 @@ public class FileServiceImpl implements IFileService {
     private final RemoveInfoUtils removeInfoUtils;
     private static Integer mid = 1;
     private final EquipmentFeignClient equipmentFeignClient;
+    private final MqttUtil mqttUtil;
+    private final FileCommonUtils fileCommonUtils;
+    private final DeviceFtpFeignClient deviceFtpFeignClient;
+    private final PortableOffLogFeignClient portableOffLogFeignClient;
 
     @Override
     public void analysisFileInfo(AppFileMessage appFileMessage) {
@@ -163,8 +177,6 @@ public class FileServiceImpl implements IFileService {
                     csEventLogs.setNowStep(1);
                     csEventLogs.setAllStep(1);
                     csEventLogs.setIsAll(1);
-                    redisUtil.delete(AppRedisKey.FILE_PART_TIME.concat(appFileMessage.getMsg().getName()));
-                    redisUtil.delete(AppRedisKey.RMQ_FILE_CONSUME_KEY.concat(fileName));
                     //存储文件信息
                     fileStreamDto.setTotal(appFileMessage.getMsg().getFrameTotal());
                     fileStreamDto.setNDid(appFileMessage.getId());
@@ -172,6 +184,16 @@ public class FileServiceImpl implements IFileService {
                     list.add(appFileMessage.getMsg().getFrameCurr());
                     fileStreamDto.setList(list);
                     redisUtil.saveByKey(AppRedisKey.FILE_PART.concat(appFileMessage.getMsg().getName()), fileStreamDto);
+                    log.info("当前文件1帧,全部收到,解析成功!");
+                    //针对补召文件
+                    String key = AppRedisKey.MAKE_UP_FILES + appFileMessage.getId();
+                    Object object = redisUtil.getObjectByKey(key);
+                    //清空redis缓存
+                    fileCommonUtils.cleanRedisData(appFileMessage.getId(),fileName);
+                    if (Objects.nonNull(object)) {
+                        DownloadMakeUpDto dto = channelObjectUtil.objectToSingleObject(object, DownloadMakeUpDto.class);
+                        channelMakeUpFile(dto,appFileMessage.getId(),fileName,filePath,lsFileName);
+                    }
                 } else {
                     //收到数据就刷新缓存值
                     redisUtil.saveByKeyWithExpire(AppRedisKey.FILE_DOWN_TIME.concat(appFileMessage.getMsg().getName()), null, 60L);
@@ -212,9 +234,15 @@ public class FileServiceImpl implements IFileService {
                                 if (file.exists()) {
                                     file.delete();
                                 }
-                                redisUtil.delete(AppRedisKey.FILE_PART_TIME.concat(appFileMessage.getMsg().getName()));
+                                //针对补召文件
-                                redisUtil.delete(AppRedisKey.RMQ_FILE_CONSUME_KEY.concat(fileName));
+                                String key = AppRedisKey.MAKE_UP_FILES + appFileMessage.getId();
-                                redisUtil.delete(AppRedisKey.FILE_DOWN_TIME.concat(appFileMessage.getMsg().getName()));
+                                Object object = redisUtil.getObjectByKey(key);
+                                //清空redis缓存
+                                fileCommonUtils.cleanRedisData(appFileMessage.getId(),fileName);
+                                if (Objects.nonNull(object)) {
+                                    DownloadMakeUpDto dto2 = channelObjectUtil.objectToSingleObject(object, DownloadMakeUpDto.class);
+                                    channelMakeUpFile(dto2,appFileMessage.getId(),fileName,filePath,lsFileName);
+                                }
                             } else {
                                 csEventLogs.setStatus(1);
                                 csEventLogs.setRemark("当前文件" + appFileMessage.getMsg().getFrameTotal() + "帧,这是第" + appFileMessage.getMsg().getFrameCurr() + "帧,记录成功!");
@@ -235,11 +263,16 @@ public class FileServiceImpl implements IFileService {
                         }
                     }
                 }
+                String userIndex = redisUtil.getObjectByKey("fileDownUserId"+appFileMessage.getId()+appFileMessage.getMsg().getName()).toString();
                 //推送mqtt
-                String json = "{fileName:"+appFileMessage.getMsg().getName()+",allStep:"+appFileMessage.getMsg().getFrameTotal()+",nowStep:"+ appFileMessage.getMsg().getFrameCurr() +"}";
+                String json = "{fileName:" + appFileMessage.getMsg().getName()
+                        + ",allStep:" + appFileMessage.getMsg().getFrameTotal()
+                        + ",nowStep:" + appFileMessage.getMsg().getFrameCurr()
+                        + ",userId:" + userIndex
+                        +"}";
                 publisher.send("/Web/Progress/" + appFileMessage.getId(), new Gson().toJson(json), 1, false);
                 if (!Objects.isNull(filePath)){
-                    redisUtil.saveByKey("downloadFilePath:" + appFileMessage.getId() + appFileMessage.getMsg().getName(),filePath);
+                    redisUtil.saveByKeyWithExpire("downloadFilePath:" + appFileMessage.getId() + appFileMessage.getMsg().getName(),filePath,60L);
                 }
             }
             //录波文件下载
@@ -353,9 +386,6 @@ public class FileServiceImpl implements IFileService {
             csEventLogs.setCompleteTime(LocalDateTime.now());
             //记录日志
             csEventLogsService.save(csEventLogs);
-            //清空缓存
-            redisUtil.delete(AppRedisKey.FILE_PART_TIME.concat(appFileMessage.getMsg().getName()));
-            redisUtil.delete(AppRedisKey.FILE_PART.concat(appFileMessage.getMsg().getName()));
             //删除临时文件
             String fileName = appFileMessage.getMsg().getName();
             String lsFileName = generalInfo.getBusinessTempPath() + File.separator + fileName.split(StrUtil.SLASH)[fileName.split(StrUtil.SLASH).length - 1];
@@ -365,6 +395,96 @@ public class FileServiceImpl implements IFileService {
             }
             //继续消费
             removeInfoUtils.deleteEventInfo(appFileMessage.getId(),appFileMessage.getMsg().getName());
+            //清空redis缓存
+            fileCommonUtils.cleanRedisData(appFileMessage.getId(),fileName);
+        }
+    }
+
+    @Override
+    public void downloadMakeUpFile(String nDid) {
+        try {
+            //判断客户端是否在线，在线再处理文件
+            String clientName = "NJCN-" + nDid.substring(nDid.length() - 6);
+            boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
+            if (mqttClient){
+                String key = AppRedisKey.MAKE_UP_FILES + nDid;
+                Object object = redisUtil.getObjectByKey(key);
+                if (Objects.nonNull(object)) {
+                    DownloadMakeUpDto dto = channelObjectUtil.objectToSingleObject(object, DownloadMakeUpDto.class);
+                    if (CollectionUtil.isNotEmpty(dto.getFileList())){
+                        Object object1 = channelObjectUtil.getDeviceMid(nDid);
+                        if (!Objects.isNull(object1)) {
+                            mid = (Integer) object1;
+                        }
+                        String file = dto.getFileList().get(0);
+                        fileCommonUtils.askFileInfo(nDid,mid,file);
+                        mid = mid + 1;
+                        if (mid > 10000) {
+                            mid = 1;
+                        }
+                        redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,mid);
+                        Thread.sleep(10000);
+                        String infoKey = AppRedisKey.PROJECT_INFO + nDid;
+                        FileDto.FileInfo info = channelObjectUtil.objectToSingleObject(redisUtil.getObjectByKey(infoKey), FileDto.FileInfo.class);
+                        deviceFtpFeignClient.downloadFile(nDid,file,info.getFileSize(),info.getFileCheck()).getData();
+                    }
+                }
+            } else {
+                throw new BusinessException(AlgorithmResponseEnum.DEV_OFFLINE);
+            }
+        } catch (Exception e) {
+            String key = AppRedisKey.MAKE_UP_FILES + nDid;
+            redisUtil.delete(key);
+            Object object = redisUtil.getObjectByKey(AppRedisKey.MAKE_UP_FILES + nDid);
+            if (Objects.nonNull(object)) {
+                DownloadMakeUpDto dto = channelObjectUtil.objectToSingleObject(object, DownloadMakeUpDto.class);
+                if (CollectionUtil.isNotEmpty(dto.getFileList())){
+                    String file = dto.getFileList().get(0);
+                    fileCommonUtils.cleanRedisData(nDid,file);
+                }
+            }
+        }
+    }
+
+    /**
+     * 处理补召文件
+     */
+    public void channelMakeUpFile(DownloadMakeUpDto dto, String nDid, String fileName, String oldPath, String lsFileName){
+        try {
+            //如果是补召文件,则将文件复制到补召目录下
+            moveFile(oldPath,getFilePath(fileName,nDid),lsFileName);
+            //删除临时文件
+            File file = new File(lsFileName);
+            if (file.exists()) {
+                file.delete();
+            }
+            //删除下载文件
+            fileStorageUtil.deleteFile(oldPath);
+            List<String> list = dto.getFileList();
+            list.removeIf(item -> item.equals(fileName));
+            dto.setFileList(list);
+            redisUtil.saveByKey(AppRedisKey.MAKE_UP_FILES + nDid,dto);
+            //判断是否还有缓存的文件
+            if (CollectionUtil.isNotEmpty(list)){
+                //推送进度条
+                String json = "{allStep:" + dto.getAllStep() * 2 + ",nowStep:" + (dto.getAllStep() - list.size()) + "}";
+                publisher.send("/dataOnlineRecruitment/Progress/" + dto.getLineId(), new Gson().toJson(json), 1, false);
+                //下载文件
+                downloadMakeUpFile(nDid);
+            } else {
+                //清空缓存文件
+                String key = AppRedisKey.MAKE_UP_FILES + nDid;
+                redisUtil.delete(key);
+                //推送进度条
+                String json = "{allStep:" + dto.getAllStep() * 2 + ",nowStep:" + dto.getAllStep() + "}";
+                publisher.send("/dataOnlineRecruitment/Progress/" + dto.getLineId(), new Gson().toJson(json), 1, false);
+                //调用方法
+                portableOffLogFeignClient.dataOnlineRecruitment(dto.getDevId(),dto.getLineId(),dto.getEngineeringName());
+            }
+        } catch (Exception e) {
+            String key = AppRedisKey.MAKE_UP_FILES + nDid;
+            redisUtil.delete(key);
+            fileCommonUtils.cleanRedisData(nDid,fileName);
         }
     }
 
@@ -570,5 +690,47 @@ public class FileServiceImpl implements IFileService {
      * -----------------------------------------------
      */
 
+    /**
+     * 迁移文件
+     */
+    public void moveFile(String oldPath, String newPath, String lsPath) {
+        try {
+            InputStream inputStream = fileStorageUtil.getFileStream(oldPath);
+            FileOutputStream fileOutputStream = new FileOutputStream(lsPath);
+            // 创建一个缓冲区
+            byte[] buffer = new byte[1024];
+            int bytesRead;
+            while ((bytesRead = inputStream.read(buffer)) != -1) {
+                fileOutputStream.write(buffer, 0, bytesRead);
+            }
+            File src = new File(lsPath);
+            src.getParentFile().mkdirs();
+            InputStream is = Files.newInputStream(src.toPath());
+            fileStorageUtil.uploadStreamSpecifyName(is, OssPath.DEV_MAKE_UP_PATH,newPath);
+            inputStream.close();
+            fileOutputStream.close();
+            is.close();
+        } catch (Exception e) {
+            throw new BusinessException(CsHarmonicResponseEnum.MAKE_UP_ERROR);
+        }
+    }
+
+    /**
+     * 调整文件路径
+     */
+    private String getFilePath(String path, String nDid) {
+        String[] parts = path.split("/");
+        StringBuilder sb = new StringBuilder();
+        boolean first = true;
+        for (int i = 3; i < parts.length; i++) {
+            if (!first) {
+                sb.append("/");
+            }
+            sb.append(parts[i]);
+            first = false;
+        }
+        return nDid + "/" + sb.toString();
+    }
+
 
 }