填加子表

build.gradle

@@ -38,7 +38,8 @@ dependencies {
     implementation('org.apache.poi:poi-ooxml:5.2.3')
     implementation 'cn.hutool:hutool-poi:5.8.12'
     implementation 'cn.hutool:hutool-core:5.8.13'
-    implementation("com.taosdata.jdbc:taos-jdbcdriver:3.1.0")
+    implementation 'com.taosdata.jdbc:taos-jdbcdriver:3.1.0'
+    implementation 'com.alibaba:druid:1.2.16'
 }
 
 tasks.named('test') {

src/main/java/com/gyee/redis2taos/config/TaosConfig.java

@@ -1,5 +1,6 @@
 package com.gyee.redis2taos.config;
 
+import com.alibaba.druid.pool.DruidDataSource;
 import com.taosdata.jdbc.TSDBDriver;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
@@ -45,6 +46,7 @@ public class TaosConfig {
 
     private Connection connection = null;
     private Connection createCon = null;
+    private DruidDataSource dataSource = null;
 
     public Connection getInstance() throws Exception {
         if (null == connection || connection.isClosed())
@@ -53,6 +55,11 @@ public class TaosConfig {
         return connection;
     }
 
+    /**
+     * 建表用的链接，没有数据库
+     * @return
+     * @throws Exception
+     */
     public Connection getCreateCon() throws Exception {
         if (null == createCon || createCon.isClosed()) {
             Class.forName(driverType);
@@ -100,4 +107,27 @@ public class TaosConfig {
     public String getDIStableName() {
         return diStableName;
     }
+
+    private DruidDataSource getConnectionPool() throws ClassNotFoundException {
+        if(dataSource==null){
+            dataSource = new DruidDataSource();
+            // jdbc properties
+            Class.forName(driverType);
+            String jdbcUrl = "jdbc:TAOS-RS://" + serverIp + ":" + serverPort + "/" + dbName;
+            if (driverType.equals("com.taosdata.jdbc.TSDBDriver"))
+                jdbcUrl = "jdbc:TAOS://" + serverIp + ":" + serverPort + "/" + dbName;
+            //
+            dataSource.setDriverClassName(driverType);
+            dataSource.setUrl(jdbcUrl);
+            dataSource.setUsername(userName);
+            dataSource.setPassword(password);
+            // pool configurations
+            dataSource.setInitialSize(10);
+            dataSource.setMinIdle(10);
+            dataSource.setMaxActive(10);
+            dataSource.setMaxWait(30000);
+            dataSource.setValidationQuery("select server_status()");
+        }
+        return dataSource;
+    }
 }

src/main/java/com/gyee/redis2taos/config/ThreadPoolTaskConfig.java

@@ -18,11 +18,11 @@ public class ThreadPoolTaskConfig {
     /**
      * 核心线程数（默认线程数）
      */
-    private static final int corePoolSize = 64;
+    private static final int corePoolSize = 256;
     /**
      * 最大线程数
      */
-    private static final int maxPoolSize = 128;
+    private static final int maxPoolSize = 512;
     /**
      * 允许线程空闲时间（单位：默认为秒）
      */
@@ -30,7 +30,7 @@ public class ThreadPoolTaskConfig {
     /**
      * 缓冲队列大小
      */
-    private static final int queueCapacity = 1000;
+    private static final int queueCapacity = 128;
     /**
      * 允许等待最长时间
      */
@@ -57,6 +57,9 @@ public class ThreadPoolTaskConfig {
         executor.setQueueCapacity(queueCapacity);
         executor.setKeepAliveSeconds(keepAliveTime);
         executor.setThreadNamePrefix(threadNamePrefix);
+        // 该方法用来设置 线程池关闭 的时候 等待 所有任务都完成后，再继续 销毁 其他的 Bean，
+        // 这样这些 异步任务 的 销毁 就会先于 数据库连接池对象 的销毁。
+        executor.setWaitForTasksToCompleteOnShutdown(true);
         executor.setAwaitTerminationSeconds(awaitTime);
 
         // 线程池对拒绝任务的处理策略

src/main/java/com/gyee/redis2taos/controller/TaosController.java

@@ -44,7 +44,7 @@ public class TaosController {
     @GetMapping("/table/create")
     private void tableCreat() {
         try {
-            taosService.createTables();
+            taosService.createTables2();
         } catch (Exception e) {
             e.printStackTrace();
         }

src/main/java/com/gyee/redis2taos/service/TaosService.java

@@ -1,6 +1,9 @@
 package com.gyee.redis2taos.service;
 
 import com.gyee.redis2taos.config.TaosConfig;
+import com.gyee.redis2taos.timeseries.BasicTsPoint;
+import com.gyee.redis2taos.timeseries.TsDataType;
+import com.gyee.redis2taos.timeseries.TsPoint;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 
@@ -9,6 +12,9 @@ import java.sql.ResultSet;
 import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 @Slf4j
 @Service
@@ -18,6 +24,8 @@ public class TaosService {
     private TaosConfig taosConfig;
     @Resource
     private FileService fileService;
+    @Resource
+    private RedisService redisService;
 
     public boolean createDataBase(int keepday) throws Exception {
         log.info("创建taos库!");
@@ -75,6 +83,49 @@ public class TaosService {
         return i;
     }
 
+    public int createTables2() throws Exception {
+        log.info("创建taos表!");
+        Statement st = taosConfig.getInstance().createStatement();
+
+        Set<String> keys = redisService.getKeys();
+        if(keys.size()==0) return 0;
+        log.info("读取rediskeys成功！！！" + keys.size() + "个点。");
+        //按类型分点
+        Map<TsDataType, List<BasicTsPoint>> di = keys.stream().map(ss -> {
+            if (ss.contains("DI") || ss.contains("di")) {
+                return new BasicTsPoint(ss, TsDataType.BOOLEAN);
+            } else {
+                return new BasicTsPoint(ss, TsDataType.DOUBLE);
+            }
+        }).collect(Collectors.groupingBy(TsPoint::getTsDataType));
+        //di点
+        List<String> boolCollect = di.get(TsDataType.BOOLEAN).stream().map(k1 -> k1.getId()).collect(Collectors.toList());
+        //ai点
+        List<String> douCollect = di.get(TsDataType.DOUBLE).stream().map(k1 -> k1.getId()).collect(Collectors.toList());
+
+        int k = 0;
+        for (String s : boolCollect) {
+            st.executeUpdate(getDiSql(s));
+            log.info(String.valueOf(++k));
+        }
+        for (String s : douCollect) {
+            st.executeUpdate(getAiSql(s));
+            log.info(String.valueOf(++k));
+        }
+        log.info("taos表创建完成！");
+        return 1;
+    }
+
+    private String getAiSql(String s) {
+        return "CREATE TABLE IF NOT EXISTS "+taosConfig.getDbName()+"."+s+" USING "+taosConfig.getDbName()
+                +"."+taosConfig.getAIStableName()+" TAGS('','','','','',1,'');";
+    }
+
+    private String getDiSql(String s) {
+        return "CREATE TABLE IF NOT EXISTS "+taosConfig.getDbName()+"."+s+" USING "+taosConfig.getDbName()
+                +"."+taosConfig.getDIStableName()+" TAGS('','','','','',1,'');";
+    }
+
     public List<String> deleteTables() throws Exception {
         Statement st = taosConfig.getInstance().createStatement();
         String showSTableSql = "SHOW DATABASES;";

src/main/java/com/gyee/redis2taos/service/TaskRedis2Taos.java

@@ -73,7 +73,7 @@ public class TaskRedis2Taos implements Runnable {
                     response = (List<String>) objects.get(i);
 
                     value = response.get(1);
-                    if (value == null || "".equals(value) || "-nan".equals(value)) continue;
+                    if (value == null || "".equals(value) || value.contains("-nan")) continue;
                     datetime = response.get(2);
                     if (StringUtils.isEmpty(datetime)) continue;
                     pointid = response.get(0);
@@ -122,7 +122,7 @@ public class TaskRedis2Taos implements Runnable {
                     response = (List<String>) objects.get(i);
 
                     value = response.get(1);
-                    if (value == null || "".equals(value) || "-nan".equals(value)) continue;
+                    if (value == null || "".equals(value) || value.contains("-nan")) continue;
                     datetime = response.get(2);
                     if (StringUtils.isEmpty(datetime)) continue;
                     pointid = response.get(0);