fix(JG)：批量导入模版调整

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/config/PressureVesselListener.java

@@ -28,6 +28,10 @@ import java.time.LocalDate;
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicLong;
 import static com.alibaba.fastjson.JSON.toJSONString;
 
 @Slf4j
@@ -35,7 +39,7 @@ import static com.alibaba.fastjson.JSON.toJSONString;
 @EqualsAndHashCode(callSuper = true)
 @Component
 public class PressureVesselListener extends AnalysisEventListener<EquipInfoCylinderExcelDto> {
-    private static final int BATCH_COUNT = 1000;
+    private static final int BATCH_COUNT = 200;
     //数据集合
     List<String> useInnerCodeList = new ArrayList<>();//单位内部编号集合
     List<String> equCodeList = new ArrayList<>();//设备代码集合
@@ -66,6 +70,18 @@ public class PressureVesselListener extends AnalysisEventListener<EquipInfoCylin
     private Map<String, Object> paramMap;
     private CompanyBo company;
 
+    final private static AtomicLong sendThreadPoolCounter = new AtomicLong(0);
+    final public static ExecutorService pooledExecutor =
+            Executors.newFixedThreadPool(20 + Runtime.getRuntime().availableProcessors(),
+                    createThreadFactory());
+    private static ThreadFactory createThreadFactory() {
+        return runnable -> {
+            Thread thread = new Thread(runnable);
+            thread.setName(String.format("kafka-consumer-iot-pool-%d", PressureVesselListener.sendThreadPoolCounter.getAndIncrement()));
+            return thread;
+        };
+    }
+
     @Override
     public void invoke(EquipInfoCylinderExcelDto data, AnalysisContext context) {
         ReadRowHolder readRowHolder = context.readRowHolder();
@@ -99,6 +115,8 @@ public class PressureVesselListener extends AnalysisEventListener<EquipInfoCylin
                     result.append("设备代码不能重复；");
                 }
                 this.checkEquCodeUniqueness(data.getEquCode());
+            } else {
+                data.setEquCode("");
             }
             checkNotBlank(data.getDesignUnitCreditCode(), "设计单位统一社会信用代码不能为空");
             checkNotBlank(data.getDesignUnitName(), "设计单位名称不能为空");
@@ -297,32 +315,20 @@ public class PressureVesselListener extends AnalysisEventListener<EquipInfoCylin
     @Override
     public void doAfterAllAnalysed(AnalysisContext context) {
         if (paramsVesselList.size() > BATCH_COUNT) {
-            throw new BadRequest(result.append("单次导入限制1000条") + "");
-        }
-        if (!useInfoList.isEmpty()) {
-            idxBizJgUseInfoService.saveBatch(useInfoList);
-        }
-        if (!designInfoList.isEmpty()) {
-            idxBizJgDesignInfoService.saveBatch(designInfoList);
-        }
-        if (!registerInfoList.isEmpty()) {
-            idxBizJgRegisterInfoService.saveBatch(registerInfoList);
-        }
-        if (!factoryInfoList.isEmpty()) {
-            idxBizJgFactoryInfoService.saveBatch(factoryInfoList);
-        }
-        if (!otherInfoList.isEmpty()) {
-            idxBizJgOtherInfoService.saveBatch(otherInfoList);
-        }
-        if (!paramsVesselList.isEmpty()) {
-            idxBizJgTechParamsVesselService.saveBatch(paramsVesselList);
-        }
-        if (!inspectionDetectionInfoList.isEmpty()) {
-            idxBizJgInspectionDetectionInfoService.saveBatch(inspectionDetectionInfoList);
-        }
-        if (!esEquipmentCategoryList.isEmpty()) {
-            esEquipmentCategory.saveAll(esEquipmentCategoryList);
+            throw new BadRequest(result.append("单次导入限制200条") + "");
         }
+
+       pooledExecutor.execute(() -> {
+            Optional.of(useInfoList).filter(list -> !list.isEmpty()).ifPresent(idxBizJgUseInfoService::saveBatch);
+            Optional.of(designInfoList).filter(list -> !list.isEmpty()).ifPresent(idxBizJgDesignInfoService::saveBatch);
+            Optional.of(registerInfoList).filter(list -> !list.isEmpty()).ifPresent(idxBizJgRegisterInfoService::saveBatch);
+            Optional.of(factoryInfoList).filter(list -> !list.isEmpty()).ifPresent(idxBizJgFactoryInfoService::saveBatch);
+            Optional.of(otherInfoList).filter(list -> !list.isEmpty()).ifPresent(idxBizJgOtherInfoService::saveBatch);
+            Optional.of(paramsVesselList).filter(list -> !list.isEmpty()).ifPresent(idxBizJgTechParamsVesselService::saveBatch);
+            Optional.of(inspectionDetectionInfoList).filter(list -> !list.isEmpty()).ifPresent(idxBizJgInspectionDetectionInfoService::saveBatch);
+            Optional.of(esEquipmentCategoryList).filter(list -> !list.isEmpty()).ifPresent(esEquipmentCategory::saveAll);
+        });
+
         result.append("导入完成，成功导入");
         result.append(useInfoList.size());
         result.append("条数据");