实现中环scada历史数据迁移

custom-gateway-app/pom.xml

@@ -75,7 +75,6 @@
                 <configuration>
                     <skipTests>true</skipTests>
                     <systemPropertyVariables>
-                        <spring.datasource.config-path>../db_config.yml</spring.datasource.config-path>
                         <device.map-path>../dev_${enable.module}.yml</device.map-path>
                         <spring.datasource.multi.embedded.url>jdbc:h2:file:../data
                         </spring.datasource.multi.embedded.url>

custom-gateway-app/src/main/resources/application-zhscada.yml

@@ -1,6 +1,33 @@
+#spring
+spring:
+  #数据源
+  datasource:
+    #多数据源
+    multi:
+      #scada
+      scada:
+        url: jdbc:oracle:thin:@//140.246.183.164:1521/FREEPDB1
+        username: zh_scada
+        password: kpr.65231.zh_scada
+        driver-class-name: oracle.jdbc.OracleDriver
+  #influx
+  influx:
+    read-timeout: 30000
+    clients:
+      #中环
+      - url: http://111.170.129.117:8086/
+        user: kpr
+        password: kpr.2025@117.influxdb
+        database: iot
 #对接
 calling:
   #对接点
   endpoints:
     zhonghuan-scada:
       url: http://119.96.174.191:9434
+#迁移
+migrate:
+  #表名
+  schema: zh_scada
+  #传感器映射文件
+  sensor-map-path: classpath*:sensor-id.yml

custom-gateway-app/src/main/resources/application-zydma.yml

@@ -1,3 +1,19 @@
+#spring
+spring:
+  #influx
+  influx:
+    read-timeout: 30000
+    clients:
+      #枣阳
+      - url: http://192.168.44.58:8086/
+        user: kpr
+        password: kpr.2025@117.influxdb
+        database: iot
+      #测试
+      - url: http://119.96.165.176:8086/
+        user: kpr
+        password: kpr.2024@117.influxdb
+        database: iot
 #对接
 calling:
   #对接点
@@ -10,6 +26,7 @@ calling:
       url: http://223.75.194.87:8200/pdserver
       access-key: Data
       secret-key: panda666.
+#认证
 security:
   permit-pattern:
     - ${gateway.routes.zydma}/users/ticket-exchange

custom-gateway-app/src/main/resources/application.yml

@@ -41,7 +41,7 @@ spring:
         size: 2
   #数据源
   datasource:
-    config-path: ./db_config.yml
+    #连接池
     hikari:
       maximum-pool-size: 200
       minimum-idle: 1
@@ -51,6 +51,20 @@ spring:
       connection-timeout: 30000
       validation-timeout: 30000
       initialization-fail-timeout: -1
+    #多数据源
+    multi:
+      #主要
+      primary:
+        url: jdbc:postgresql://223.75.194.87:6057/watersmart?useSSL=false&useAffectedRows=false&allowMultiQueries=true&rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai
+        username: postgres
+        password: kpr.23417.postgres
+        driver-class-name: org.postgresql.Driver
+      #嵌入式
+      embedded:
+        url: jdbc:h2:file:./data
+        driver-class-name: org.h2.Driver
+        username: sa
+        password: kpr.25535.sa
 #mybatis
 mybatis:
   configuration:

custom-gateway-app/src/main/resources/zhscada-backup/application-zhscada.yml

@@ -0,0 +1,33 @@
+#spring
+spring:
+  #数据源
+  datasource:
+    #多数据源
+    multi:
+      #scada
+      scada:
+        url: jdbc:oracle:thin:@//119.96.190.182:1521/orcl
+        username: scada
+        password: scada
+        driver-class-name: oracle.jdbc.OracleDriver
+  #influx
+  influx:
+    read-timeout: 30000
+    clients:
+      #中环
+      - url: http://111.170.129.117:8086/
+        user: kpr
+        password: kpr.2025@117.influxdb
+        database: iot
+#对接
+calling:
+  #对接点
+  endpoints:
+    zhonghuan-scada:
+      url: http://119.96.174.191:9434
+#迁移
+migrate:
+  #表名
+  table-name: scada
+  #传感器映射文件
+  sensor-map-path: file:E:\kpr\kpr-custom-gateway\sensor-id.yml

custom-gateway-app/src/main/resources/zhscada-backup/application-zydma.yml

@@ -0,0 +1,32 @@
+#spring
+spring:
+  #influx
+  influx:
+    read-timeout: 30000
+    clients:
+      #枣阳
+      - url: http://192.168.44.58:8086/
+        user: kpr
+        password: kpr.2025@117.influxdb
+        database: iot
+      #测试
+      - url: http://119.96.165.176:8086/
+        user: kpr
+        password: kpr.2024@117.influxdb
+        database: iot
+#对接
+calling:
+  #对接点
+  endpoints:
+    zaoyang-middle:
+      url: http://223.75.194.87:8200
+      access-key: lousunkongzhi
+      secret-key: g+4UWJ6360SxDVu+9BRRQfOg0/tT+33o3S8Q5APMLIn+JQirprtdGd0cf5Y3WO7iiKo24T5mN1U697zHp/iGNA==
+    zaoyang-iot:
+      url: http://223.75.194.87:8200/pdserver
+      access-key: Data
+      secret-key: panda666.
+#认证
+security:
+  permit-pattern:
+    - ${gateway.routes.zydma}/users/ticket-exchange

