调整influxdb插入条数20000，增加线程池

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataFlickerImpl.java

@@ -53,7 +53,7 @@ public class InsertDataFlickerImpl implements InsertIDataFlicker {
     @Override
     public void insertDataFlicker(List<DataFlicker> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataFlicker>> partition = ListUtils.partition(list, 100000);
+            List<List<DataFlicker>> partition = ListUtils.partition(list, 20000);
             for (List<DataFlicker> dataFlickerList : partition) {
                 for (DataFlicker data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataFlucImpl.java

@@ -52,7 +52,7 @@ public class InsertDataFlucImpl implements InsertIDataFluc {
     @Override
     public void insertDataFluc(List<DataFluc> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataFluc>> partition = ListUtils.partition(list, 100000);
+            List<List<DataFluc>> partition = ListUtils.partition(list, 20000);
             for (List<DataFluc> dataFlickerList : partition) {
                 for (DataFluc data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataHarmRateIImpl.java

@@ -49,7 +49,7 @@ public class InsertDataHarmRateIImpl  implements InsertIDataHarmRateI {
     @Override
     public void insertDataHarmrateI(List<DataHarmrateI> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataHarmrateI>> partition = ListUtils.partition(list, 100000);
+            List<List<DataHarmrateI>> partition = ListUtils.partition(list, 20000);
             for (List<DataHarmrateI> dataFlickerList : partition) {
                 for (DataHarmrateI data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataHarmRateVImpl.java

@@ -52,7 +52,7 @@ public class InsertDataHarmRateVImpl  implements InsertIDataHarmRateV {
     @Override
     public void insertDataHarmrateV(List<DataHarmrateV> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataHarmrateV>> partition = ListUtils.partition(list, 100000);
+            List<List<DataHarmrateV>> partition = ListUtils.partition(list, 20000);
             for (List<DataHarmrateV> dataFlickerList : partition) {
                 for (DataHarmrateV data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataHarmphasicIImpl.java

@@ -55,7 +55,7 @@ public class InsertDataHarmphasicIImpl implements InsertIDataHarmphasicI {
     @Override
     public void insertDataHarmphasicI(List<DataHarmphasicI> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataHarmphasicI>> partition = ListUtils.partition(list, 100000);
+            List<List<DataHarmphasicI>> partition = ListUtils.partition(list, 20000);
             for (List<DataHarmphasicI> dataFlickerList : partition) {
                 for (DataHarmphasicI data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataHarmphasicVImpl.java

@@ -54,7 +54,7 @@ public class InsertDataHarmphasicVImpl  implements InsertIDataHarmphasicV {
     @Override
     public void insertDataHarmphasicV(List<DataHarmphasicV> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataHarmphasicV>> partition = ListUtils.partition(list, 100000);
+            List<List<DataHarmphasicV>> partition = ListUtils.partition(list, 20000);
             for (List<DataHarmphasicV> dataFlickerList : partition) {
                 for (DataHarmphasicV data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataHarmpowerPImpl.java

@@ -57,7 +57,7 @@ public class InsertDataHarmpowerPImpl implements InsertIDataHarmpowerP {
     @Override
     public void insertDataHarmpowerP(List<DataHarmpowerP> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataHarmpowerP>> partition = ListUtils.partition(list, 100000);
+            List<List<DataHarmpowerP>> partition = ListUtils.partition(list, 20000);
             for (List<DataHarmpowerP> dataFlickerList : partition) {
                 for (DataHarmpowerP data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataHarmpowerQImpl.java

@@ -54,7 +54,7 @@ public class InsertDataHarmpowerQImpl  implements InsertIDataHarmpowerQ {
     @Override
     public void insertDataHarmpowerQ(List<DataHarmpowerQ> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataHarmpowerQ>> partition = ListUtils.partition(list, 100000);
+            List<List<DataHarmpowerQ>> partition = ListUtils.partition(list, 20000);
             for (List<DataHarmpowerQ> dataFlickerList : partition) {
                 for (DataHarmpowerQ data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataHarmpowerSImpl.java

@@ -54,7 +54,7 @@ public class InsertDataHarmpowerSImpl implements InsertIDataHarmpowerS {
     @Override
     public void insertDataHarmpowerS(List<DataHarmpowerS> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataHarmpowerS>> partition = ListUtils.partition(list, 100000);
+            List<List<DataHarmpowerS>> partition = ListUtils.partition(list, 20000);
             for (List<DataHarmpowerS> dataFlickerList : partition) {
                 for (DataHarmpowerS data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataIImpl.java

@@ -63,7 +63,7 @@ public class InsertDataIImpl  implements InsertIDataI {
     @Override
     public void insertDataI(List<DataI> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataI>> partition = ListUtils.partition(list, 100000);
+            List<List<DataI>> partition = ListUtils.partition(list, 20000);
             for (List<DataI> dataFlickerList : partition) {
                 for (DataI data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataInharmIImpl.java

@@ -53,7 +53,7 @@ public class InsertDataInharmIImpl  implements InsertIDataInharmI {
     @Override
     public void insertDataInharmI(List<DataInharmI> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataInharmI>> partition = ListUtils.partition(list, 100000);
+            List<List<DataInharmI>> partition = ListUtils.partition(list, 20000);
             for (List<DataInharmI> dataFlickerList : partition) {
                 for (DataInharmI data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataInharmVImpl.java

@@ -56,7 +56,7 @@ public class InsertDataInharmVImpl implements InsertIDataInharmV {
     @Override
     public void insertDataInharmV(List<DataInharmV> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataInharmV>> partition = ListUtils.partition(list, 100000);
+            List<List<DataInharmV>> partition = ListUtils.partition(list, 20000);
             for (List<DataInharmV> dataFlickerList : partition) {
                 for (DataInharmV data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataPltImpl.java

@@ -50,7 +50,7 @@ public class InsertDataPltImpl implements InsertIDataPlt {
     @Override
     public void insertDataPlt(List<DataPlt> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataPlt>> partition = ListUtils.partition(list, 100000);
+            List<List<DataPlt>> partition = ListUtils.partition(list, 20000);
             for (List<DataPlt> dataFlickerList : partition) {
                 for (DataPlt data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertDataVImpl.java

@@ -98,7 +98,7 @@ public class InsertDataVImpl implements InsertIDataV {
     @Override
     public void insertDataV(List<DataV> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<DataV>> partition = ListUtils.partition(list, 100000);
+            List<List<DataV>> partition = ListUtils.partition(list, 20000);
             for (List<DataV> dataFlickerList : partition) {
                 for (DataV dataV : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(dataV.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-insert-boot/src/main/java/com/njcn/migration/insert/service/impl/InsertPqsCommunicateImpl.java

@@ -48,7 +48,7 @@ public class InsertPqsCommunicateImpl implements InsertIPqsCommunicate {
     @Override
     public void insertPqsCommunicate(List<PqsCommunicate> list) {
         if(CollUtil.isNotEmpty(list)){
-            List<List<PqsCommunicate>> partition = ListUtils.partition(list, 100000);
+            List<List<PqsCommunicate>> partition = ListUtils.partition(list, 20000);
             for (List<PqsCommunicate> dataFlickerList : partition) {
                 for (PqsCommunicate data : dataFlickerList) {
                     Instant instant = LocalDateTimeUtil.parse(data.getTimeId(), DatePattern.NORM_DATETIME_PATTERN) .atZone(ZoneId.systemDefault()).toInstant();

migration-influxdb/migration-influxdb-read-boot/src/main/java/com/njcn/migration/read/MigrationReadBootApplication.java

@@ -7,6 +7,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.openfeign.EnableFeignClients;
 import org.springframework.context.annotation.DependsOn;
 import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 
 /**
@@ -18,6 +19,7 @@ import org.springframework.scheduling.annotation.EnableAsync;
 @MapperScan("com.njcn.**.mapper")
 @EnableFeignClients(basePackages = "com.njcn")
 @SpringBootApplication(scanBasePackages = "com.njcn")
+@EnableScheduling
 public class MigrationReadBootApplication {
 
     public static void main(String[] args)  {

migration-influxdb/migration-influxdb-read-boot/src/main/java/com/njcn/migration/read/config/AsyncInfluxDBConfiguration.java

@@ -0,0 +1,48 @@
+package com.njcn.migration.read.config;
+
+import com.njcn.common.config.GeneralInfo;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.annotation.Order;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ * @author hongawen
+ * @version 1.0.0
+ * @date 2022年03月11日 09:32
+ */
+@Data
+@Order(100)
+@Configuration
+@EnableAsync
+@AllArgsConstructor
+public class AsyncInfluxDBConfiguration {
+
+    private final GeneralInfo generalInfo;
+
+    @Bean("asyncInfluxDBExecutor")
+    public Executor asyncExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        // 核心线程数：线程池创建时候初始化的线程数
+        executor.setCorePoolSize(generalInfo.getCorePoolSize());
+        // 最大线程数：线程池最大的线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程
+        executor.setMaxPoolSize(generalInfo.getMaxPoolSize());
+        // 缓冲队列：用来缓冲执行任务的队列
+        executor.setQueueCapacity(generalInfo.getQueueCapacity());
+        // 允许线程的空闲时间60秒：当超过了核心线程之外的线程在空闲时间到达之后会被销毁
+        executor.setKeepAliveSeconds(generalInfo.getKeepAliveSeconds());
+        // 线程池名的前缀：设置好了之后可以方便我们定位处理任务所在的线程池
+        executor.setThreadNamePrefix(generalInfo.getMicroServiceName());
+        // 缓冲队列满了之后的拒绝策略：由调用线程处理（一般是主线程）
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+        executor.initialize();
+        return executor;
+    }
+
+}

migration-influxdb/migration-influxdb-read-boot/src/main/java/com/njcn/migration/read/service/impl/MigrationServiceImpl.java

@@ -52,7 +52,7 @@ public class MigrationServiceImpl implements MigrationService {
 
 
     @Override
-    @Async
+    @Async("asyncInfluxDBExecutor")
     public void hourseLineDataBacthSysc(LineCountEvaluateParam param) {
         Map<String, String> map = TimeUtil.getLineMap();
         int size = map.size();
@@ -97,7 +97,7 @@ public class MigrationServiceImpl implements MigrationService {
     }
 
     @Override
-    @Async
+    @Async("asyncInfluxDBExecutor")
     public void hourseDevDataBacthSysc(LineCountEvaluateParam param) {
         Map<String, String> map = TimeUtil.getDevMap();
         int size = map.size();
@@ -127,6 +127,7 @@ public class MigrationServiceImpl implements MigrationService {
     }
 
     @Override
+    @Async("asyncInfluxDBExecutor")
     public void initializeExcel() throws IOException {
         File file = new File("/usr/local/jar/sj.xlsx");
         Map<String, String> map = TimeUtil.getLineMap();

migration-influxdb/migration-influxdb-read-boot/src/main/resources/bootstrap.yml

@@ -58,6 +58,11 @@ mybatis-plus:
     #配置sql日志输出
     log-impl: org.apache.ibatis.logging.nologging.NoLoggingImpl
 
+threadPool:
+  corePoolSize: 12
+  maxPoolSize: 24
+  queueCapacity: 500
+  keepAliveSeconds: 60
 
 #  type-aliases-package: com.njcn.harmonic.pojo
 #  type-handlers-package: com.njcn.db.handler