Merge remote-tracking branch 'origin/develop_tzs_main' into develop_tzs_main

amos-boot-system-tzs/amos-boot-module-common/amos-boot-module-common-api/src/main/java/com/yeejoin/amos/boot/module/common/api/dto/ESEquipmentCategoryDto.java

 package com.yeejoin.amos.boot.module.common.api.dto;
 
-import com.baomidou.mybatisplus.annotation.TableField;
-import com.yeejoin.amos.boot.biz.common.annotation.FieldDisplayDefine;
 import lombok.Data;
 import lombok.experimental.Accessors;
 import org.springframework.data.annotation.Id;

amos-boot-system-tzs/amos-boot-module-common/amos-boot-module-common-api/src/main/java/com/yeejoin/amos/boot/module/common/api/entity/ESEquipmentInfo.java

@@ -652,4 +652,7 @@ public class ESEquipmentInfo {
         @Field(type = FieldType.Keyword)
         private String meMaster1Phone;
     }
+
+    @Field(type = FieldType.Keyword)
+    private String version;
 }

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

+package com.yeejoin.amos.boot.module.common.biz.service.impl;
+
+import lombok.RequiredArgsConstructor;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.springframework.stereotype.Service;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static com.yeejoin.amos.boot.module.common.api.constant.TZSCommonConstant.ES_INDEX_NAME_EQUIPMENT_INFO;
+
+@Service
+@RequiredArgsConstructor
+public class ESEquipmentInfoService {
+
+    private final RestHighLevelClient restHighLevelClient;
+
+    public List<String> findSequenceNbrByIds(List<String> ids) {
+        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
+                .query(QueryBuilders.termsQuery("SEQUENCE_NBR.keyword", ids))
+                .fetchSource(false);
+        SearchRequest request = new SearchRequest(ES_INDEX_NAME_EQUIPMENT_INFO)
+                .source(sourceBuilder);
+        try {
+            SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
+            return Arrays.stream(response.getHits().getHits())
+                    .map(SearchHit::getId)
+                    .collect(Collectors.toList());
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

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

@@ -3,9 +3,26 @@ package com.yeejoin.amos.boot.module.common.biz.service.impl;
 import com.yeejoin.amos.boot.module.common.api.dao.ESEquipmentCategory;
 import com.yeejoin.amos.boot.module.common.api.dto.ESEquipmentCategoryDto;
 import lombok.RequiredArgsConstructor;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
+import org.springframework.data.elasticsearch.core.SearchHits;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
 import org.springframework.stereotype.Service;
 
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static com.yeejoin.amos.boot.module.common.api.constant.TZSCommonConstant.ES_INDEX_NAME_JG_ALL;
+
 @Service
 @RequiredArgsConstructor
 public class EquipmentCategoryService {
@@ -14,10 +31,28 @@ public class EquipmentCategoryService {
 
     private final ElasticsearchOperations elasticsearchOperations;
 
+    private final RestHighLevelClient restHighLevelClient;
+
     public ESEquipmentCategoryDto saveWithImmediateRefresh(ESEquipmentCategoryDto dto) {
         ESEquipmentCategoryDto saved = equipmentCategoryDao.save(dto);
         // 手动触发索引刷新
         elasticsearchOperations.indexOps(ESEquipmentCategoryDto.class).refresh();
         return saved;
     }
+
+    public List<String> findSequenceNbrByIds(List<String> ids) {
+        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
+                .query(QueryBuilders.termsQuery("SEQUENCE_NBR.keyword", ids))
+                .fetchSource(false);
+        SearchRequest request = new SearchRequest(ES_INDEX_NAME_JG_ALL)
+                .source(sourceBuilder);
+        try {
+            SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
+            return Arrays.stream(response.getHits().getHits())
+                    .map(SearchHit::getId)
+                    .collect(Collectors.toList());
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }
\ No newline at end of file

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

@@ -159,7 +159,7 @@ public class DataDockController extends BaseController {
     @ApiOperation(httpMethod = "POST", value = "工业 、公用、长输管道批量保存", notes = "工业 、公用、长输管道批量保存")
     @RestEventTrigger(value = "operateLogRestEventHandler")
     public ResponseModel<?> savePipingData(@RequestBody Map<String, Object> paramMap) {
-        return ResponseHelper.buildResponse(dataDockService.savePipingData(paramMap, getSelectedOrgInfo().getCompany()));
+        return ResponseHelper.buildResponse(dataDockService.savePipingData(paramMap, getSelectedOrgInfo()));
     }
 
     /**

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

@@ -456,4 +456,13 @@ public class DataHandlerController extends BaseController {
         return ResponseHelper.buildResponse(dataHandlerService.insertEquipFromDb2NewEs(paramMap));
     }
 
+    @TycloudOperation(ApiLevel = UserType.AGENCY)
+    @PutMapping(value = "/sync-equip/shard")
+    @ApiOperation(httpMethod = "PUT", value = "分片对齐设备的数据与es数据", notes = "分片对齐设备的数据与es数据")
+    public ResponseModel<Boolean> syncEquipShard(@RequestParam Integer shard,
+                                                 @RequestParam Integer slots,
+                                                 @RequestParam String batchNo){
+        return ResponseHelper.buildResponse(dataHandlerService.syncEquipShard(shard, slots, batchNo));
+    }
+
 }
\ No newline at end of file

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/job/DataConsistencyJob.java

+package com.yeejoin.amos.boot.module.jg.biz.job;
+
+import cn.hutool.core.date.DateUtil;
+import com.yeejoin.amos.component.feign.model.FeignClientResult;
+import com.yeejoin.amos.component.robot.AmosRequestContext;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.cloud.client.discovery.DiscoveryClient;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestTemplate;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class DataConsistencyJob {
+
+    private final AmosRequestContext requestContext;
+
+    private final DiscoveryClient discoveryClient;
+
+    private final RestTemplate restTemplate; // 需要提前注入
+
+    @Value("${spring.application.name}")
+    private String applicationName;
+
+//    @Scheduled(cron = "0 0 3 * * ?")
+//    @SchedulerLock(name = "equipConsistencyJob", lockAtMostFor = "PT1H")
+    public void execute() {
+        List<ServiceInstance> instanceList = discoveryClient.getInstances(applicationName);
+        int slots = instanceList.size();
+        Executor executor = Executors.newFixedThreadPool(slots);
+        HttpHeaders headers = this.builderHeaders();
+        HttpEntity<?> requestEntity = new HttpEntity<>(headers);
+        List<CompletableFuture<Void>> futures = new ArrayList<>();
+        String batchNo = DateUtil.today();
+        for (int i = 0; i < slots; i++) {
+            int index = i;
+            ServiceInstance instance = instanceList.get(i);
+            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+                String baseUrl = instance.getUri().toString();
+                String apiUrl = baseUrl + "/jg/dataHandler/sync-equip/shard?shard={shard}&slots={slots}&batchNo={batchNo}";
+                Map<String, Object> params = new HashMap<>();
+                params.put("shard", index);
+                params.put("slots", slots);
+                params.put("batchNo", batchNo);
+                restTemplate.exchange(
+                        apiUrl,
+                        HttpMethod.PUT,
+                        requestEntity,
+                        FeignClientResult.class,
+                        params
+                );
+            }, executor);
+            futures.add(future);
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
+                    .exceptionally(ex -> {
+                        log.error("任务执行异常", ex);
+                        return null;
+                    })
+                    .join();
+        }
+    }
+
+    private HttpHeaders builderHeaders() {
+        HttpHeaders httpheaders = new HttpHeaders();
+        httpheaders.add("appKey", requestContext.getAppKey());
+        httpheaders.add("product", requestContext.getProduct());
+        httpheaders.add("token", requestContext.getToken());
+        httpheaders.setContentType(MediaType.APPLICATION_JSON);
+        return httpheaders;
+    }
+
+}

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/DataDockServiceImpl.java

@@ -20,6 +20,7 @@ import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import com.yeejoin.amos.boot.biz.common.bo.CompanyBo;
 import com.yeejoin.amos.boot.biz.common.bo.ReginParams;
+import com.yeejoin.amos.boot.biz.common.entity.BaseEntity;
 import com.yeejoin.amos.boot.biz.common.utils.*;
 import com.yeejoin.amos.boot.module.common.api.converter.CommonCustomConverter;
 import com.yeejoin.amos.boot.module.common.api.dao.ESEquipmentCategory;
@@ -54,8 +55,8 @@ import com.yeejoin.amos.boot.module.jg.biz.utils.CodeUtil;
 import com.yeejoin.amos.boot.module.jg.biz.utils.JsonUtils;
 import com.yeejoin.amos.boot.module.ymt.api.entity.*;
 import com.yeejoin.amos.boot.module.ymt.api.enums.ApplicationFormTypeEnum;
-import com.yeejoin.amos.boot.module.ymt.api.enums.EquipmentEnum;
 import com.yeejoin.amos.boot.module.ymt.api.enums.EquipmentClassifityEnum;
+import com.yeejoin.amos.boot.module.ymt.api.enums.EquipmentEnum;
 import com.yeejoin.amos.boot.module.ymt.api.enums.FlowStatusEnum;
 import com.yeejoin.amos.boot.module.ymt.api.mapper.*;
 import com.yeejoin.amos.feign.privilege.Privilege;
@@ -70,9 +71,8 @@ import org.apache.commons.lang3.math.NumberUtils;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.core.io.Resource;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
-import org.springframework.transaction.support.TransactionSynchronization;
-import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.ObjectUtils;
 import org.springframework.web.multipart.MultipartFile;
 import org.typroject.tyboot.core.foundation.context.RequestContext;
@@ -89,9 +89,11 @@ import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.function.Function;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
 
 import static com.alibaba.fastjson.JSON.toJSONString;
 import static com.yeejoin.amos.boot.module.common.api.constant.TZSCommonConstant.EQU_CATEGORY_CODE_GDYLRQ;
@@ -174,11 +176,9 @@ public class DataDockServiceImpl {
     private final IdxBizJgConstructionInfoMapper idxBizJgConstructionInfoMapper;
     private final IdxBizJgOtherInfoMapper idxBizJgOtherInfoMapper;
     private final IdxBizJgInspectionDetectionInfoMapper idxBizJgInspectionDetectionInfoMapper;
-
+    private final ESEquipmentCategory esEquipmentCategoryDao;
     private final PieLineDataChangeServiceImpl pieLineDataChangeService;
 
-    private final ElevatorDataTransactionService elevatorDataTransactionService;
-
     @Value("${equip.detail.path:/mixuap?appId=1742358052905971713&id=1734100233714954241&formType=detail&record=%s&DATA_SOURCE=%s}")
     private String equipRoutePath;
 
@@ -862,7 +862,9 @@ public class DataDockServiceImpl {
             equipmentCategoryDto.setSTATUS("已认领");
             equipmentCategoryDto.setIS_DO_BUSINESS(Boolean.TRUE);
             equipmentCategoryDto.setIS_COMPLETE_XA(Objects.equals("0", isCompleteXa));
+            if (ValidationUtil.isEmpty(equipmentCategoryDto.getProblemStatus())) {
                 equipmentCategoryDto.setProblemStatus(SafetyProblemStatusEnum.HANDLED.getCode());
+            }
             esEquipmentCategory.save(equipmentCategoryDto);
         }
     }
@@ -1970,7 +1972,7 @@ public class DataDockServiceImpl {
     }
 
     @Transactional(rollbackFor = Exception.class)
-    public Object savePipingData(Map<String, Object> paramMap, CompanyBo company) {
+    public Object savePipingData(Map<String, Object> paramMap, ReginParams reginParams) {
         // 获取数据
         List<String> records = new ArrayList<>();
         try {
@@ -1998,17 +2000,16 @@ public class DataDockServiceImpl {
                     }
                 }
             }
-            ReginParams reginParams = JSONObject.parseObject(redisUtils.get(RedisKey.buildReginKey(RequestContext.getExeUserId(), RequestContext.getToken())).toString(), ReginParams.class);
             // 监管——历史——管道——批量
             String dataSource = "jg_his_gd_pl";
             // 公共参数提取
             EquipRequestParamsDto paramsDto = this.extractCommonData(dataSource, reginParams, equipInfo, equLists);
             // 是否工业管道
             boolean isGYGD = "8300".equals(paramsDto.getEquCategoryCode());
-            boolean haveUseRegistration = false;
+            List<JgUseRegistrationManage> jgUseRegistrationManages = new ArrayList<>();
             // 确保使用登记证和工程装置在该单位下都是不存在的，方可导入数据
             if (isGYGD) {
-                haveUseRegistration = this.checkTheUnitRegManageAlreadyExists(paramsDto, haveUseRegistration);
+                jgUseRegistrationManages = this.checkTheUnitRegManageAlreadyExists(paramsDto);
             }
             idxBizJgProjectContraptionServiceImpl.checkTheUnitProConMatching(paramsDto);
             this.checkTheUnitProConAlreadyExists(paramsDto);
@@ -2020,28 +2021,24 @@ public class DataDockServiceImpl {
                 // 生成使用登记表信息 + 设备关系表 + 历史表
                 JgUseRegistration useReg = this.saveUseRegAndEq(paramsDto, records, proCon);
                 // 生成使用登记证表数据
-                if (!haveUseRegistration) {
+                if (jgUseRegistrationManages.isEmpty()) {
                     JgUseRegistrationManage regManage = this.saveUseRegManage(paramsDto, useReg);
                     // 流水表
                     this.saveCertChangeRecord(useReg, regManage, records, paramsDto);
                 } else {
-                    // 1.更新使用登记证信息
-                    JgUseRegistrationManage jgUseRegistrationManage = jgUseRegistrationManageService.lambdaQuery()
-                            .eq(JgUseRegistrationManage::getUseRegistrationCode, paramsDto.getUseOrgCode())
-                            .eq(JgUseRegistrationManage::getIsDelete, 0)
-                            .one();
-                    if (jgUseRegistrationManage != null) {
-                        jgUseRegistrationManage.setSuperviseOrgCode(paramsDto.getOrgBranchCode());
-                        jgUseRegistrationManage.setSuperviseOrgName(paramsDto.getOrgBranchName());
-                        jgUseRegistrationManage.setEquUseAddress(Stream.of(paramsDto.getProvinceName(), paramsDto.getCityName(), paramsDto.getCountyName(), paramsDto.getStreetName(), paramsDto.getAddress())
+                    // 1.更新使用登记证信息，健壮处理，系统业务限制一个证号只能有一个证管理表
+                    LambdaUpdateWrapper<JgUseRegistrationManage> manageLambdaUpdateWrapper = new LambdaUpdateWrapper<>();
+                    manageLambdaUpdateWrapper.set(JgUseRegistrationManage::getSuperviseOrgCode, paramsDto.getOrgBranchCode());
+                    manageLambdaUpdateWrapper.set(JgUseRegistrationManage::getSuperviseOrgName, paramsDto.getOrgBranchName());
+                    manageLambdaUpdateWrapper.set(JgUseRegistrationManage::getEquUseAddress, Stream.of(paramsDto.getProvinceName(), paramsDto.getCityName(), paramsDto.getCountyName(), paramsDto.getStreetName(), paramsDto.getAddress())
                             .map(value -> value == null || "null".equals(value) ? "" : value)
                             .collect(Collectors.joining()));
-                        jgUseRegistrationManage.setRecDate(new Date());
-                        jgUseRegistrationManageService.updateById(jgUseRegistrationManage);
-                    }
+                    manageLambdaUpdateWrapper.set(JgUseRegistrationManage::getRecDate, new Date());
+                    manageLambdaUpdateWrapper.eq(BaseEntity::getSequenceNbr, jgUseRegistrationManages.stream().map(BaseEntity::getSequenceNbr).collect(Collectors.toList()));
+                    jgUseRegistrationManageService.update(manageLambdaUpdateWrapper);
                 }
             }
-            this.createResumePipeline(proCon.getSequenceNbr(), String.format(pipelineRoutePath, proCon.getSequenceNbr() + ""), company);
+            this.createResumePipeline(proCon.getSequenceNbr(), String.format(pipelineRoutePath, proCon.getSequenceNbr() + ""), reginParams.getCompany());
             eventPublisher.publish(new EquipCreateOrEditEvent(this, BusinessTypeEnum.JG_NEW_PROJECT.name(), Sets.newHashSet(proCon.getSequenceNbr() + ""), EquipCreateOrEditEvent.EquipType.project));
             return Boolean.TRUE;
         } catch (Exception e) {
@@ -2113,17 +2110,9 @@ public class DataDockServiceImpl {
      *
      * @param paramsDto 设备请求参数
      */
-    public boolean checkTheUnitRegManageAlreadyExists(EquipRequestParamsDto paramsDto, boolean haveUseRegistration) {
+    public List<JgUseRegistrationManage> checkTheUnitRegManageAlreadyExists(EquipRequestParamsDto paramsDto) {
         // 1. 判断是否使用未来系统生成编号
         idxBizJgRegisterInfoServiceImpl.checkUseRegistrationCode(paramsDto.getUseOrgCode(), "unit");
-        // 2. 获取使用登记管理列表
-//        List<JgUseRegistration> useRegistrationList = jgUseRegistrationMapper.selectList(
-//                new LambdaQueryWrapper<JgUseRegistration>()
-//                        .eq(JgUseRegistration::getUseRegistrationCode, paramsDto.getUseOrgCode())
-//                        .eq(JgUseRegistration::getIsDelete, 0)
-//                        .eq(JgUseRegistration::getStatus, "已完成")
-//                        .orderByDesc(JgUseRegistration::getRecDate)
-//                        .last("limit 1"));
         List<JgUseRegistrationManage> useRegistrationManageList = jgUseRegistrationManageMapper.selectList(
                 new LambdaQueryWrapper<JgUseRegistrationManage>()
                         .eq(JgUseRegistrationManage::getUseRegistrationCode, paramsDto.getUseOrgCode())
@@ -2132,15 +2121,11 @@ public class DataDockServiceImpl {
                         .orderByDesc(JgUseRegistrationManage::getRecDate));
 
         if (!useRegistrationManageList.isEmpty()) {
-//            if (useRegistrationList.stream().filter(map -> !"1".equals(map.getRegType())).count() > 1) {
-//                throw new BadRequest("该使用登记证编号已由系统自动生成，不能导入，请确认使用登记证编号是否正确！");
-//            }
             if (useRegistrationManageList.stream().anyMatch(map -> !paramsDto.getCompanyCode().equals(map.getUseUnitCreditCode()))) {
                 throw new BadRequest("该使用登记证编号在其他企业使用，不能导入，请确认使用登记证编号是否正确！");
             }
-            haveUseRegistration = true;
         }
-        return haveUseRegistration;
+        return useRegistrationManageList;
     }
 
     /**
@@ -2722,17 +2707,29 @@ public class DataDockServiceImpl {
      * @param equLists 数据集
      * @return 保存结果
      */
+    @Transactional
     public Set<String> saveElevatorData(List<Map<String, Object>> equLists) {
         RequestContextWrapper contextWrapper = RequestContextWrapper.capture();
-        Set<String> recordSet = Sets.newConcurrentHashSet();
-        Set<String> resultSet = Sets.newConcurrentHashSet();
+        Set<String> successfulRecordSet = Sets.newConcurrentHashSet();
+        Set<String> failedRecordSet = Sets.newConcurrentHashSet();
+        Set<String> failedResultSet = Sets.newConcurrentHashSet();
 
         // 分批处理大小
         int batchSize = 100;
         int totalSize = equLists.size();
-        int successCount = 0;
+        int successfulCount = 0;
         int failedCount = 0;
 
+        // 从equList中取出record
+        Set<String> recordList = equLists.stream().map(equ -> MapUtil.getStr(equ, "record")).collect(Collectors.toSet());
+        // 记录es数据快照
+        Map<String, ESEquipmentCategoryDto> originalEsResultMap = StreamSupport.stream(
+                        esEquipmentCategoryDao.findAllById(recordList).spliterator(), false)
+                .collect(Collectors.toMap(
+                        ESEquipmentCategoryDto::getSEQUENCE_NBR,
+                        Function.identity()
+                ));
+
         // 分批处理数据
         for (int i = 0; i < totalSize; i += batchSize) {
             int endIndex = Math.min(i + batchSize, totalSize);
@@ -2743,7 +2740,7 @@ public class DataDockServiceImpl {
                     .map(equ -> CompletableFuture.supplyAsync(() -> {
                         contextWrapper.apply();
                         try {
-                            Object resultObj = elevatorDataTransactionService.saveSingleElevatorData(equ);
+                            Object resultObj = this.saveElevatorDataInTransaction(equ, "jg_his_xa", null);
                             Map<String, Object> result = new HashMap<>();
                             result.put("success", true);
                             result.put("data", resultObj);
@@ -2769,17 +2766,19 @@ public class DataDockServiceImpl {
                     if ((Boolean) result.get("success")) {
                         Object resultObj = result.get("data");
                         if (resultObj instanceof String) {
-                            recordSet.add(resultObj.toString());
-                            successCount++;
+                            successfulRecordSet.add(resultObj.toString());
+                            successfulCount++;
                         }
+                        // 记录做了其他业务则不能更新的record
                         if (resultObj instanceof Map) {
-                            resultSet.add(JSONObject.toJSONString(resultObj));
+                            failedResultSet.add(JSONObject.toJSONString(resultObj));
                             failedCount++;
                         }
                     } else {
                         Exception e = (Exception) result.get("exception");
                         Map<String, Object> equ = (Map<String, Object>) result.get("equ");
-                        handleProcessException(e, equ, resultSet);
+                        failedRecordSet.add(MapUtil.getStr(equ, "record"));
+                        handleProcessException(e, equ, failedResultSet);
                         failedCount++;
                     }
                 } catch (Exception e) {
@@ -2789,25 +2788,49 @@ public class DataDockServiceImpl {
             }
 
             log.info("已处理数据 {}/{} - 成功: {}, 失败: {}",
-                    Math.min(i + batchSize, totalSize), totalSize, successCount, failedCount);
+                    Math.min(i + batchSize, totalSize), totalSize, successfulCount, failedCount);
         }
 
         // 发布成功记录的事件
-        if (!recordSet.isEmpty()) {
-            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
-                @Override
-                public void afterCommit() {
-                    eventPublisher.publish(new EquipCreateOrEditEvent(this, BusinessTypeEnum.JG_NEW_EQUIP.name(), recordSet, EquipCreateOrEditEvent.EquipType.equip));
+        if (!ValidationUtil.isEmpty(successfulRecordSet)) {
+            eventPublisher.publish(new EquipCreateOrEditEvent(this, BusinessTypeEnum.JG_NEW_EQUIP.name(), successfulRecordSet, EquipCreateOrEditEvent.EquipType.equip));
+        }
+        // 处理失败记录对应es
+        if (!ValidationUtil.isEmpty(failedRecordSet)) {
+            // 分离需要更新和需要删除的记录
+            List<ESEquipmentCategoryDto> updateList = new ArrayList<>();
+            List<ESEquipmentCategoryDto> deleteList = new ArrayList<>();
+
+            for (String record : failedRecordSet) {
+                ESEquipmentCategoryDto esEquipment = originalEsResultMap.get(record);
+                if (esEquipment != null) {
+                    updateList.add(esEquipment);
+                } else {
+                    // 创建仅包含ID的DTO对象用于删除
+                    ESEquipmentCategoryDto deleteDto = new ESEquipmentCategoryDto();
+                    deleteDto.setSEQUENCE_NBR(record); // 设置主键字段
+                    deleteList.add(deleteDto);
                 }
-            });
             }
+
+            // 批量更新存在的记录
+            if (!updateList.isEmpty()) {
+                esEquipmentCategoryDao.saveAll(updateList);
+            }
+
+            // 批量删除不存在的记录
+            if (!deleteList.isEmpty()) {
+                esEquipmentCategoryDao.deleteAll(deleteList);
+            }
+        }
+
         log.info("数据导入完成 - 总计: {}条, 成功: {}条, 失败: {}条",
-                totalSize, successCount, failedCount);
-        log.info("未更新设备record：{}", resultSet);
-        return resultSet;
+                totalSize, successfulCount, failedCount);
+        log.info("未更新设备record：{}", failedResultSet);
+        return failedResultSet;
     }
 
-    private void handleProcessException(Exception e, Map<String, Object> equ, Set<String> inUseRecordSet) {
+    private void handleProcessException(Exception e, Map<String, Object> equ, Set<String> failedResultSet) {
         try {
             String traceInfo = Arrays.stream(e.getStackTrace())
                     .filter(x -> x.getClassName().contains("com.yeejoin"))
@@ -2819,20 +2842,20 @@ public class DataDockServiceImpl {
                     .map(String::valueOf)
                     .orElse(UUID.randomUUID().toString());
 
-//            esEquipmentCategory.deleteById(record);
             ExcelImportErrorLogDto errorLogDto = JSON.parseObject(toJSONString(equ), ExcelImportErrorLogDto.class);
             errorLogDto.setErrorInfo(e.getMessage());
             errorLogDto.setTraceInfo(traceInfo);
             errorLogDto.setCreateTime(new Date());
             excelImportErrorLogDao.save(errorLogDto);
 
-            inUseRecordSet.add(JSONObject.toJSONString(errorLogDto));
+            failedResultSet.add(JSONObject.toJSONString(errorLogDto));
             log.error("处理数据时发生异常，record: {}", record, e);
         } catch (Exception handleException) {
             log.error("处理异常信息时发生错误", handleException);
         }
     }
 
+    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
     public Object saveElevatorDataInTransaction(Map<String, Object> equ, String dataSource, String remark) {
         Map<String, String> errorResult = Maps.newConcurrentMap();
         String record = Optional.ofNullable(equ.get("record")).map(String::valueOf).orElse(UUID.randomUUID().toString());
@@ -2869,7 +2892,6 @@ public class DataDockServiceImpl {
             IdxBizJgOtherInfo otherInfo = this.saveOtherInfo(equ, record, equList);
             this.saveInspectInfo(equ, record);
             this.saveTechParams(equ, record, equList);
-            this.saveEquInfoToEs(record, isCompleteXa);
             this.saveInstallInfo(equ);//安装信息
             this.historyEquUpdateMaintenanceInfo(equ);//维保信息
 
@@ -2886,6 +2908,7 @@ public class DataDockServiceImpl {
                 this.updateUseManagementStatus(useInfo);
                 this.saveNewEquipResumeInfo(useInfo);
             }
+            this.saveEquInfoToEs(record, isCompleteXa);
         } else if (!businessId.isEmpty()) {
             this.saveResumeInfo(equ);
             equ.put("record", equ.get("resumeSeq"));
@@ -2969,7 +2992,7 @@ public class DataDockServiceImpl {
         jgUseRegistrationServiceImpl.saveOrUpdate(jgUseRegistration);
 
         // 更新设备es（jg-all）信息
-        this.updateEsData2(equ, otherInfo);
+//        this.updateEsData2(equ, otherInfo);
 
         // 更新登记证信息
         this.updateRegistrationManagementInfo(equ, oldUseRegistrationCode);
@@ -3258,7 +3281,7 @@ public class DataDockServiceImpl {
             map.put("equCode", equCode);
 
             // 更新es
-            this.updateEsData2(map, otherInfo);
+//            this.updateEsData2(map, otherInfo);
             jgUseRegistrationServiceImpl.save(jgUseRegistration);
             jgRelationEquip.setEquipTransferId(jgUseRegistration.getSequenceNbr() + "");
             jgRelationEquipMapper.insert(jgRelationEquip);

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

@@ -30,6 +30,8 @@ import com.yeejoin.amos.boot.module.common.api.dto.ESEquipmentCategoryDto;
 import com.yeejoin.amos.boot.module.common.api.entity.*;
 import com.yeejoin.amos.boot.module.common.api.enums.CylinderTypeEnum;
 import com.yeejoin.amos.boot.module.common.biz.refresh.cm.RefreshCmService;
+import com.yeejoin.amos.boot.module.common.biz.service.impl.ESEquipmentInfoService;
+import com.yeejoin.amos.boot.module.common.biz.service.impl.EquipmentCategoryService;
 import com.yeejoin.amos.boot.module.common.biz.service.impl.EsSearchServiceImpl;
 import com.yeejoin.amos.boot.module.common.biz.utils.RefreshDataUtils;
 import com.yeejoin.amos.boot.module.jg.api.common.BizCommonConstant;
@@ -60,15 +62,21 @@ import com.yeejoin.amos.boot.module.ymt.api.enums.EquipmentEnum;
 import com.yeejoin.amos.boot.module.ymt.api.enums.FlowStatusEnum;
 import com.yeejoin.amos.boot.module.ymt.api.mapper.*;
 import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.lucene.queryparser.classic.QueryParser;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestHighLevelClient;
 import org.elasticsearch.client.core.CountRequest;
 import org.elasticsearch.client.core.CountResponse;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.query.BoolQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.search.SearchHit;
@@ -98,6 +106,7 @@ import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
 
 import static com.alibaba.fastjson.JSON.toJSONString;
 import static com.yeejoin.amos.boot.module.jg.biz.service.impl.JgInstallationNoticeServiceImpl.CONSTRUCTION_TYPE;
@@ -219,6 +228,12 @@ public class DataHandlerServiceImpl {
 
     private final IdxBizJgInspectionDetectionInfoMapper inspectionDetectionInfoMapper;
 
+    private final ESEquipmentInfoService esEquipmentInfoService;
+
+    private final EquipmentCategoryService equipmentCategoryService;
+
+
+
     /**
      * 安装告知压力管道历史数据修复-详情中的设备列表修改为汇总表格式
      *
@@ -2948,4 +2963,91 @@ public class DataHandlerServiceImpl {
         esEquipmentInfos = esEquipmentInfos.stream().filter(e -> StringUtils.isNotEmpty(e.getSEQUENCE_NBR())).collect(Collectors.toList());
         return esEquipmentInfos;
     }
+
+
+
+    @SneakyThrows
+    public boolean syncEquipShard(Integer shard, Integer slots, String batchNo) {
+        StopWatch watch = new StopWatch();
+        watch.start();
+        // 0.数据准备
+        // 查询数据库 批量
+        List<String> records = idxBizJgUseInfoMapper.selectRecordList(shard, slots);
+        List<List<String>> partitions = Lists.partition(records, 2000);
+        // 查询es-旧索引
+        List<String> existsOldEsEquipRecords = partitions.parallelStream().map(equipmentCategoryService::findSequenceNbrByIds).flatMap(List::stream).collect(Collectors.toList());
+        //  标记老索引es-缺少数据
+        List<String> oldEsMissingRecords = records.parallelStream().filter(r -> !existsOldEsEquipRecords.contains(r)).collect(Collectors.toList());
+        // 查询es-新索引
+        List<String> existsNewEsEquipRecords =  partitions.parallelStream().map(esEquipmentInfoService::findSequenceNbrByIds).flatMap(List::stream).collect(Collectors.toList());
+        // 标记新索引es-缺少数据
+        List<String> newEsMissingRecords = records.parallelStream().filter(r -> !existsNewEsEquipRecords.contains(r)).collect(Collectors.toList());
+        // 1.分批次保存,数据补充至es
+        Set<String> allMissingRecords = Stream.concat(oldEsMissingRecords.stream(), newEsMissingRecords.stream()).collect(Collectors.toSet());
+        List<List<String>> savePartitions = Lists.partition(new ArrayList<>(allMissingRecords), 2000);
+        savePartitions.parallelStream().forEach(rs->{
+            List<Map<String, Object>> details = idxBizJgUseInfoMapper.queryDetailBatch(rs);
+            save2OldEquipEs(details, oldEsMissingRecords, batchNo);
+            save2NewEquipEs(rs, details, newEsMissingRecords, batchNo);
+        });
+        // 2.更新本切片数据的es的版本号,用来清除es的数据标记用
+        Map<String, Object> updateFields = MapUtil.of("version", batchNo);
+        partitions.parallelStream().forEach(batch->{
+            updateEsVersion(batch, updateFields, IDX_BIZ_VIEW_JG_ALL);
+            updateEsVersion(batch, updateFields, IDX_BIZ_EQUIPMENT_INFO);
+        });
+        watch.stop();
+        log.info("处理耗时:{}", watch.getTotalTimeSeconds());
+        return true;
+    }
+
+    private void updateEsVersion(List<String> batch, Map<String, Object> updateFields, String index) {
+        BulkRequest bulkRequest = new BulkRequest();
+        // 构建批量请求
+        batch.stream()
+                .map(id -> new UpdateRequest(index, id)
+                        .doc(updateFields, XContentType.JSON))
+                .forEach(bulkRequest::add);
+        try {
+            BulkResponse response = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
+            // 更详细的错误日志记录
+            if (response.hasFailures()) {
+                Arrays.stream(response.getItems())
+                        .filter(BulkItemResponse::isFailed)
+                        .forEach(item -> log.error("Failed to update {}: {}",
+                                item.getId(),
+                                item.getFailureMessage()));
+            }
+        } catch (IOException e) {
+            log.error("批量更新版本号失败: batchSize={}, index={}", batch.size(), index, e);
+            throw new RuntimeException("ES批量更新失败", e);
+        }
+    }
+
+    private void save2NewEquipEs(List<String> rs, List<Map<String, Object>> details, List<String> waitSync2EquipmentInfoRecords, String batchNo) {
+        Map<String, Map<String, Object>> recordDetailMap = details.stream().collect(Collectors.toMap(e -> (String) e.get("SEQUENCE_NBR"), Function.identity(), (k1, k2) -> k2));
+        List<String> equipmentInfoSaveRecords = rs.stream().filter(waitSync2EquipmentInfoRecords::contains).collect(Collectors.toList());
+        List<ESEquipmentInfo> esEquipmentInfos = getEsEquipmentInfos(equipmentInfoSaveRecords, recordDetailMap);
+        esEquipmentInfos.forEach(e->e.setVersion(batchNo));
+        esBulkService.bulkUpsert(IDX_BIZ_EQUIPMENT_INFO, esEquipmentInfos.stream().map(e -> new EsEntity<>(e.getSEQUENCE_NBR(), e)).collect(Collectors.toList()));
+    }
+
+    private void save2OldEquipEs(List<Map<String, Object>> details, List<String> waitSync2JgAllRecords, String batchNo) {
+        List<Map<String, Object>> jgAllSaves = details.stream().filter(e-> waitSync2JgAllRecords.contains((String) e.get("SEQUENCE_NBR"))).collect(Collectors.toList());
+        List<ESEquipmentCategoryDto> esEquipmentInfos = jgAllSaves.parallelStream().map(data -> {
+            ESEquipmentCategoryDto esEquipmentInfo = null;
+            try {
+                esEquipmentInfo = new ESEquipmentCategoryDto();
+                StatisticsDataUpdateService.formatUseDate(data);
+                BeanUtil.copyProperties(data, esEquipmentInfo, true);
+                esEquipmentInfo.setVersion(batchNo);
+            } catch (Exception e) {
+                log.error("批次{}设备刷数据处理失败", record, e);
+            }
+            return esEquipmentInfo;
+        }).collect(Collectors.toList());
+        if(!esEquipmentInfos.isEmpty()){
+            esBulkService.bulkUpsert(IDX_BIZ_VIEW_JG_ALL, esEquipmentInfos.stream().map(e -> new EsEntity<>(e.getSEQUENCE_NBR(), e)).collect(Collectors.toList()));
+        }
+    }
 }

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/ElevatorDataTransactionService.java

-package com.yeejoin.amos.boot.module.jg.biz.service.impl;
-
-import io.seata.spring.annotation.GlobalTransactional;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.annotation.Lazy;
-import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
-
-import java.util.Map;
-
-@Service
-public class ElevatorDataTransactionService {
-
-    private final DataDockServiceImpl dataDockService;
-
-    @Autowired
-    public ElevatorDataTransactionService(@Lazy DataDockServiceImpl dataDockService) {
-        this.dataDockService = dataDockService;
-    }
-
-    @Transactional(rollbackFor = Exception.class)
-    public Object saveSingleElevatorData(Map<String, Object> equ) {
-        return dataDockService.saveElevatorDataInTransaction(equ, "jg_his_xa", null);
-    }
-}

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/IdxBizJgRegisterInfoServiceImpl.java

@@ -34,7 +34,6 @@ import com.yeejoin.amos.boot.module.common.api.dto.ESEquipmentCategoryDto;
 import com.yeejoin.amos.boot.module.common.api.entity.ESEquipmentInfo;
 import com.yeejoin.amos.boot.module.common.api.enums.CylinderTypeEnum;
 import com.yeejoin.amos.boot.module.common.api.mapper.CustomBaseMapper;
-import com.yeejoin.amos.boot.module.common.biz.refresh.DataRefreshEvent;
 import com.yeejoin.amos.boot.module.common.biz.service.impl.EquipmentCategoryService;
 import com.yeejoin.amos.boot.module.jg.api.common.DataDockTemplateVersionUtils;
 import com.yeejoin.amos.boot.module.jg.api.dto.*;
@@ -56,7 +55,6 @@ import com.yeejoin.amos.boot.module.ymt.api.dto.TzsUserInfoDto;
 import com.yeejoin.amos.boot.module.ymt.api.entity.*;
 import com.yeejoin.amos.boot.module.ymt.api.enums.*;
 import com.yeejoin.amos.boot.module.ymt.api.mapper.*;
-import com.yeejoin.amos.component.feign.config.InnerInvokException;
 import com.yeejoin.amos.component.feign.model.FeignClientResult;
 import com.yeejoin.amos.component.feign.utils.FeignUtil;
 import com.yeejoin.amos.feign.systemctl.Systemctl;
@@ -88,8 +86,6 @@ import org.springframework.http.HttpStatus;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
-import org.springframework.transaction.support.TransactionSynchronization;
-import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.ObjectUtils;
 import org.springframework.web.multipart.MultipartFile;
 import org.typroject.tyboot.core.foundation.context.RequestContext;
@@ -122,7 +118,8 @@ import static com.yeejoin.amos.boot.module.common.api.enums.CylinderTypeEnum.SPE
 import static com.yeejoin.amos.boot.module.jg.api.enums.CertificateStatusEnum.YIDENGJI;
 import static com.yeejoin.amos.boot.module.jg.api.enums.VehicleApanageEnum.XIAN_YANG;
 import static com.yeejoin.amos.boot.module.jg.api.enums.VehicleApanageEnum.XI_XIAN;
-import static com.yeejoin.amos.boot.module.jg.biz.service.impl.DataHandlerServiceImpl.*;
+import static com.yeejoin.amos.boot.module.jg.biz.service.impl.DataHandlerServiceImpl.IDX_BIZ_EQUIPMENT_INFO;
+import static com.yeejoin.amos.boot.module.jg.biz.service.impl.DataHandlerServiceImpl.IDX_BIZ_VIEW_JG_ALL;
 
 
 /**
@@ -818,7 +815,8 @@ public class IdxBizJgRegisterInfoServiceImpl extends BaseService<IdxBizJgRegiste
             if (inspectionDetectionInfo.getNextInspectDate() != null) {
                 esEquipmentDto.setNEXT_INSPECT_DATE(inspectionDetectionInfo.getNextInspectDate().getTime());
             }
-            esEquipmentDto.setREC_DATE(System.currentTimeMillis());
+            esEquipmentDto.setCREATE_DATE(date.getTime());
+            esEquipmentDto.setREC_DATE(date.getTime());
             esEquipmentDto.setSEQUENCE_NBR(record);
             esEquipmentDto.setIS_INTO_MANAGEMENT(dataSource.contains("jg_his_black") && managedCategories.contains(equipmentInfoForm.get("EQU_CATEGORY")));
             esEquipmentDto.setEQU_CATEGORY_CODE(registerInfo.getEquCategory());
@@ -826,7 +824,7 @@ public class IdxBizJgRegisterInfoServiceImpl extends BaseService<IdxBizJgRegiste
             esEquipmentDto.setEQU_LIST_CODE(registerInfo.getEquList());
             esEquipmentDto.setEQU_LIST(equListName);
             esEquipmentDto.setEQU_DEFINE_CODE(registerInfo.getEquDefine());
-            esEquipmentDto.setEQU_DEFINE(equListName);
+            esEquipmentDto.setEQU_DEFINE(equDefineName);
             esEquipmentDto.setSTATUS("已认领");
             esEquipmentDto.setUSC_UNIT_CREDIT_CODE(projectContraption.getUscUnitCreditCode());
             esEquipmentDto.setUSC_UNIT_NAME(projectContraption.getUscUnitName());
@@ -835,9 +833,6 @@ public class IdxBizJgRegisterInfoServiceImpl extends BaseService<IdxBizJgRegiste
             esEquipmentDto.setPROJECT_CONTRAPTION(projectContraption.getProjectContraption());
             esEquipmentDto.setPRODUCT_NAME(pipelineInfo.getPipeName());
             esEquipmentDto.setProjectContraptionId(String.valueOf(sequenceNbr));
-            if (inspectionDetectionInfo.getNextInspectDate() != null) {
-                esEquipmentDto.setNEXT_INSPECT_DATE(inspectionDetectionInfo.getNextInspectDate().getTime());
-            }
             esEquipmentCategoryList.add(esEquipmentDto);
         }
         idxBizJgUseInfoService.saveOrUpdateBatch(useInfoList);
@@ -849,13 +844,14 @@ public class IdxBizJgRegisterInfoServiceImpl extends BaseService<IdxBizJgRegiste
         iIdxBizJgSupervisionInfoService.saveOrUpdateBatch(supervisionInfoList);
         iIdxBizJgInspectionDetectionInfoService.saveOrUpdateBatch(inspectionDetectionInfoList);
         iIdxBizJgTechParamsPipelineService.saveOrUpdateBatch(paramsPipelineList);
-        esEquipmentCategory.saveAll(esEquipmentCategoryList);
         // 更新管道长度
         updatePipelineLength(projectContraption, paramsPipelineList);
         if(OPERATESAVE.equals(operateType)){
             // 记录设备创建履历
             this.createResumePipeline(sequenceNbr, String.format(pipelineRoutePath, sequenceNbr + ""), company);
         }
+        // 最后更新es
+        esEquipmentCategory.saveAll(esEquipmentCategoryList);
         return sequenceNbr;
     }
 
@@ -4511,8 +4507,9 @@ public class IdxBizJgRegisterInfoServiceImpl extends BaseService<IdxBizJgRegiste
                 IdxBizJgUseInfo useInfo = new IdxBizJgUseInfo();
                 BeanUtils.copyProperties(data, useInfo);
                 useInfo.setRecord(record);
-                useInfo.setRecDate(new Date());
-                useInfo.setCreateDate(new Date());
+                Date now = new Date();
+                useInfo.setRecDate(now);
+                useInfo.setCreateDate(now);
                 String source = equipInfoDto.getDataSource();
                 useInfo.setDataSource("his".equals(source) ? "jg_his_pl" : "black".equals(source) ? "jg_his_black_pl" : "jg_pl");// 区分历史设备和新增设备
                 useInfo.setIsIntoManagement("his".equals(equipInfoDto.getDataSource()));// 历史气瓶导入为已纳管设备
@@ -4539,7 +4536,7 @@ public class IdxBizJgRegisterInfoServiceImpl extends BaseService<IdxBizJgRegiste
                 IdxBizJgDesignInfo designInfo = new IdxBizJgDesignInfo();
                 BeanUtils.copyProperties(data, designInfo);
                 designInfo.setRecord(record);
-                designInfo.setRecDate(new Date());
+                designInfo.setRecDate(now);
                 if (data.getDesignDate() != null) {
                     designInfo.setDesignDate(DateUtil.parse(data.getDesignDate(), "yyyy-MM-dd"));
                 }
@@ -4553,7 +4550,7 @@ public class IdxBizJgRegisterInfoServiceImpl extends BaseService<IdxBizJgRegiste
                 IdxBizJgFactoryInfo factoryInfo = new IdxBizJgFactoryInfo();
                 BeanUtils.copyProperties(data, factoryInfo);
                 factoryInfo.setRecord(record);
-                factoryInfo.setRecDate(new Date());
+                factoryInfo.setRecDate(now);
                 factoryInfo.setProductQualityYieldProve(productQualityYieldProve);
                 factoryInfo.setFactoryStandard(factoryStandard);
                 factoryInfo.setInsUseMaintainExplain(insUseMaintainExplain);
@@ -4572,7 +4569,7 @@ public class IdxBizJgRegisterInfoServiceImpl extends BaseService<IdxBizJgRegiste
                 BeanUtils.copyProperties(data, registerInfo);
                 registerInfo.setEquCode("2".equals(data.getEquCodeType()) ? "" : registerInfo.getEquCode());
                 registerInfo.setRecord(record);
-                registerInfo.setRecDate(new Date());
+                registerInfo.setRecDate(now);
                 registerInfo.setEquCategory(equipInfoDto.getEquCategoryCode());
                 registerInfo.setEquDefine(equipInfoDto.getEquDefineCode());
                 registerInfo.setEquList(equipInfoDto.getEquListCode());
@@ -4593,7 +4590,7 @@ public class IdxBizJgRegisterInfoServiceImpl extends BaseService<IdxBizJgRegiste
                 BeanUtils.copyProperties(data, inspectionDetectionInfo);
                 inspectionDetectionInfo.setInspectOrgCode(data.getInspectOrgCode());
                 inspectionDetectionInfo.setRecord(record);
-                inspectionDetectionInfo.setRecDate(new Date());
+                inspectionDetectionInfo.setRecDate(now);
                 inspectionDetectionInfo.setInspectType("ZZJDJY");
                 inspectionDetectionInfo.setInspectConclusion("6040");// 默认合格
                 Optional.ofNullable(data.getInspectDate())
@@ -4622,7 +4619,7 @@ public class IdxBizJgRegisterInfoServiceImpl extends BaseService<IdxBizJgRegiste
                 IdxBizJgOtherInfo otherInfo = new IdxBizJgOtherInfo();
                 BeanUtils.copyProperties(data, otherInfo);
                 otherInfo.setRecord(record);
-                otherInfo.setRecDate(new Date());
+                otherInfo.setRecDate(now);
                 otherInfo.setCylinderStampAttachment(cylinderStampAttachment);
                 if ("his".equals(equipInfoDto.getDataSource())) {
                     otherInfo.setSupervisoryCode(supervisoryCodeIterator.next());
@@ -4637,7 +4634,7 @@ public class IdxBizJgRegisterInfoServiceImpl extends BaseService<IdxBizJgRegiste
                 IdxBizJgTechParamsVessel paramsVessel = new IdxBizJgTechParamsVessel();
                 BeanUtils.copyProperties(data, paramsVessel);
                 paramsVessel.setRecord(record);
-                paramsVessel.setRecDate(new Date());
+                paramsVessel.setRecDate(now);
                 paramsVessel.setChargingMedium((String) fillingMediumMap.get(data.getChargingMedium()));
                 paramsVesselList.add(paramsVessel);
 
@@ -4646,8 +4643,8 @@ public class IdxBizJgRegisterInfoServiceImpl extends BaseService<IdxBizJgRegiste
                 if (inspectionDetectionInfo.getNextInspectDate() != null) {
                     esEquipmentDto.setNEXT_INSPECT_DATE(inspectionDetectionInfo.getNextInspectDate().getTime());
                 }
-                esEquipmentDto.setCREATE_DATE(System.currentTimeMillis());
-                esEquipmentDto.setREC_DATE(System.currentTimeMillis());
+                esEquipmentDto.setCREATE_DATE(now.getTime());
+                esEquipmentDto.setREC_DATE(now.getTime());
                 esEquipmentDto.setSEQUENCE_NBR(record);
                 esEquipmentDto.setFACTORY_NUM(factoryInfo.getFactoryNum());
                 esEquipmentDto.setUSE_INNER_CODE(useInfo.getUseInnerCode());

amos-boot-system-tzs/amos-boot-module-statistics/amos-boot-module-statistics-biz/src/main/java/com/yeejoin/amos/boot/module/statistcs/biz/service/impl/DPSubServiceImpl.java

@@ -884,7 +884,7 @@ public class DPSubServiceImpl {
             }
         }
         Object body = apiObj.get("body");
-        ResponseEntity<String> responseEntity = null;
+        ResponseEntity<String> sresponseEntity = null;
         //如果url以/开头，则调用本服务内接口
         if (url != null && url.trim().startsWith("/")) {
             url = "http://" + GATEWAY_SERVER_NAME + url;

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

@@ -58,4 +58,6 @@ public interface IdxBizJgUseInfoMapper extends CustomBaseMapper<IdxBizJgUseInfo>
     List<String> selectEquips(@Param("equListCode") String equListCode, @Param("equCategoryCode") String equCategoryCode, @Param("orgCode") String orgCode);
 
     List<String> selectUseInfoOfOneVersionWithParams(@Param("version") Integer version,@Param("params")  Map<String, Object> params);
+
+    List<String> selectRecordList(@Param("shard") Integer shard, @Param("slots") Integer slots);
 }

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

@@ -74,48 +74,49 @@
         WHERE ibjoi."RECORD" = #{record}
     </select>
     <select id="selectDataById" resultType="java.util.Map">
-        SELECT SEQUENCE_NBR,
-               REC_DATE,
-               CREATE_DATE,
-               DATA_QUALITY_SCORE,
-               ORG_BRANCH_NAME,
-               ORG_BRANCH_CODE,
-               USE_UNIT_NAME,
-               USE_UNIT_CREDIT_CODE,
-               EQU_LIST_CODE,
-               EQU_LIST,
-               EQU_CATEGORY,
-               EQU_CATEGORY_CODE,
-               USE_ORG_CODE,
-               CODE96333,
-               EQU_CODE,
-               SUPERVISORY_CODE,
-               USE_PLACE,
-               ADDRESS,
-               EQU_STATE,
-               STATUS,
-               USE_INNER_CODE,
-               FACTORY_NUM,
-               PRODUCE_UNIT_NAME,
-               INSPECT_REPORT,
-               NEXT_INSPECT_DATE,
-               CONSTRUCTION_TYPE,
-               USC_UNIT_CREDIT_CODE,
-               USC_UNIT_NAME,
-               EQU_DEFINE,
-               EQU_DEFINE_CODE,
-               PRODUCT_NAME,
-               BRAND_NAME,
-               EQU_TYPE,
-               DATA_SOURCE,
-               IS_INTO_MANAGEMENT,
-               WHETHER_VEHICLE_CYLINDER,
-               WHETHER_SKID_MOUNTED_PRESSURE_VESSEL,
-               DATE_FORMAT(PRODUCE_DATE, '%Y-%m-%d %H:%i:%s') as PRODUCE_DATE,
-               PROJECT_CONTRAPTION,
-               TAG1
+        SELECT "SEQUENCE_NBR",
+               "REC_DATE",
+               "CREATE_DATE",
+               "DATA_QUALITY_SCORE",
+               "ORG_BRANCH_NAME",
+               "ORG_BRANCH_CODE",
+               "USE_UNIT_NAME",
+               "USE_UNIT_CREDIT_CODE",
+               "EQU_LIST_CODE",
+               "EQU_LIST",
+               "EQU_CATEGORY",
+               "EQU_CATEGORY_CODE",
+               "USE_ORG_CODE",
+               "CODE96333",
+               "EQU_CODE",
+               "SUPERVISORY_CODE",
+               "USE_PLACE",
+               "ADDRESS",
+               "EQU_STATE",
+               "STATUS",
+               "USE_INNER_CODE",
+               "FACTORY_NUM",
+               "PRODUCE_UNIT_NAME",
+               "INSPECT_REPORT",
+               "NEXT_INSPECT_DATE",
+               "CONSTRUCTION_TYPE",
+               "USC_UNIT_CREDIT_CODE",
+               "USC_UNIT_NAME",
+               "EQU_DEFINE",
+               "EQU_DEFINE_CODE",
+               "PRODUCT_NAME",
+               "BRAND_NAME",
+               "EQU_TYPE",
+               "DATA_SOURCE",
+               "IS_INTO_MANAGEMENT",
+               "WHETHER_VEHICLE_CYLINDER",
+               "WHETHER_SKID_MOUNTED_PRESSURE_VESSEL",
+               DATE_FORMAT("PRODUCE_DATE", '%Y-%m-%d %H:%i:%s') as PRODUCE_DATE,
+               "PROJECT_CONTRAPTION",
+               TAG1,
+               "problemStatus"
         from idx_biz_view_jg_all
-        WHERE SEQUENCE_NBR = #{id}
+        WHERE "SEQUENCE_NBR" = #{id}
     </select>
     <select id="queryInitCode" resultType="com.yeejoin.amos.boot.module.ymt.api.entity.CategoryOtherInfo">
         SELECT CODE96333 code

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

@@ -137,53 +137,56 @@
              ibjui."RECORD" AS "SEQUENCE_NBR",
              ibjui."PROJECT_CONTRAPTION",
              ibjui."PROJECT_CONTRAPTION_ID",
-             ibjsi."ORG_BRANCH_NAME",
-             ibjsi."ORG_BRANCH_CODE",
              ibjui."REC_DATE",
              ibjui."USE_UNIT_NAME",
              ibjui."USE_UNIT_CREDIT_CODE",
-             ibjri."EQU_LIST" AS "EQU_LIST_CODE",
-             ibjri."EQU_DEFINE" AS "EQU_DEFINE_CODE",
-             ibjri."PRODUCT_NAME",
-             ibjri."BRAND_NAME",
-             ibjri."EQU_TYPE",
-             ibjri."EQU_CATEGORY" AS "EQU_CATEGORY_CODE",
-             tec2.NAME AS "EQU_DEFINE",
-             tec1.NAME AS "EQU_LIST",
-             tec.NAME AS "EQU_CATEGORY",
-             ibjri."USE_ORG_CODE",
              ibjui."USE_INNER_CODE",
              ibjui."DATA_SOURCE",
-             ibjoi."CODE96333",
+             ibjui."CREATE_DATE" ,
+             ibjui."SAFETY_MANAGER" ,
+             ibjui."PHONE",
+             ibjui."USE_DATE",
+             ibjui."USE_PLACE" as USE_SITE_CODE,
+             ibjui."DATA_QUALITY_SCORE",
+             ibjui."IS_NOT_ES",
+             concat_ws ( '/', ibjui."PROVINCE_NAME", ibjui."CITY_NAME", ibjui."COUNTY_NAME", ibjui."STREET_NAME" ) AS "USE_PLACE",
+             concat_ws ( '#', ibjui."PROVINCE", ibjui."CITY", ibjui."COUNTY", ibjui."FACTORY_USE_SITE_STREET" ) AS "USE_PLACE_CODE",
+             ibjui."ADDRESS",
+             ibjui."EQU_STATE",
+             ibjui."IS_INTO_MANAGEMENT",
              ibjri."EQU_CODE",
-             ibjoi."SUPERVISORY_CODE",
              ibjri."WHETHER_VEHICLE_CYLINDER",
              ibjri."WHETHER_SKID_MOUNTED_PRESSURE_VESSEL",
              ibjri."CYLINDER_CATEGORY",
              ibjri."WHETHER_SPHERICAL_TANK",
              ibjri."CAR_NUMBER",
              ibjri."EQU_PRICE",
-             concat_ws ( '/', ibjui."PROVINCE_NAME", ibjui."CITY_NAME", ibjui."COUNTY_NAME", ibjui."STREET_NAME" ) AS "USE_PLACE",
-             concat_ws ( '#', ibjui."PROVINCE", ibjui."CITY", ibjui."COUNTY", ibjui."FACTORY_USE_SITE_STREET" ) AS "USE_PLACE_CODE",
-             ibjui."ADDRESS",
-             ibjui."EQU_STATE",
-             ibjui."IS_INTO_MANAGEMENT",
-             ibjoi."CLAIM_STATUS" AS "STATUS",
-             ibjoi."STATUS" AS problemStatus,
+             ibjri."EQU_LIST" AS "EQU_LIST_CODE",
+             ibjri."EQU_DEFINE" AS "EQU_DEFINE_CODE",
+             ibjri."PRODUCT_NAME",
+             ibjri."BRAND_NAME",
+             ibjri."EQU_TYPE",
+             ibjri."EQU_CATEGORY" AS "EQU_CATEGORY_CODE",
+             ibjri."TAG1" AS tag1,
+             ibjri."USE_ORG_CODE",
+             ibjsi."ORG_BRANCH_NAME",
+             ibjsi."ORG_BRANCH_CODE",
              ibjfi."FACTORY_NUM",
              ibjfi."PRODUCE_UNIT_NAME",
              ibjfi."PRODUCE_UNIT_CREDIT_CODE",
              ibjfi."PRODUCE_DATE",
-             ibjui."CREATE_DATE" ,
-             ibjui."SAFETY_MANAGER" ,
-             ibjui."PHONE",
-             ibjui."USE_DATE",
-             ibjui."USE_PLACE" as USE_SITE_CODE,
-             ibjui."DATA_QUALITY_SCORE",
+             ibjoi."CODE96333",
+             ibjoi."SUPERVISORY_CODE",
+             ibjoi."CLAIM_STATUS" AS "STATUS",
+             ibjoi."STATUS" AS problemStatus,
              ibjoi."INFORMATION_SITUATION",
+             ibjoi."EDIT_STATUS",
              di."DESIGN_UNIT_CREDIT_CODE",
              di."DESIGN_UNIT_NAME",
              di."DESIGN_DATE",
+             tec2.NAME AS "EQU_DEFINE",
+             tec1.NAME AS "EQU_LIST",
+             tec.NAME AS "EQU_CATEGORY",
              construction_info."USC_UNIT_CREDIT_CODE",
              construction_info."USC_UNIT_NAME",
              construction_info."CONSTRUCTION_TYPE",
@@ -324,5 +327,13 @@
             AND r."EQU_CATEGORY" = #{equCategoryCode}
         </if>
     </select>
+    <select id="selectRecordList" resultType="java.lang.String">
+        SELECT
+            "RECORD"
+        FROM
+            idx_biz_jg_use_info
+        WHERE
+            ABS(MOD(hashtext("RECORD"), #{slots})) = #{shard}
+    </select>
 
 </mapper>