custom-gateway-app/src/main/resources/zhscada-backup/application.yml

@@ -0,0 +1,96 @@
+#web
+server:
+  port: 9011
+  servlet:
+    context-path: /custom-gw/
+#网关
+gateway:
+  routes:
+    zydma: /zy-dma
+#spring
+spring:
+  #app
+  application:
+    name: KprCustomGateway
+  #servlet
+  servlet:
+    multipart:
+      max-file-size: 200MB
+      max-request-size: 200MB
+  #启用配置
+  profiles:
+    active: @enable.module@
+  #thymeleaf
+  thymeleaf:
+    prefix: classpath*:/templates/
+  #任务
+  task:
+    #任务线程池
+    execution:
+      thread-name-prefix: task-
+      pool:
+        core-size: 8
+        max-size: 0x7fffffff
+        queue-capacity: 0x7fffffff
+        keep-alive: PT30S
+        allow-core-thread-timeout: true
+    #定时线程池
+    scheduling:
+      thread-name-prefix: scheduling-
+      pool:
+        size: 2
+  #数据源
+  datasource:
+    #连接池
+    hikari:
+      maximum-pool-size: 200
+      minimum-idle: 1
+      connection-test-query: SELECT 1
+      max-lifetime: 120000
+      idle-timeout: 30000
+      connection-timeout: 30000
+      validation-timeout: 30000
+      initialization-fail-timeout: -1
+    #多数据源
+    multi:
+      #主要
+      primary:
+        url: jdbc:postgresql://223.75.194.87:6057/watersmart?useSSL=false&useAffectedRows=false&allowMultiQueries=true&rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai
+        username: postgres
+        password: kpr.23417.postgres
+        driver-class-name: org.postgresql.Driver
+      #嵌入式
+      embedded:
+        url: jdbc:h2:file:./data
+#        url: jdbc:h2:file:E:\kpr\kpr-custom-gateway\data
+        driver-class-name: org.h2.Driver
+        username: sa
+        password: kpr.25535.sa
+#mybatis
+mybatis:
+  configuration:
+    call-setters-on-nulls: true
+    map-underscore-to-camel-case: true
+    cache-enabled: false
+#security
+security:
+  expiration: P7D
+  secret: TRICP_ALAM_DMA
+#对接
+calling:
+  connect-timeout: PT30S
+  read-timeout: PT30S
+#异步
+async:
+  result-path-pattern: /common/async-results
+  result-resource-location: ./async-results/
+#临时文件
+temp-file:
+  path-pattern: /common/temp-files
+  resource-location: ./temp-files/
+  cleanup-interval: PT1M
+  max-age: PT4H
+#设备
+device:
+  map-path: ./dev_@enable.module@.yml
+  #map-path: E:\kpr\kpr-custom-gateway\dev_@enable.module@.yml

custom-gateway-core/pom.xml

@@ -47,6 +47,15 @@
             <groupId>org.postgresql</groupId>
             <artifactId>postgresql</artifactId>
         </dependency>
+        <!--oracle-->
+        <dependency>
+            <groupId>com.oracle.database.jdbc</groupId>
+            <artifactId>ojdbc8</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.oracle.database.nls</groupId>
+            <artifactId>orai18n</artifactId>
+        </dependency>
         <!--influxdb-->
         <dependency>
             <groupId>org.influxdb</groupId>

custom-gateway-core/src/main/java/com/shkpr/service/customgateway/core/config/DataSourceConfig.java

@@ -1,6 +1,5 @@
 package com.shkpr.service.customgateway.core.config;
 
-import com.shkpr.service.customgateway.core.io.YamlPropertySourceFactory;
 import com.shkpr.service.customgateway.core.properties.DataSourceProperties;
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
@@ -9,7 +8,6 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.boot.jdbc.DataSourceBuilder;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.PropertySource;
 
 import javax.sql.DataSource;
 
@@ -24,7 +22,6 @@ import static com.shkpr.service.customgateway.core.constants.DataSourceNames.PRI
  */
 @Configuration
 @EnableConfigurationProperties(DataSourceProperties.class)
