调整设备增量添加接口及实现

amos-boot-system-tzs/amos-boot-module-common/amos-boot-module-common-biz/src/main/java/com/yeejoin/amos/boot/module/common/biz/utils/TechParamUtil.java

@@ -41,7 +41,7 @@ public class TechParamUtil {
     }
 
     public static void main(String[] args) {
-        List<TechParamItem> techParamItems = getParamMetaList("2000", "2300", "23T0", "vehicleCylinder");
+        List<TechParamItem> techParamItems = getParamMetaList("3000", "3300", "3320");
         System.out.println(techParamItems);
     }
 

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/controller/DataHandlerController.java

@@ -256,8 +256,9 @@ public class DataHandlerController extends BaseController {
     @TycloudOperation(ApiLevel = UserType.AGENCY)
     @ApiOperation(httpMethod = "PUT", value = "增量添加ORG_BRANCH_CODE为50X综合搜索数据-设备", notes = "增量添加ORG_BRANCH_CODE为50X综合搜索数据-设备")
     @PutMapping(value = "/equip/addStatistData2Es")
-    public ResponseModel<Integer> addStatistData2EsBatch() {
-        return ResponseHelper.buildResponse(dataHandlerService.addStatistData2EsBatch());
+    public ResponseModel<Object> addStatistData2EsBatch() {
+        dataHandlerService.addStatistData2EsBatch();
+        return ResponseHelper.buildResponse(true);
     }
 
 

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/service/impl/DataHandlerServiceImpl.java

@@ -144,7 +144,6 @@ public class DataHandlerServiceImpl {
     private final IdxBizJgRegisterInfoMapper registerInfoMapper;
 
 
-
     /**
      * 安装告知压力管道历史数据修复-详情中的设备列表修改为汇总表格式
      *
@@ -1470,7 +1469,7 @@ public class DataHandlerServiceImpl {
         Map<String, TzBaseEnterpriseInfo> useCodeEnterpriseMap = units.stream().collect(Collectors.toMap(TzBaseEnterpriseInfo::getUseCode, Function.identity(), (k1, k2) -> k2));
         Iterable<EsBaseEnterpriseInfo> esBaseEnterpriseInfoIterable = enterpriseInfoDao.findAll();
         List<EsBaseEnterpriseInfo> esBaseEnterpriseInfos = Lists.newArrayList(esBaseEnterpriseInfoIterable);
-        Map<String, String> esUseCodeEnterpriseMap = esBaseEnterpriseInfos.stream().collect(Collectors.toMap(EsBaseEnterpriseInfo::getUseCode, EsBaseEnterpriseInfo::getUnitType,(k1,k2)->k2));
+        Map<String, String> esUseCodeEnterpriseMap = esBaseEnterpriseInfos.stream().collect(Collectors.toMap(EsBaseEnterpriseInfo::getUseCode, EsBaseEnterpriseInfo::getUnitType, (k1, k2) -> k2));
         LambdaQueryWrapper<TzsUserInfo> wrapper = new LambdaQueryWrapper<>();
         wrapper.eq(BaseEntity::getIsDelete, false)
                 .orderByDesc(BaseEntity::getSequenceNbr);
@@ -1536,7 +1535,7 @@ public class DataHandlerServiceImpl {
             Map<String, List<IdxBizJgInspectionDetectionInfo>> recordInspectInfosGroupByInspectTypeMap = lastedInspectInfosGroupByInspectType.stream().collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord));
 
             // 设备最新的检验信息-维度：设备
-            Map<String, Optional<IdxBizJgInspectionDetectionInfo>> recordLastInspectionMap = lastedInspectInfosGroupByInspectType.stream().filter(e->e.getInspectDate() != null).collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord, Collectors.maxBy(Comparator.comparing(IdxBizJgInspectionDetectionInfo::getInspectDate))));
+            Map<String, Optional<IdxBizJgInspectionDetectionInfo>> recordLastInspectionMap = lastedInspectInfosGroupByInspectType.stream().filter(e -> e.getInspectDate() != null).collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord, Collectors.maxBy(Comparator.comparing(IdxBizJgInspectionDetectionInfo::getInspectDate))));
             List<ESEquipmentInfo> esEquipmentInfos = refreshRecords.parallelStream().map(record -> {
                 ESEquipmentInfo esEquipmentInfo = null;
                 try {
@@ -1552,7 +1551,7 @@ public class DataHandlerServiceImpl {
                         esEquipmentInfo.setIS_DO_BUSINESS(dto.getIS_DO_BUSINESS());
                     }
                     // 最新检验信息-维度record
-                    IdxBizJgInspectionDetectionInfo inspectionDetectionInfo = Optional.ofNullable(recordLastInspectionMap.get(record)).flatMap(i->i).orElse(new IdxBizJgInspectionDetectionInfo());
+                    IdxBizJgInspectionDetectionInfo inspectionDetectionInfo = Optional.ofNullable(recordLastInspectionMap.get(record)).flatMap(i -> i).orElse(new IdxBizJgInspectionDetectionInfo());
                     // 最新维保信息-维度record
                     IdxBizJgMaintenanceRecordInfo lastMaintenanceRecordInfo = Optional.ofNullable(recordLastMaintMap.get(record)).filter(l -> !l.isEmpty()).map(list -> list.get(0)).orElse(new IdxBizJgMaintenanceRecordInfo());
                     // 最新检验信息-维度record、检验类型，存最新的一条
@@ -1597,74 +1596,92 @@ public class DataHandlerServiceImpl {
     }
 
 
-    public Integer addStatistData2EsBatch() {
+    public void addStatistData2EsBatch() {
         log.info("增量添加ORG_BRANCH_CODE为50X综合搜索数据-设备信息入库开始");
         StopWatch watch = new StopWatch();
         watch.start();
-        List<String> refreshRecords = useInfoService.getBaseMapper().selectAddDataRecords();
-        while (!refreshRecords.isEmpty()) {
-            StopWatch watch0 = new StopWatch();
-            watch0.start();
-            List<Map<String, Object>> details = useInfoService.getBaseMapper().queryDetailBatch(refreshRecords);
-            Map<String, Map<String, Object>> recordDetailMap = details.stream().collect(Collectors.toMap(e -> (String) e.get("SEQUENCE_NBR"), Function.identity(), (k1, k2) -> k2));
-
-            // 设备最新的维保信息-维度：设备
-            List<IdxBizJgMaintenanceRecordInfo> lastMaintenanceRecordInfos = maintenanceRecordInfoService.getBaseMapper().selectLastedMainInfoBatch(refreshRecords);
-            Map<String, List<IdxBizJgMaintenanceRecordInfo>> recordLastMaintMap = lastMaintenanceRecordInfos.stream().collect(Collectors.groupingBy(IdxBizJgMaintenanceRecordInfo::getRecord));
-
-            // 设备、各检验类型下最新的检验信息-维度：设备、检验类型
-            List<IdxBizJgInspectionDetectionInfo> lastedInspectInfosGroupByInspectType = idxBizJgInspectionDetectionInfoService.getBaseMapper().selectLastedGroupByInspectTypeBatch(refreshRecords);
-            Map<String, List<IdxBizJgInspectionDetectionInfo>> recordInspectInfosGroupByInspectTypeMap = lastedInspectInfosGroupByInspectType.stream().collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord));
-
-            // 设备最新的检验信息-维度：设备
-            Map<String, Optional<IdxBizJgInspectionDetectionInfo>> recordLastInspectionMap = lastedInspectInfosGroupByInspectType.stream().filter(e->e.getInspectDate() != null).collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord, Collectors.maxBy(Comparator.comparing(IdxBizJgInspectionDetectionInfo::getInspectDate))));
-            List<ESEquipmentInfo> esEquipmentInfos = refreshRecords.parallelStream().map(record -> {
-                ESEquipmentInfo esEquipmentInfo = null;
-                try {
-                    esEquipmentInfo = new ESEquipmentInfo();
-                    Map<String, Object> detail = recordDetailMap.get(record);
-                    StatisticsDataUpdateService.formatUseDate(detail);
-                    BeanUtil.copyProperties(detail, esEquipmentInfo, true);
-                    Optional<ESEquipmentCategoryDto> esOptional = esEquipmentCategory.findById(record);
-                    if (esOptional.isPresent()) {
-                        ESEquipmentCategoryDto dto = esOptional.get();
-                        esEquipmentInfo.setUSC_UNIT_CREDIT_CODE(dto.getUSC_UNIT_CREDIT_CODE());
-                        esEquipmentInfo.setUSC_UNIT_NAME(dto.getUSC_UNIT_NAME());
-                        esEquipmentInfo.setIS_DO_BUSINESS(dto.getIS_DO_BUSINESS());
+        Integer count = useInfoService.getBaseMapper().selectAddDataRecordsCount();
+        Integer times = 0;
+        if (count != 0) {
+            times = count / 5000;
+            int last = count % 5000;
+            if (last > 0) {
+                times++;
+            }
+        } else {
+            return;
+        }
+        Page<String> recordDtoPage = new Page<>();
+        for (int j = 0; j <= times; j++) {
+            recordDtoPage.setCurrent(j + 1);
+            recordDtoPage.setSize(5000);
+
+            Page<String> refreshRecords = useInfoService.getBaseMapper().selectAddDataRecords(recordDtoPage);
+
+            if(!ObjectUtils.isEmpty(refreshRecords)&& refreshRecords.getRecords().size() > 0){
+                List<String> records = refreshRecords.getRecords();
+                StopWatch watch0 = new StopWatch();
+                watch0.start();
+                List<Map<String, Object>> details = useInfoService.getBaseMapper().queryDetailBatch(records);
+                Map<String, Map<String, Object>> recordDetailMap = details.stream().collect(Collectors.toMap(e -> (String) e.get("SEQUENCE_NBR"), Function.identity(), (k1, k2) -> k2));
+
+                // 设备最新的维保信息-维度：设备
+                List<IdxBizJgMaintenanceRecordInfo> lastMaintenanceRecordInfos = maintenanceRecordInfoService.getBaseMapper().selectLastedMainInfoBatch(records);
+                Map<String, List<IdxBizJgMaintenanceRecordInfo>> recordLastMaintMap = lastMaintenanceRecordInfos.stream().collect(Collectors.groupingBy(IdxBizJgMaintenanceRecordInfo::getRecord));
+
+                // 设备、各检验类型下最新的检验信息-维度：设备、检验类型
+                List<IdxBizJgInspectionDetectionInfo> lastedInspectInfosGroupByInspectType = idxBizJgInspectionDetectionInfoService.getBaseMapper().selectLastedGroupByInspectTypeBatch(records);
+                Map<String, List<IdxBizJgInspectionDetectionInfo>> recordInspectInfosGroupByInspectTypeMap = lastedInspectInfosGroupByInspectType.stream().collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord));
+
+                // 设备最新的检验信息-维度：设备
+                Map<String, Optional<IdxBizJgInspectionDetectionInfo>> recordLastInspectionMap = lastedInspectInfosGroupByInspectType.stream().filter(e -> e.getInspectDate() != null).collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord, Collectors.maxBy(Comparator.comparing(IdxBizJgInspectionDetectionInfo::getInspectDate))));
+                List<ESEquipmentInfo> esEquipmentInfos = records.parallelStream().map(record -> {
+                    ESEquipmentInfo esEquipmentInfo = null;
+                    try {
+                        esEquipmentInfo = new ESEquipmentInfo();
+                        Map<String, Object> detail = recordDetailMap.get(record);
+                        StatisticsDataUpdateService.formatUseDate(detail);
+                        BeanUtil.copyProperties(detail, esEquipmentInfo, true);
+                        Optional<ESEquipmentCategoryDto> esOptional = esEquipmentCategory.findById(record);
+                        if (esOptional.isPresent()) {
+                            ESEquipmentCategoryDto dto = esOptional.get();
+                            esEquipmentInfo.setUSC_UNIT_CREDIT_CODE(dto.getUSC_UNIT_CREDIT_CODE());
+                            esEquipmentInfo.setUSC_UNIT_NAME(dto.getUSC_UNIT_NAME());
+                            esEquipmentInfo.setIS_DO_BUSINESS(dto.getIS_DO_BUSINESS());
+                        }
+                        // 最新检验信息-维度record
+                        IdxBizJgInspectionDetectionInfo inspectionDetectionInfo = Optional.ofNullable(recordLastInspectionMap.get(record)).flatMap(i -> i).orElse(new IdxBizJgInspectionDetectionInfo());
+                        // 最新维保信息-维度record
+                        IdxBizJgMaintenanceRecordInfo lastMaintenanceRecordInfo = Optional.ofNullable(recordLastMaintMap.get(record)).filter(l -> !l.isEmpty()).map(list -> list.get(0)).orElse(new IdxBizJgMaintenanceRecordInfo());
+                        // 最新检验信息-维度record、检验类型，存最新的一条
+                        List<IdxBizJgInspectionDetectionInfo> inspectionDetectionInfos = recordInspectInfosGroupByInspectTypeMap.getOrDefault(record, new ArrayList<>());
+                        StatisticsDataUpdateService.formatInspectDate(esEquipmentInfo, inspectionDetectionInfo, record);
+                        esEquipmentInfo.setIssueDate(getIssueDate(esEquipmentInfo.getUSE_ORG_CODE()));
+                        esEquipmentInfo.setMAINTAIN_UNIT(lastMaintenanceRecordInfo.getMeUnitCreditCode());
+                        esEquipmentInfo.setMAINTAIN_UNIT_NAME(lastMaintenanceRecordInfo.getMeUnitName());
+                        esEquipmentInfo.setInspections(BeanUtil.copyToList(inspectionDetectionInfos, ESEquipmentInfo.Inspection.class));
+                        esEquipmentInfo.setMaintenances(lastMaintenanceRecordInfo.getSequenceNbr() != null ? Collections.singletonList(BeanUtil.copyProperties(lastMaintenanceRecordInfo, ESEquipmentInfo.Maintenance.class)) : new ArrayList<>());
+                        esEquipmentInfo.setTechParams(this.buildTechParamByEquList(record, esEquipmentInfo.getEQU_LIST_CODE()));
+                    } catch (Exception e) {
+                        // 异常数据跳过
+                        log.error("设备刷数据处理失败：{}", record, e);
                     }
-                    // 最新检验信息-维度record
-                    IdxBizJgInspectionDetectionInfo inspectionDetectionInfo = Optional.ofNullable(recordLastInspectionMap.get(record)).flatMap(i->i).orElse(new IdxBizJgInspectionDetectionInfo());
-                    // 最新维保信息-维度record
-                    IdxBizJgMaintenanceRecordInfo lastMaintenanceRecordInfo = Optional.ofNullable(recordLastMaintMap.get(record)).filter(l -> !l.isEmpty()).map(list -> list.get(0)).orElse(new IdxBizJgMaintenanceRecordInfo());
-                    // 最新检验信息-维度record、检验类型，存最新的一条
-                    List<IdxBizJgInspectionDetectionInfo> inspectionDetectionInfos = recordInspectInfosGroupByInspectTypeMap.getOrDefault(record, new ArrayList<>());
-                    StatisticsDataUpdateService.formatInspectDate(esEquipmentInfo, inspectionDetectionInfo, record);
-                    esEquipmentInfo.setIssueDate(getIssueDate(esEquipmentInfo.getUSE_ORG_CODE()));
-                    esEquipmentInfo.setMAINTAIN_UNIT(lastMaintenanceRecordInfo.getMeUnitCreditCode());
-                    esEquipmentInfo.setMAINTAIN_UNIT_NAME(lastMaintenanceRecordInfo.getMeUnitName());
-                    esEquipmentInfo.setInspections(BeanUtil.copyToList(inspectionDetectionInfos, ESEquipmentInfo.Inspection.class));
-                    esEquipmentInfo.setMaintenances(lastMaintenanceRecordInfo.getSequenceNbr() != null ? Collections.singletonList(BeanUtil.copyProperties(lastMaintenanceRecordInfo, ESEquipmentInfo.Maintenance.class)) : new ArrayList<>());
-                    esEquipmentInfo.setTechParams(this.buildTechParamByEquList(record, esEquipmentInfo.getEQU_LIST_CODE()));
-                } catch (Exception e) {
-                    // 异常数据跳过
-                    log.error("设备刷数据处理失败：{}", record, e);
+                    return esEquipmentInfo;
+                }).collect(Collectors.toList());
+                watch0.stop();
+                log.warn("多线程处理查询设备详情耗时：{}", watch0.getTotalTimeSeconds());
+                esEquipmentInfos = esEquipmentInfos.stream().filter(e -> StringUtils.isNotEmpty(e.getSEQUENCE_NBR())).collect(Collectors.toList());
+                if (!esEquipmentInfos.isEmpty()) {
+                    StopWatch watch1 = new StopWatch();
+                    watch1.start();
+                    esEquipmentDao.saveAll(esEquipmentInfos);
+                    watch1.stop();
+                    log.warn("es批量入库条数：{}，耗时：{}s", esEquipmentInfos.size(), watch1.getTotalTimeSeconds());
                 }
-                return esEquipmentInfo;
-            }).collect(Collectors.toList());
-            watch0.stop();
-            log.warn("多线程处理查询设备详情耗时：{}", watch0.getTotalTimeSeconds());
-            esEquipmentInfos = esEquipmentInfos.stream().filter(e -> StringUtils.isNotEmpty(e.getSEQUENCE_NBR())).collect(Collectors.toList());
-            if (!esEquipmentInfos.isEmpty()) {
-                StopWatch watch1 = new StopWatch();
-                watch1.start();
-                esEquipmentDao.saveAll(esEquipmentInfos);
-                watch1.stop();
-                log.warn("es批量入库条数：{}，耗时：{}s", esEquipmentInfos.size(), watch1.getTotalTimeSeconds());
             }
         }
         watch.stop();
         log.info("增量添加ORG_BRANCH_CODE为50X综合搜索数据-设备信息入库结束，耗时：{}秒", watch.getTotalTimeSeconds());
-        return refreshRecords.size();
     }
 
 

amos-boot-system-tzs/amos-boot-module-ymt/amos-boot-module-ymt-api/src/main/java/com/yeejoin/amos/boot/module/ymt/api/mapper/IdxBizJgUseInfoMapper.java

 package com.yeejoin.amos.boot.module.ymt.api.mapper;
 
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.yeejoin.amos.boot.module.ymt.api.dto.EquipWaitRefreshDataQualityScore;
 import com.yeejoin.amos.boot.module.ymt.api.dto.RefreshDataDto;
 import com.yeejoin.amos.boot.module.ymt.api.entity.IdxBizJgUseInfo;
@@ -34,5 +35,7 @@ public interface IdxBizJgUseInfoMapper extends BaseMapper<IdxBizJgUseInfo> {
 
     List<Map<String,Object>> queryDetailBatch(@Param("records") List<String> records);
 
-    List<String> selectAddDataRecords();
+    Page<String> selectAddDataRecords(Page<String> page);
+
+    Integer selectAddDataRecordsCount();
 }

amos-boot-system-tzs/amos-boot-module-ymt/amos-boot-module-ymt-api/src/main/resources/mapper/IdxBizJgUseInfoMapper.xml

@@ -149,7 +149,10 @@
             </foreach>
     </select>
     <select id="selectAddDataRecords" resultType="java.lang.String">
-        select "RECORD" from amos_tzs_biz.idx_biz_jg_supervision_info where "ORG_BRANCH_CODE" = '50*X'
+        select "RECORD" from amos_tzs_biz.idx_biz_jg_supervision_info where "ORG_BRANCH_CODE" = '50*X' order by "RECORD"
+    </select>
+    <select id="selectAddDataRecordsCount" resultType="java.lang.Integer">
+        select count(1) from amos_tzs_biz.idx_biz_jg_supervision_info where "ORG_BRANCH_CODE" = '50*X'
     </select>
 
 </mapper>