-@PropertySource(value = "file:${spring.datasource.config-path}", ignoreResourceNotFound = true, encoding = "UTF-8", factory = YamlPropertySourceFactory.class)
 public class DataSourceConfig {
     /**
      * @param properties 属性
@@ -40,6 +37,7 @@ public class DataSourceConfig {
                 .url(source.getUrl())
                 .username(source.getUsername())
                 .password(source.getPassword())
+                .driverClassName(source.getDriverClassName())
                 .build();
 
         configHikari(properties.getHikari(), dataSource);
@@ -61,6 +59,7 @@ public class DataSourceConfig {
                 .url(source.getUrl())
                 .username(source.getUsername())
                 .password(source.getPassword())
+                .driverClassName(source.getDriverClassName())
                 .build();
     }
 

custom-gateway-core/src/main/java/com/shkpr/service/customgateway/core/config/InfluxDbConfig.java

@@ -2,7 +2,6 @@ package com.shkpr.service.customgateway.core.config;
 
 import com.shkpr.service.customgateway.core.components.InfluxDBClients;
 import com.shkpr.service.customgateway.core.domain.InfluxDbClient;
-import com.shkpr.service.customgateway.core.io.YamlPropertySourceFactory;
 import com.shkpr.service.customgateway.core.properties.InfluxDbProperties;
 import okhttp3.OkHttpClient;
 import org.springframework.beans.factory.ObjectProvider;
@@ -10,7 +9,6 @@ import org.springframework.boot.autoconfigure.influx.InfluxDbOkHttpClientBuilder
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.PropertySource;
 
 import java.util.List;
 import java.util.stream.Collectors;
@@ -23,7 +21,6 @@ import java.util.stream.Collectors;
  */
 @Configuration
 @EnableConfigurationProperties(InfluxDbProperties.class)
-@PropertySource(value = "file:${spring.datasource.config-path}", ignoreResourceNotFound = true, encoding = "utf-8", factory = YamlPropertySourceFactory.class)
 public class InfluxDbConfig {
     /**
      * influxDb注册器

custom-gateway-core/src/main/java/com/shkpr/service/customgateway/core/config/MybatisConfig.java

@@ -25,10 +25,8 @@ import static com.shkpr.service.customgateway.core.constants.DataSourceNames.PRI
  */
 @Configuration
 @EnableConfigurationProperties(MybatisProperties.class)
-@MapperScan(sqlSessionTemplateRef = PRIMARY + "SqlSessionTemplate",
-        basePackages = "com.shkpr.service.customgateway.**.mapper." + PRIMARY)
-public class MybatisConfig {
-
+@MapperScan(sqlSessionTemplateRef = PRIMARY + "SqlSessionTemplate", basePackages = "com.shkpr.service.customgateway.**.mapper." + PRIMARY)
+public class PrimaryMybatisConfig {
     private static final ResourcePatternResolver resourceResolver = new PathMatchingResourcePatternResolver();
 
     /**
@@ -39,10 +37,16 @@ public class MybatisConfig {
     @Bean(PRIMARY + "SqlSessionFactory")
     public SqlSessionFactory mainSqlSessionFactoryBean(MybatisProperties properties
             , @Qualifier(PRIMARY + "DataSource") DataSource dataSource) throws Exception {
+        //配置工厂
         SqlSessionFactoryBean sessionFactoryBean = new SqlSessionFactoryBean();
         sessionFactoryBean.setDataSource(dataSource);
         sessionFactoryBean.setMapperLocations(resourceResolver.getResources("classpath*:mapper/*.xml"));
-        sessionFactoryBean.setConfiguration(properties.getConfiguration());
+        //配置session
+        org.apache.ibatis.session.Configuration configuration = new org.apache.ibatis.session.Configuration();
+        configuration.setCallSettersOnNulls(properties.getConfiguration().isCallSettersOnNulls());
+        configuration.setMapUnderscoreToCamelCase(properties.getConfiguration().isMapUnderscoreToCamelCase());
+        configuration.setCacheEnabled(properties.getConfiguration().isCacheEnabled());
+        sessionFactoryBean.setConfiguration(configuration);
         return sessionFactoryBean.getObject();
     }
 

custom-gateway-zhscada/src/main/java/com/shkpr/service/customgateway/zhscada/components/DataCollector.java

@@ -33,7 +33,7 @@ public class DataCollector {
      * log
      */
     private static final String CLASS_NAME = "DataCollector";
-    private static final String BIZ_TYPE = LogFlagBusiType.BUSI_ALL.toStrValue();
+    private static final String BIZ_TYPE = LogFlagBusiType.ZHONG_HUAN_SACADA.toStrValue();
 
     final
     CallingProperties callingProperties;

custom-gateway-zhscada/src/main/java/com/shkpr/service/customgateway/zhscada/components/DataMigrator.java

@@ -0,0 +1,124 @@
+package com.shkpr.service.customgateway.zhscada.components;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import com.global.base.log.LogLevelFlag;
+import com.global.base.log.LogPrintMgr;
+import com.shkpr.service.customgateway.core.components.DeviceRegistry;
+import com.shkpr.service.customgateway.core.constants.LogFlagBusiType;
+import com.shkpr.service.customgateway.core.domain.Device;
+import com.shkpr.service.customgateway.core.domain.DeviceTag;
+import com.shkpr.service.customgateway.core.utils.InfluxDbUtil;
+import com.shkpr.service.customgateway.zhscada.constants.ScadaPlatformMetadata;
+import com.shkpr.service.customgateway.zhscada.domain.po.SensorCollectData;
+import com.shkpr.service.customgateway.zhscada.properties.MigrateProperties;
+import com.shkpr.service.customgateway.zhscada.service.SensorCollectDataService;
+import org.apache.commons.lang3.StringUtils;
+import org.influxdb.dto.Point;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
+import org.springframework.core.io.support.ResourcePatternResolver;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * 数据迁移器
+ *
+ * @author 欧阳劲驰
+ * @serial 1.0.4
+ */
+@Component
+public class DataMigrator {
+    //log
+    private static final String CLASS_NAME = "DataCollector";
+    private static final String BIZ_TYPE = LogFlagBusiType.ZHONG_HUAN_SACADA.toStrValue();
+
+    final
+    MigrateProperties migrateProperties;
+    final
+    DeviceRegistry deviceRegistry;
+    final
+    SensorCollectDataService sensorCollectDataService;
+    final
+    InfluxDbUtil influxDbUtil;
+
+    //传感器映射,k:tag,v:code
+    private Map<String, String> sensorMap = new HashMap<>();
+
+    public DataMigrator(MigrateProperties migrateProperties, DeviceRegistry deviceRegistry, SensorCollectDataService sensorCollectDataService, InfluxDbUtil influxDbUtil) {
+        this.migrateProperties = migrateProperties;
+        this.deviceRegistry = deviceRegistry;
+        this.sensorCollectDataService = sensorCollectDataService;
+        this.influxDbUtil = influxDbUtil;
+    }
+
+    /**
+     * 迁移scada
+     */
+    public void migrateScada() {
+        LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_INFO, BIZ_TYPE, CLASS_NAME
+                , "开始迁移Scada数据");
+        long begin = System.currentTimeMillis();
+
+        //加载传感器配置
+        loadSensors();
+        //设备
+        List<Device> devices = deviceRegistry.findAll(ScadaPlatformMetadata.NAME);
+        //遍历设备,读取数据并写入
+        devices.forEach(device -> {
+            //遍历tag
+            for (DeviceTag tag : device.getTags()) {
+                //传感器code
+                String code = sensorMap.getOrDefault(tag.getTag(), null);
+                if (StringUtils.isBlank(code)) continue;
+                //起止时间
+                SensorCollectData timeRange = sensorCollectDataService.findTimeRange(migrateProperties.getSchema(),code);
+                LocalDateTime minTime = timeRange.getMinTime();
+                LocalDateTime maxTime = timeRange.getMaxTime();
+                //查询数据
+                List<SensorCollectData> dates = sensorCollectDataService.findAlign(migrateProperties.getSchema(),
+                        minTime, maxTime, migrateProperties.getInterval(), migrateProperties.getIntervalUnit(),
+                        migrateProperties.getAlignUnit(), code);
+
+                //构建influxdb
+                List<Point> points = dates.parallelStream().map(d -> d.toPoint(devices, sensorMap))
+                        .filter(Objects::nonNull).collect(Collectors.toList());
+                //写入influxDb
+                influxDbUtil.write(points);
+            }
+        });
+
+
+        long end = System.currentTimeMillis();
+        LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_INFO, BIZ_TYPE, CLASS_NAME
+                , String.format("结束迁移Scada数据 用时(毫秒):%d", (end - begin))
+        );
+
+    }
+
+    /**
+     * 加载传感器
+     */
+    private void loadSensors() {
+        try {
+            //解析文件
+            ResourcePatternResolver resourceResolver = new PathMatchingResourcePatternResolver();
+            Resource resources = resourceResolver.getResources(migrateProperties.getSensorMapPath())[0];
+            //解析map
+            ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+            sensorMap = mapper.readValue(resources.getFile(), mapper.getTypeFactory()
+                    .constructMapType(HashMap.class, String.class, String.class));
+        } catch (IOException e) {
+            LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR, BIZ_TYPE, CLASS_NAME
+                    , String.format("传感器映射配置文件读取失败 error:%s", e)
+            );
+        }
+    }
+}

custom-gateway-zhscada/src/main/java/com/shkpr/service/customgateway/zhscada/config/ScadaDataSourceConfig.java

@@ -0,0 +1,41 @@
+package com.shkpr.service.customgateway.zhscada.config;
+
+import com.shkpr.service.customgateway.core.properties.DataSourceProperties;
+import com.zaxxer.hikari.HikariDataSource;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.jdbc.DataSourceBuilder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.sql.DataSource;
+
+/**
+ * scada数据源配置
+ *
+ * @author 欧阳劲驰
+ * @since 1.0.0
+ */
+@Configuration
+@EnableConfigurationProperties(DataSourceProperties.class)
+public class ScadaDataSourceConfig {
+    //名称
+    public final static String SCADA = "scada";
+
+    /**
+     * @param properties 属性
+     * @return 主要数据源
+     */
+    @Bean(name = SCADA + "DataSource")
+    public DataSource primaryDataSource(DataSourceProperties properties) {
+        //获取配置
+        DataSourceProperties.DataSource source = properties.getMulti().get(SCADA);
+        //构建数据源
+        return DataSourceBuilder.create()
+                .type(HikariDataSource.class)
+                .url(source.getUrl())
+                .username(source.getUsername())
+                .password(source.getPassword())
+                .driverClassName(source.getDriverClassName())
+                .build();
+    }
+}

custom-gateway-zhscada/src/main/java/com/shkpr/service/customgateway/zhscada/config/ScadaMybatisConfig.java

@@ -0,0 +1,70 @@
+package com.shkpr.service.customgateway.zhscada.config;
+
+import org.apache.ibatis.session.SqlSessionFactory;
+import org.mybatis.spring.SqlSessionFactoryBean;
+import org.mybatis.spring.SqlSessionTemplate;
+import org.mybatis.spring.annotation.MapperScan;
+import org.mybatis.spring.boot.autoconfigure.MybatisProperties;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
+import org.springframework.core.io.support.ResourcePatternResolver;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+
+import javax.sql.DataSource;
+
+import static com.shkpr.service.customgateway.zhscada.config.ScadaDataSourceConfig.SCADA;
+
+/**
+ * scada mybatis配置
+ *
+ * @author 欧阳劲驰
+ * @since 1.0.0
+ */
+@Configuration
+@EnableConfigurationProperties(MybatisProperties.class)
+@MapperScan(sqlSessionTemplateRef = SCADA + "SqlSessionTemplate", basePackages = "com.shkpr.service.customgateway.zhscada.mapper." + SCADA)
+public class ScadaMybatisConfig {
+    private static final ResourcePatternResolver resourceResolver = new PathMatchingResourcePatternResolver();
+
+    /**
+     * @param dataSource 数据源
+     * @return SQL会话工厂
+     * @throws Exception NotInitializedException
+     */
+    @Bean(SCADA + "SqlSessionFactory")
+    public SqlSessionFactory mainSqlSessionFactoryBean(MybatisProperties properties
+            , @Qualifier(SCADA + "DataSource") DataSource dataSource) throws Exception {
+        //配置工厂
+        SqlSessionFactoryBean sessionFactoryBean = new SqlSessionFactoryBean();
+        sessionFactoryBean.setDataSource(dataSource);
+        sessionFactoryBean.setMapperLocations(resourceResolver.getResources("classpath*:mapper/*.xml"));
+        //配置session
+        org.apache.ibatis.session.Configuration configuration = new org.apache.ibatis.session.Configuration();
+        configuration.setCallSettersOnNulls(properties.getConfiguration().isCallSettersOnNulls());
+        configuration.setMapUnderscoreToCamelCase(properties.getConfiguration().isMapUnderscoreToCamelCase());
+        configuration.setCacheEnabled(properties.getConfiguration().isCacheEnabled());
+        sessionFactoryBean.setConfiguration(configuration);
+        return sessionFactoryBean.getObject();
+    }
+
+    /**
+     * @param sessionFactory SQL会话工厂
+     * @return SQL会话模版
+     */
+    @Bean(SCADA + "SqlSessionTemplate")
+    public SqlSessionTemplate mainSqlSessionTemplate(@Qualifier(SCADA + "SqlSessionFactory") SqlSessionFactory sessionFactory) {
+        return new SqlSessionTemplate(sessionFactory);
+    }
+
+    /**
+     * @param dataSource 数据源
+     * @return 事物管理器
+     */
+    @Bean(SCADA + "DbTransactionManager")
+    public DataSourceTransactionManager mainDbTransactionManager(@Qualifier(SCADA + "DataSource") DataSource dataSource) {
+        return new DataSourceTransactionManager(dataSource);
+    }
+}

custom-gateway-zhscada/src/main/java/com/shkpr/service/customgateway/zhscada/domain/po/SensorCollectData.java

@@ -0,0 +1,93 @@
+package com.shkpr.service.customgateway.zhscada.domain.po;
+
+import com.shkpr.service.customgateway.core.constants.InfluxdbMetadata;
+import com.shkpr.service.customgateway.core.domain.Device;
+import com.shkpr.service.customgateway.core.domain.DeviceTag;
+import lombok.Data;
+import org.apache.commons.lang3.StringUtils;
+import org.influxdb.dto.Point;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 传感器采集数据表
+ *
+ * @author 欧阳劲驰
+ * @since 1.0.0
+ */
+@Data
+public class SensorCollectData {
+    /**
+     * 传感器code
+     */
+    private String code;
+    /**
+     * 时间
+     */
+    private LocalDateTime time;
+    /**
+     * 最小时间
+     */
+    private LocalDateTime minTime;
+    /**
+     * 最大时间
+     */
+    private LocalDateTime maxTime;
+    /**
+     * 值
+     */
+    private Double value;
+
+    /**
+     * 转influxdb对象
+     *
+     * @param devices   设备列表
+     * @param sensorMap 传感器映射
+     * @return influxdb对象
+     */
+    public Point toPoint(List<Device> devices, Map<String, String> sensorMap) {
+        if (StringUtils.isBlank(code) || value == null) return null;
+
+        //获取标签
+        String tag = sensorMap.entrySet()
+                .stream()
+                .filter(entry -> Objects.equals(entry.getValue(), code))
+                .map(Map.Entry::getKey)
+                .findFirst()
+                .orElse(null);
+        if (tag == null) return null;
+        //获取设备
+        Device device = devices.stream()
+                .filter(d -> d.getTags().stream().map(DeviceTag::getTag).anyMatch(tag::equals))
+                .findFirst().orElse(null);
+        if (device == null) return null;
+        //获取标签
+        DeviceTag deviceTag = device.getTags().stream()
+                .filter(t -> Objects.equals(tag, t.getTag()))
+                .findFirst().orElse(null);
+        if (deviceTag == null) return null;
+
+        //获取时间戳
+        long timestamp = time.truncatedTo(ChronoUnit.MINUTES)
+                .atZone(ZoneId.systemDefault())
+                .toInstant()
+                .toEpochMilli();
+        //构建influxdb
+        return Point
+                //表名
+                .measurement(deviceTag.getMeasurement())
+                //设备id
+                .tag(InfluxdbMetadata.Tags.DEVICE_ID, device.getDeviceId())
+                //时间
+                .time(timestamp, TimeUnit.MILLISECONDS)
+                //值
+                .addField(deviceTag.getField(), value)
+                .build();
+    }
+}

custom-gateway-zhscada/src/main/java/com/shkpr/service/customgateway/zhscada/manager/DataManager.java

@@ -9,19 +9,19 @@ import org.springframework.stereotype.Component;
 import javax.annotation.PostConstruct;
 
 /**
- * 物联网管理
+ * 数据采集管理
  *
  * @author 欧阳劲驰
  * @serial 1.0.0
  */
 @Component
-public class DataManager {
+public class DataCollectManager {
     final
     ThreadPoolTaskExecutor taskScheduler;
     final
     DataCollector dataCollector;
 
-    public DataManager(ThreadPoolTaskExecutor taskScheduler, DataCollector dataCollector) {
+    public DataCollectManager(ThreadPoolTaskExecutor taskScheduler, DataCollector dataCollector) {
         this.taskScheduler = taskScheduler;
         this.dataCollector = dataCollector;
     }

custom-gateway-zhscada/src/main/java/com/shkpr/service/customgateway/zhscada/manager/DataMigrateManager.java

@@ -0,0 +1,38 @@
+package com.shkpr.service.customgateway.zhscada.manager;
+
+import com.shkpr.service.customgateway.zhscada.components.DataMigrator;
+import com.shkpr.service.customgateway.zhscada.properties.MigrateProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+
+/**
+ * 数据迁移管理
+ *
+ * @author 欧阳劲驰
+ * @serial 1.0.0
+ */
+@Component
+@EnableConfigurationProperties(MigrateProperties.class)
+public class DataMigrateManager {
+    final
+    ThreadPoolTaskExecutor taskScheduler;
+    final
+    DataMigrator dataMigrator;
+
+    public DataMigrateManager(ThreadPoolTaskExecutor taskScheduler, DataMigrator dataMigrator) {
+        this.taskScheduler = taskScheduler;
+        this.dataMigrator = dataMigrator;
+    }
+
+    /**
+     * 初始化
+     */
+    @PostConstruct
+    public void init() {
+        //迁移scada
+//        taskScheduler.execute(dataMigrator::migrateScada);
+    }
+}

custom-gateway-zhscada/src/main/java/com/shkpr/service/customgateway/zhscada/mapper/scada/SensorCollectDataMapper.java

@@ -0,0 +1,46 @@
+package com.shkpr.service.customgateway.zhscada.mapper.scada;
+
+import com.shkpr.service.customgateway.zhscada.domain.po.SensorCollectData;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.cursor.Cursor;
+
+import java.time.LocalDateTime;
+
+/**
+ * 采集历史mapper
+ *
+ * @author 欧阳劲驰
+ * @serial 1.0.0
+ */
+@Mapper
+public interface SensorCollectDataMapper {
+    /**
+     * 根据传感器code查询时间范围
+     *
+     * @param code 传感器code
+     * @return 时间范围
+     */
+    SensorCollectData findTimeRangeByCode(@Param("schema") String schema,@Param("code") String code);
+
+    /**
+     * 根据传感器code查询对齐值
+     *
+     * @param startTime    开始时间
+     * @param endTime      结束时间
+     * @param interval     采集间隔
+     * @param intervalUnit 采集间隔单位
+     * @param alignUnit    对齐单位
+     * @param code         传感器code
+     * @return 历史数据
+     */
+    Cursor<SensorCollectData> findAlignByTimeAndCode(
+            @Param("schema") String schema,
+            @Param("startTime") LocalDateTime startTime,
+            @Param("endTime") LocalDateTime endTime,
+            @Param("interval") Integer interval,
+            @Param("intervalUnit") String intervalUnit,
+            @Param("alignUnit") String alignUnit,
+            @Param("code") String code
+    );
+}

custom-gateway-zhscada/src/main/java/com/shkpr/service/customgateway/zhscada/properties/MigrateProperties.java

@@ -0,0 +1,37 @@
+package com.shkpr.service.customgateway.zhscada.properties;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * 迁移属性
+ *
+ * @author 欧阳劲驰
+ * @serial 1.0.4
+ */
+@Getter
+@Setter
+@ConfigurationProperties(prefix = "migrate")
+public class MigrateProperties {
+    /**
+     * 传感器映射文件路径
+     */
+    private String sensorMapPath;
+    /**
+     * 表名
+     */
+    private String schema;
+    /**
+     * 数据间隔
+     */
+    private Integer interval = 1;
+    /**
+     * 数据间隔单位
+     */
+    private String intervalUnit = "minute";
+    /**
+     * 对齐单位
+     */
+    private String alignUnit = "mi";
+}

custom-gateway-zhscada/src/main/java/com/shkpr/service/customgateway/zhscada/service/SensorCollectDataService.java

@@ -0,0 +1,38 @@
+package com.shkpr.service.customgateway.zhscada.service;
+
+import com.shkpr.service.customgateway.zhscada.domain.po.SensorCollectData;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+/**
+ * 传感器采集数据service
+ *
+ * @author 欧阳劲驰
+ * @since 1.0.0
+ */
+public interface SensorCollectDataService {
+    /**
+     * 查询时间范围
+     *
+     * @param schema 架构名
+     * @param code   传感器code
+     * @return 时间范围
+     */
+    SensorCollectData findTimeRange(String schema, String code);
+
+    /**
+     * 查询对齐值
+     *
+     * @param schema       架构名
+     * @param startTime    开始时间
+     * @param endTime      结束时间
+     * @param interval     采集间隔
+     * @param intervalUnit 采集间隔单位
+     * @param alignUnit    对齐单位
+     * @param code         传感器code
+     * @return 历史数据
+     */
+    List<SensorCollectData> findAlign(String schema, LocalDateTime startTime, LocalDateTime endTime
+            , Integer interval, String intervalUnit, String alignUnit, String code);
+}

custom-gateway-zhscada/src/main/java/com/shkpr/service/customgateway/zhscada/service/impl/SensorCollectDataServiceImpl.java

@@ -0,0 +1,107 @@
+package com.shkpr.service.customgateway.zhscada.service.impl;
+
+import com.global.base.log.LogLevelFlag;
+import com.global.base.log.LogPrintMgr;
+import com.shkpr.service.customgateway.core.constants.LogFlagBusiType;
+import com.shkpr.service.customgateway.zhscada.domain.po.SensorCollectData;
+import com.shkpr.service.customgateway.zhscada.mapper.scada.SensorCollectDataMapper;
+import com.shkpr.service.customgateway.zhscada.service.SensorCollectDataService;
+import org.apache.ibatis.cursor.Cursor;
+import org.apache.ibatis.session.SqlSession;
+import org.apache.ibatis.session.SqlSessionFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.io.IOException;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+
+import static com.shkpr.service.customgateway.zhscada.config.ScadaDataSourceConfig.SCADA;
+
+/**
+ * 传感器采集数据service实现
+ *
+ * @author 欧阳劲驰
+ * @since 1.0.0
+ */
+@Service
+public class SensorCollectDataServiceImpl implements SensorCollectDataService {
+    /**
+     * log
+     */
+    private final static String CLASS_NAME = "SensorCollectDataServiceImpl";
+    private final static String BIZ_TYPE = LogFlagBusiType.ZHONG_HUAN_SACADA.toStrValue();
+
+    final
+    SqlSessionFactory sqlSessionFactory;
+    final
+    SensorCollectDataMapper sensorCollectDataMapper;
+
+    public SensorCollectDataServiceImpl(@Qualifier(SCADA + "SqlSessionFactory") SqlSessionFactory sqlSessionFactory
+            , SensorCollectDataMapper sensorCollectDataMapper) {
+        this.sqlSessionFactory = sqlSessionFactory;
+        this.sensorCollectDataMapper = sensorCollectDataMapper;
+    }
+
+    /**
+     * 根据传感器code查询时间范围
+     *
+     * @param schema 架构名
+     * @param code   传感器code
+     * @return 时间范围
+     */
+    @Override
+    public SensorCollectData findTimeRange(String schema, String code) {
+        return sensorCollectDataMapper.findTimeRangeByCode(schema, code);
+    }
+
+    /**
+     * 根据传感器code查询对齐值
+     *
+     * @param schema       架构名
+     * @param startTime    开始时间
+     * @param endTime      结束时间
+     * @param interval     采集间隔
+     * @param intervalUnit 采集间隔单位
+     * @param alignUnit    对齐单位
+     * @param code         传感器code
+     * @return 历史数据
+     */
+    @Override
+    @Transactional(transactionManager = SCADA + "DbTransactionManager")
+    public List<SensorCollectData> findAlign(String schema, LocalDateTime startTime, LocalDateTime endTime, Integer interval, String intervalUnit, String alignUnit, String code) {
+        //获取session和mapper
+        try (SqlSession session = sqlSessionFactory.openSession()) {
+            SensorCollectDataMapper mapper = session.getMapper(SensorCollectDataMapper.class);
+
+            //采集历史集合
+            List<SensorCollectData> collectHistories = new ArrayList<>();
+            LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_INFO, BIZ_TYPE, CLASS_NAME,
+                    String.format("开始根据时间和标签查询传感器对齐数据, 开启事务和游标 开始时间:%s, 结束时间:%s, 间隔:%d, 间隔单位:%s ,对齐单位:%s, 传感器:%s",
+                            startTime, endTime, interval, intervalUnit, alignUnit, code));
+            long begin = System.currentTimeMillis();
+
+            //获取游标
+            try (Cursor<SensorCollectData> cursor = mapper.findAlignByTimeAndCode(schema, startTime, endTime, interval, intervalUnit, alignUnit, code)) {
+                //迭代游标
+                for (SensorCollectData data : cursor) {
+                    //检查游标完成
+                    if (cursor.isConsumed()) break;
+                    //填入数据
+                    collectHistories.add(data);
+                }
+            } catch (IOException e) {
+                LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR, BIZ_TYPE, CLASS_NAME, String.format("关闭游标失败  error:%s", e));
+            }
+
+            long end = System.currentTimeMillis();
+            LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_INFO, BIZ_TYPE, CLASS_NAME,
+                    String.format("完成根据时间和标签查询传感器对齐数据, 关闭事务和游标 开始时间:%s, 结束时间:%s, 间隔:%d, 间隔单位:%s, 用时(毫秒):%d",
+                            startTime, endTime, interval, intervalUnit, (end - begin)));
+
+            return collectHistories;
+        }
+    }
+}

custom-gateway-zhscada/src/main/resources/mapper/SensorCollectDataMapper.xml

@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.shkpr.service.customgateway.zhscada.mapper.scada.SensorCollectDataMapper">
+    <select id="findTimeRangeByCode" resultType="com.shkpr.service.customgateway.zhscada.domain.po.SensorCollectData">
+        select min(coll_time) as min_time,
+               max(coll_time) as max_time
+        from ${schema}.t_sersor_coll_data_day
+        where sensor_code = #{code,jdbcType=VARCHAR}
+    </select>
+
+    <select id="findAlignByTimeAndCode" resultType="com.shkpr.service.customgateway.zhscada.domain.po.SensorCollectData"
+            fetchSize="3000">
+        with time_seq (start_time)
+                 as (select trunc(#{startTime,jdbcType=TIMESTAMP}, #{alignUnit,jdbcType=VARCHAR}) as start_time
+                     from dual
+                     union all
+                     select start_time + numtodsinterval(#{interval,jdbcType=INTEGER}, #{intervalUnit,jdbcType=VARCHAR})
+                     from time_seq
+                     where start_time +
+                           numtodsinterval(#{interval,jdbcType=INTEGER}, #{intervalUnit,jdbcType=VARCHAR}) &lt;
+                           #{endTime,jdbcType=TIMESTAMP})
+        select ts.start_time  as time,
+               sh.coll_value  as value,
+               sh.sensor_code as code
+        from time_seq ts
+                 left join (select sensor_code,
+                                   coll_value,
+                                   trunc(coll_time, #{alignUnit,jdbcType=VARCHAR})                                as minute_trunc,
+                                   row_number() over (partition by trunc(coll_time, #{alignUnit,jdbcType=VARCHAR}), sensor_code
+                                       order by abs(coll_time - trunc(coll_time, #{alignUnit,jdbcType=VARCHAR}))) as rn
+                            from ${schema}.t_sersor_coll_data_day
+                            where sensor_code = #{code,jdbcType=VARCHAR}
+                              and coll_time &gt;= #{startTime,jdbcType=TIMESTAMP}
+                              and coll_time &lt; #{endTime,jdbcType=TIMESTAMP}) sh
+                           on sh.minute_trunc = ts.start_time
+                               and sh.rn = 1
+        order by ts.start_time
+    </select>
+
+</mapper>

custom-gateway-zhscada/src/main/resources/sensor-id.yml

@@ -0,0 +1,2 @@
+sc_ywbz_ssll: 800
+sc_ywbz_ljll: 801

db_config.yml

@@ -1,35 +0,0 @@
-spring:
-  #数据源
-  datasource:
-    multi:
-      #主要
-      primary:
-        url: jdbc:postgresql://223.75.194.87:6057/watersmart?useSSL=false&useAffectedRows=false&allowMultiQueries=true&rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai
-        username: postgres
-        password: kpr.23417.postgres
-        driver-class-name: org.postgresql.Driver
-      #嵌入式
-      embedded:
-        url: jdbc:h2:file:./data
-        driver-class-name: org.h2.Driver
-        username: sa
-        password: kpr.25535.sa
-  #influxdb
-  influx:
-    read-timeout: 30000
-    clients:
-      #枣阳
-      - url: http://192.168.44.58:8086/
-        user: kpr
-        password: kpr.2025@117.influxdb
-        database: iot
-      #测试
-      - url: http://119.96.165.176:8086/
-        user: kpr
-        password: kpr.2024@117.influxdb
-        database: iot
-      #中环
-#      - url: http://127.0.0.1:8086/
-#        user: kpr
-#        password: kpr.2025@117.influxdb
-#        database: iot