移植2.x带.的taosadapter

README.md

@@ -13,7 +13,8 @@
 * dao-edos -- edos访问组件
 * dao-golden -- 庚顿访问组件
 * dao-hadoop -- hadoop访问组件
-* dao-mysql -- mysql访问组件
+* dao-sql -- jpa实现组件
+* dao-simulator -- 实时数据模拟
 
 
 ### electricity -- 电量计算

build.gradle

@@ -16,10 +16,11 @@ subprojects {
         compileOnly("org.projectlombok:lombok:$lombokVersion")
         annotationProcessor("org.projectlombok:lombok:$lombokVersion")
 
-        implementation("org.apache.logging.log4j:log4j-core:$log4jVersion")
-        implementation("org.apache.logging.log4j:log4j-jul:$log4jVersion")
-        implementation("org.apache.logging.log4j:log4j-api:$log4jVersion")
-        implementation("org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion")
+//        implementation("org.apache.logging.log4j:log4j-core:$log4jVersion")
+//        implementation("org.apache.logging.log4j:log4j-jul:$log4jVersion")
+//        implementation("org.apache.logging.log4j:log4j-api:$log4jVersion")
+//        implementation("org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion")
+        implementation("org.springframework.boot:spring-boot-starter-log4j2:$springBootVersion")
     }
 
     repositories {

gradle.properties

@@ -41,7 +41,7 @@ alibabaDruidVersion=1.2.9
 groovyVersion=3.0.9
 openFeignVersion=11.8
 postgresqlDriverVersion=42.4.0
-#taosVersion=2.0.35
-taosVersion=3.0.3
+taosVersion2=2.0.39
+taosVersion3=3.0.3
 xxlJobVersion=2.3.0
 mybatisPlusVersion=3.1.1

settings.gradle

@@ -8,6 +8,7 @@ include "timeseries:dao-redis"
 include "timeseries:dao-simulator"
 include "timeseries:dao-sql"
 include "timeseries:dao-taoscz"
+include "timeseries:dao-taos"
 include "timeseries:data-adapter"
 
 include "electricity:meter"

timeseries/dao-taos/build.gradle

@@ -0,0 +1,29 @@
+buildscript {
+    repositories {
+        mavenLocal()
+        maven {
+            allowInsecureProtocol = true
+            url "http://maven.aliyun.com/nexus/content/groups/public" }
+        mavenCentral()
+    }
+    dependencies {
+        classpath("$bootGroup:spring-boot-gradle-plugin:$springBootVersion")
+    }
+}
+
+apply plugin: "$bootGroup"
+apply plugin: 'io.spring.dependency-management'
+bootJar.enabled = false
+jar.enabled = true
+
+dependencies {
+    implementation project(":common:utils")
+    implementation project(":common:data")
+    implementation project(":timeseries:dao-interface")
+//    implementation fileTree(dir: 'src/main/lib', include: '*.jar')
+//    implementation("$bootGroup:spring-boot-starter-log4j2")
+    implementation ("com.taosdata.jdbc:taos-jdbcdriver:$taosVersion2"){
+        exclude group: 'com.alibaba', module: 'fastjson'
+    }
+    implementation("com.alibaba:druid-spring-boot-starter:$alibabaDruidVersion")
+}

timeseries/dao-taos/src/main/java/com/gyee/wisdom/dao/taoscz/TaosConfiguration.java

@@ -0,0 +1,87 @@
+package com.gyee.wisdom.dao.taoscz;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+
+
+@Slf4j
+@Configuration
+public class TaosConfiguration {
+
+    @Value("${taos.server_ip:127.0.0.1}")
+    private String serverIp;
+
+    @Value("${taos.server_port:6030}")
+    private int serverPort;
+
+    @Value("${taos.user_name:root}")
+    private String userName;
+
+    @Value("${taos.password:taosdata}")
+    private String password;
+
+    @Value("${taos.pool_size:50}")
+    private int poolSize;
+
+    @Value("${taos.max_pool_size:200}")
+    private int maxPoolSize;
+
+    private DataSource dataSource;
+
+    private String dbname = "gdnxxny";
+
+
+    public Connection getConnect() {
+        try {
+            if (null == dataSource) {
+                dataSource = getDataSource();
+            }
+            return dataSource.getConnection();
+
+        } catch (Exception e) {
+            log.error(e.getMessage());
+        }
+
+        return null;
+    }
+
+    private DataSource getDataSource() {
+        DruidDataSource dataSource = new DruidDataSource();
+        // jdbc properties
+        String url = "jdbc:TAOS-RS://" + serverIp + ":" + serverPort + "/gdnxxny";
+        dataSource.setDriverClassName("com.taosdata.jdbc.rs.RestfulDriver");
+        dataSource.setUrl(url);
+        dataSource.setUsername(userName);
+        dataSource.setPassword(password);
+        // pool configurations
+        dataSource.setInitialSize(poolSize);
+        dataSource.setMinIdle(poolSize);
+        dataSource.setMaxActive(maxPoolSize);
+        dataSource.setMaxWait(300000);
+        dataSource.setRemoveAbandoned(true);
+        dataSource.setLogAbandoned(true);
+        dataSource.setRemoveAbandonedTimeoutMillis(300000);
+        dataSource.setValidationQuery("select server_status()");
+
+        log.info("涛思数据源初始化成功……");
+
+        return dataSource;
+    }
+
+    public void close(Connection conn, Statement st, ResultSet rs) {
+        try {
+            if (rs != null) rs.close();
+            if (st != null) st.close();
+            if (conn != null) conn.close();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+}

timeseries/dao-taos/src/main/java/com/gyee/wisdom/dao/taoscz/TaosCovertUtil.java

@@ -0,0 +1,44 @@
+package com.gyee.wisdom.dao.taoscz;
+
+public class TaosCovertUtil {
+
+    /**
+     * 根据传过来的点分析出场站名
+     * @param point
+     * @return
+     */
+    public static String coverStationPrefix(String point) {
+        String station = null;
+
+        if(point.toUpperCase().contains("SBQ"))
+            station = "sbq";
+        else if(point.toUpperCase().contains("MHS"))
+            station = "mhs";
+        else if(point.toUpperCase().contains("NSS"))
+            station = "nss";
+        else if(point.toUpperCase().contains("XS"))
+            station = "xs";
+        else if(point.toUpperCase().contains("QS"))
+            station = "qs";
+        else if(point.toUpperCase().contains("DWK"))
+            station = "dwk";
+        else if(point.toUpperCase().contains("PL"))
+            station = "pl";
+        else if(point.toUpperCase().contains("MCH"))
+            station = "mch";
+        else if(point.toUpperCase().contains("SH"))
+            station = "sh";
+        else if(point.toUpperCase().contains("XH"))
+            station = "xh";
+        else if(point.toUpperCase().contains("NXDQ"))
+            station = "nxdq";
+        else if(point.toUpperCase().contains("JSFW"))
+            station = "jsfw";
+        else if(point.toUpperCase().contains("HZJ"))
+            station = "hzj";
+        else
+            station = "qs";
+
+        return station;
+    }
+}

timeseries/dao-taos/src/main/java/com/gyee/wisdom/dao/taoscz/TaosDataChangeDao.java

@@ -0,0 +1,32 @@
+package com.gyee.wisdom.dao.taoscz;
+
+import com.gyee.wisdom.common.data.timeseries.TsData;
+import com.gyee.wisdom.common.data.timeseries.TsPoint;
+import com.gyee.wisdom.dao.timeseries.IDataChangeDao;
+import com.gyee.wisdom.dao.timeseries.TaosDao;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.Map;
+
+@Slf4j
+@Component
+@TaosDao
+public class TaosDataChangeDao implements IDataChangeDao {
+
+    @Override
+    public void registerTopic(String topic, List<TsPoint> tsPoints) {
+
+    }
+
+    @Override
+    public void unRegisterTopic(String topic) {
+
+    }
+
+    @Override
+    public Map<String, TsData> getTopicDataChange(String topic) {
+        return null;
+    }
+}

timeseries/dao-taos/src/main/java/com/gyee/wisdom/dao/taoscz/TaosHistoryDao.java

@@ -0,0 +1,410 @@
+package com.gyee.wisdom.dao.taoscz;
+
+import com.gyee.wisdom.common.data.timeseries.*;
+import com.gyee.wisdom.common.exception.WisdomException;
+import com.gyee.wisdom.dao.timeseries.IHistoryDao;
+import com.gyee.wisdom.dao.timeseries.TaosDao;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.sql.*;
+import java.util.*;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Component
+@TaosDao
+public class TaosHistoryDao implements IHistoryDao {
+
+    private long day_time = 86400000L;
+    private long month_time = 2592000000L;
+    private long half_year_time = 15552000000L * 4;
+    private long year_time = 31536000000L;
+
+    @Autowired
+    private TaosConfiguration tdUtil;
+    @Autowired
+    private ThreadPoolTaskConfig taskConfig;
+
+
+    /**
+     * 查询单个测点历史数据
+     * @param tsQuery
+     * @return
+     * @throws Exception
+     */
+    @Override
+    public List<TsData> getTsDataHistory(TsQuery tsQuery) throws Exception {
+        if (tsQuery.getTsPoint().getTsDataType() == TsDataType.DOUBLE) {
+            return getDoubleTsDataHistory(tsQuery);
+        } else //(tsQuery.getTsPoint().getTsDataType() == TsDataType.BOOLEAN) {
+            return getBooleanTsDataHistory(tsQuery);
+       /* } else if (tsQuery.getTsPoint().getTsDataType() == TsDataType.STRING) {
+            return getStringTsDataHistory(tsQuery);
+        } else if (tsQuery.getTsPoint().getTsDataType() == TsDataType.LONG) {
+            return getLongTsDataHistory(tsQuery);
+        }*/
+    }
+
+    /**
+     * 查询单测点min、max、avg值
+     * @param tsQuery
+     * @return
+     * @throws Exception
+     */
+    @Override
+    public List<DoubleStatData> getDoubleStatDataHistory(TsQuery tsQuery) throws Exception {
+        if (tsQuery.getTsPoint().getTsDataType() != TsDataType.DOUBLE) {
+            throw new WisdomException("无效的数据类型：" + tsQuery.getTsPoint().getTsDataType());
+        }
+
+        List<DoubleStatData> result = new ArrayList<>();
+        Connection connect = tdUtil.getConnect();
+
+        try {
+            Statement st = connect.createStatement();
+            String point = tsQuery.getTsPoint().getId();
+
+            StringBuilder sql = new StringBuilder();
+            sql.append("select avg(value),max(value),min(value) from ");
+            // sb.append(TaosCovertUtil.coverStationPrefix(point) + "." + point.replace(".", "_"));
+            sql.append("gdnxxny.`");
+            sql.append(point);
+            sql.append("` where ts between").append(tsQuery.getStartTs());
+            sql.append(" and ").append(tsQuery.getEndTs());
+            sql.append(" interval(").append(tsQuery.getInterval()).append("s)");
+//            sb.append(" where point_time>='").append(DateFormatUtils.format(tsQuery.getStartTs(), "yyyy-MM-dd HH:mm:ss:SSS"));
+//            sb.append("' and point_time<='").append(DateFormatUtils.format(tsQuery.getEndTs(), "yyyy-MM-dd HH:mm:ss:SSS"));
+//            sb.append("' interval(").append(tsQuery.getInterval()).append("s)");
+
+            // select avg(value),max(value),min(value) from `NSSDJL.NX_GD_NSSF_DD_P1_L1_001_FXWG004` where ts>=1654012800000 and ts<=1656604800000 interval(86400s)
+
+            log.info("4->" + sql);
+            ResultSet rs = st.executeQuery(sql.toString());
+            while (rs.next()) {
+                DoubleTsData avgData = new DoubleTsData(rs.getLong(1), (short) 1, rs.getDouble(2));
+                DoubleTsData maxData = new DoubleTsData(rs.getLong(1), (short) 1, rs.getDouble(3));
+                DoubleTsData minData = new DoubleTsData(rs.getLong(1), (short) 1, rs.getDouble(4));
+                DoubleStatData statData = new DoubleStatData(avgData, maxData, minData);
+                result.add(statData);
+            }
+
+        } catch (Exception e) {
+            log.error(e.getMessage());
+        } finally {
+            if (connect != null)
+                connect.close();
+        }
+
+        return result;
+    }
+
+    @Override
+    public boolean writeHistoryValue(List<TsPointData> dataList) throws Exception {
+        boolean flag = false;
+        Connection connect = tdUtil.getConnect();
+
+        try {
+            Statement st = connect.createStatement();
+
+            StringBuilder sb = new StringBuilder();
+            sb.append("insert into ");
+            for (TsPointData obj : dataList) {
+                long time = obj.getTsData().getTs();
+                String point = obj.getTagName();
+                double value = obj.getTsData().getDoubleValue().get();
+                sb.append(TaosCovertUtil.coverStationPrefix(point)).append(".");
+                sb.append(point).append(" values (");
+                sb.append(time).append(",").append(value).append(")");
+            }
+
+            flag = st.execute(sb.toString());
+
+        } catch (Exception e) {
+            log.error(e.getMessage());
+        } finally {
+            if (connect != null)
+                connect.close();
+        }
+
+        return flag;
+    }
+
+    /**
+     * 查询多测点某个时间点的数值
+     * @param tsPoints 标签点
+     * @param ts       时间点（秒级）
+     *                 上一个最近的数据
+     * @return
+     * @throws Exception
+     */
+    @Override
+    public Map<String, TsData> getHistorySection(List<TsPoint> tsPoints, Long ts) throws Exception {
+        Map<String, TsData> result = new HashMap<>();
+        Map<TsDataType, List<TsPoint>> pointGroup = tsPoints.stream().collect(Collectors.groupingBy(TsPoint::getTsDataType));
+
+        //存储线程的返回值
+        List<Future<Map<String, TsData>>> results = new LinkedList<>();
+        for (Map.Entry<TsDataType, List<TsPoint>> entry : pointGroup.entrySet()) {
+            String[] tagNames = entry.getValue().stream().map(TsPoint::getId).toArray(String[]::new);
+            if (entry.getKey() == TsDataType.DOUBLE)
+                for (String tag : tagNames) {
+                    String sql = getSectionSql(tag, ts);
+
+                    log.info("5->" + sql);
+
+                    TaskCallable task = new TaskCallable(tdUtil.getConnect(), sql, ts, tag, TsDataType.DOUBLE);
+                    Future<Map<String, TsData>> submit = taskConfig.getExecutor().submit(task);
+                    results.add(submit);
+                }
+            if (entry.getKey() == TsDataType.BOOLEAN) {
+                for (String tag : tagNames) {
+                    String sql = getSectionSql(tag, ts);
+                    TaskCallable task = new TaskCallable(tdUtil.getConnect(), sql, ts, tag, TsDataType.BOOLEAN);
+                    Future<Map<String, TsData>> submit = taskConfig.getExecutor().submit(task);
+                    results.add(submit);
+                }
+            }
+        }
+        //返回结果
+        for (int i = 0; i < results.size(); i++) {
+            result.putAll(results.get(i).get());
+        }
+
+//        for (Map.Entry<TsDataType, List<TsPoint>> entry : pointGroup.entrySet()) {
+//            String[] tagNames = tsPoints.stream().map(TsPoint::getId).toArray(String[]::new);
+//            for (String tag : tagNames) {
+//                TsData tsData = null;
+//                String point = TaosCovertUtil.coverStationPrefix(tag) + "." + tag.replace(".", "_");
+//                String sql = "select last_row(*) from " + point + " where point_time>='"
+//                        + DateFormatUtils.format(ts - year_time, "yyyy-MM-dd HH:mm:ss:SSS") + "'"
+//                        + " and point_time <='" + DateFormatUtils.format(ts, "yyyy-MM-dd HH:mm:ss:SSS") + "'";
+//                ResultSet rs = config.getConnect().createStatement().executeQuery(sql);
+//                while (rs.next()) {
+//                    if (entry.getKey() == TsDataType.DOUBLE)
+//                        tsData = new DoubleTsData(ts, (short) 0, rs.getDouble(2));
+//                    if (entry.getKey() == TsDataType.BOOLEAN)
+//                        tsData = new BooleanTsData(ts, (short) 0, rs.getBoolean(2));
+//                    result.put(tag, tsData);
+//                }
+//            }
+//        }
+
+        return result;
+    }
+
+    public List<TsData> getDoubleTsDataHistory(TsQuery tsQuery) throws Exception {
+        List<TsData> tsDataList = new ArrayList<>();
+        try {
+            //存储线程的返回值
+            List<Future<List<TsData>>> results = new LinkedList<>();
+            String sql = getHistory(tsQuery);
+            log.info("2->" + sql);
+            TaskCallableList task = new TaskCallableList(tdUtil.getConnect(), sql, TsDataType.DOUBLE,tsQuery.getStartTs());
+            Future<List<TsData>> submit = taskConfig.getExecutor().submit(task);
+            results.add(submit);
+            //返回结果
+            tsDataList = results.get(0).get();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+        return tsDataList;
+    }
+
+    //DoubleTsData{ts='', status='0', value='442.01'}
+    public List<TsData> getLongTsDataHistory(TsQuery tsQuery) throws Exception {
+        List<TsData> tsDataList = new ArrayList<>();
+//        Connection connect = config.getConnect();
+
+        try {
+            //存储线程的返回值
+            List<Future<List<TsData>>> results = new LinkedList<>();
+            String sql = getHistory(tsQuery);
+            TaskCallableList task = new TaskCallableList(tdUtil.getConnect(), sql, TsDataType.DOUBLE,tsQuery.getStartTs());
+            Future<List<TsData>> submit = taskConfig.getExecutor().submit(task);
+            results.add(submit);
+
+            //返回结果
+            tsDataList = results.get(0).get();
+
+//            Statement st = connect.createStatement();
+//            String sql = getHistory(tsQuery);
+//            ResultSet rs = st.executeQuery(sql);
+//            while (rs.next()) {
+//                tsDataList.add(new LongTsData(rs.getLong(1), (short) 0, rs.getLong(2)));
+//            }
+        } catch (Exception e) {
+            log.error(e.getMessage());
+        }
+        finally {
+//            if (connect != null)
+//                connect.close();
+        }
+
+        return tsDataList;
+    }
+
+
+
+    public List<TsData> getBooleanTsDataHistory(TsQuery tsQuery) throws Exception {
+        List<TsData> tsDataList = new ArrayList<>();
+//        Connection connect = config.getConnect();
+
+        try {
+            //存储线程的返回值
+            List<Future<List<TsData>>> results = new LinkedList<>();
+            String sql = getHistory(tsQuery);
+            log.info("2->" + sql);
+            TaskCallableList task = new TaskCallableList(tdUtil.getConnect(), sql, TsDataType.DOUBLE,tsQuery.getStartTs());
+            Future<List<TsData>> submit = taskConfig.getExecutor().submit(task);
+            results.add(submit);
+
+            //返回结果
+            tsDataList = results.get(0).get();
+
+//            Statement st = connect.createStatement();
+//            String sql = getHistory(tsQuery);
+//            ResultSet rs = st.executeQuery(sql);
+//            while (rs.next()) {
+//                tsDataList.add(new BooleanTsData(rs.getLong(1), (short) 0, rs.getBoolean(2)));
+//            }
+        } catch (Exception e) {
+            log.error(e.getMessage());
+        }
+        finally {
+//            if (connect != null)
+//                connect.close();
+        }
+
+        return tsDataList;
+    }
+
+
+    public List<TsData> getStringTsDataHistory(TsQuery tsQuery) throws Exception {
+        List<TsData> tsDataList = new ArrayList<>();
+//        Connection connect = config.getConnect();
+
+        try {
+            //存储线程的返回值
+            List<Future<List<TsData>>> results = new LinkedList<>();
+            String sql = getHistory(tsQuery);
+            TaskCallableList task = new TaskCallableList(tdUtil.getConnect(), sql, TsDataType.DOUBLE,tsQuery.getStartTs());
+            Future<List<TsData>> submit = taskConfig.getExecutor().submit(task);
+            results.add(submit);
+
+            //返回结果
+            tsDataList = results.get(0).get();
+
+//            Statement st = connect.createStatement();
+//            String sql = getHistory(tsQuery);
+//            ResultSet rs = st.executeQuery(sql);
+//            while (rs.next()) {
+//                tsDataList.add(new StringTsData(rs.getLong(1), (short) 0, rs.getString(2)));
+//            }
+        } catch (Exception e) {
+            log.error(e.getMessage());
+        }
+        finally {
+//            if (connect != null)
+//                connect.close();
+        }
+
+        return tsDataList;
+    }
+
+
+    /**
+     * 拼接sql
+     *
+     * @param tsQuery
+     * @return
+     * @throws WisdomException
+     */
+    private String getHistory(TsQuery tsQuery) throws WisdomException {
+        String point = tsQuery.getTsPoint().getId();
+        StringBuilder sb = new StringBuilder();
+        if (tsQuery.getInterpolation() == Interpolation.RAW) {
+            sb.append("select * from gdnxxny.`");
+            sb.append(point);
+            sb.append("` where ts between ").append(tsQuery.getStartTs());
+            sb.append(" and ").append(tsQuery.getEndTs());
+        } else if (tsQuery.getInterpolation() == Interpolation.SNAP) {
+            if (tsQuery.getDateArray() != null && tsQuery.getDateArray().length > 0) {
+
+                sb.append("select interp(value) from gdnxxny.`");
+                sb.append(point);
+                sb.append("` ts >= ").append(tsQuery.getStartTs());
+                sb.append(" and ts<= ").append(tsQuery.getEndTs());
+                sb.append(" every(").append(tsQuery.getInterval()).append("s) fill(prev)");
+            } else {
+                log.warn("无效的查询条件！");
+                throw new WisdomException("无效的查询条件！");
+            }
+        } else {
+            log.warn("Taos不支持历史数据插值方法：" + tsQuery.getInterpolation());
+            throw new WisdomException("Taos不支持历史数据插值方法：" + tsQuery.getInterpolation());
+        }
+
+        return sb.toString();
+    }
+
+
+    /**
+     * 拼接sql
+     *
+     * @param tag
+     * @param ts
+     * @return
+     */
+    private String getSectionSql(String tag, long ts) {
+        // String point = TaosCovertUtil.coverStationPrefix(tag) + "." + tag.replace(".", "_");
+
+        StringBuilder sb = new StringBuilder();
+        // TODO： select interp(*) from xxx where _c0 = xxxx:xx:xx fill(prev）
+
+        // sb.append("select last_row(*) from ").append(point);
+//      sb.append(" where point_time <="+ts);
+        sb.append("select last_row(*) from gdnxxny.`").append(tag);
+        sb.append("` where ts>=");
+        sb.append(ts);
+//        sb.append(" fill(prev)");
+
+//        sb.append("select last_row(*) from ").append(point);
+//        sb.append(" where point_time>='");
+//        sb.append(ts - half_year_time);
+//        sb.append("'");
+//        sb.append(" and point_time <='" + DateFormatUtils.format(ts, "yyyy-MM-dd HH:mm:ss:SSS") + "'");
+
+        return sb.toString();
+    }
+
+
+    /**
+     * 将blob转化为byte[]
+     *
+     * @param blob
+     * @return
+     */
+    private byte[] getBytes(Blob blob) {
+        try {
+            InputStream ins = blob.getBinaryStream();
+            byte[] b = new byte[1024];
+            int num = 0;
+            String res = "";
+            while ((num = ins.read(b)) != -1) {
+                res += new String(b, 0, num);
+            }
+            return res.getBytes();
+        } catch (SQLException e) {
+            e.printStackTrace();
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        return null;
+    }
+}

timeseries/dao-taos/src/main/java/com/gyee/wisdom/dao/taoscz/TaosLatestDao.java

@@ -0,0 +1,349 @@
+package com.gyee.wisdom.dao.taoscz;
+
+import com.gyee.wisdom.common.data.timeseries.*;
+import com.gyee.wisdom.dao.timeseries.ILatestDao;
+import com.gyee.wisdom.dao.timeseries.TaosDao;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Component
+@TaosDao
+public class TaosLatestDao implements ILatestDao {
+
+    @Autowired
+    private TaosConfiguration config;
+    @Autowired
+    private ThreadPoolTaskConfig taskConfig;
+
+    /**
+     * 查询多测点最新数据
+     * @param tsPoints
+     * @return
+     * @throws Exception
+     */
+    @Override
+    public Map<String, TsData> getTsDataLatest(List<TsPoint> tsPoints) throws Exception {
+        Map<String, TsData> result = new HashMap<>();
+        Map<TsDataType, List<TsPoint>> pointGroup = tsPoints.stream().collect(Collectors.groupingBy(TsPoint::getTsDataType));
+        for (Map.Entry<TsDataType, List<TsPoint>> entry : pointGroup.entrySet()) {
+            String[] tagNames = entry.getValue().stream().map(TsPoint::getId).toArray(String[]::new);
+
+            if(entry.getKey() == TsDataType.DOUBLE) {
+                result.putAll(getDoubleTsDataSnapshots(tagNames));
+            } else  {
+                result.putAll(getBooleanTsDataSnapshots(tagNames));
+            }
+        }
+
+        return result;
+    }
+
+    public Map<String, TsData> getTsDataLatestOneUniformCode(TsPoint tsPoint) throws Exception {
+        String sql = "SELECT last_row(*) from gdnxxny.`" + tsPoint.getId() + "`";
+        final ResultSet rs = config.getConnect().createStatement().executeQuery(sql);
+        rs.next();
+        Map<String, TsData> result = new HashMap<>();
+        result.put(tsPoint.getUniformCode(), new DoubleTsData(rs.getLong(1), (short) 0, rs.getDouble(2)));
+        return result;
+    }
+
+    public Map<String, TsData> getTsDataLatestBatchOneUniformCode(List<TsPoint> pointList) throws Exception {
+        final StringBuilder sql = new StringBuilder("SELECT last_row(*) from stag_double where tbname in (");
+
+        Map<String, String> pointMap = new HashMap<>();
+        for (TsPoint tsPoint : pointList) {pointMap.put(tsPoint.getId(), tsPoint.getUniformCode());}
+        for (Map.Entry<String, String> entry : pointMap.entrySet()) {
+            sql.append("'");
+            sql.append(entry.getKey());
+            sql.append("',");
+        }
+        sql.append("') GROUP BY tbname");
+        final ResultSet rs = config.getConnect().createStatement().executeQuery(sql.toString());
+        Map<String, TsData> result = new HashMap<>();
+        while (rs.next()) {
+            result.put(pointMap.get(rs.getString(3)), new DoubleTsData(rs.getLong(1), (short) 0, rs.getDouble(2)));
+        }
+        return result;
+    }
+
+    public String joinSql(final int startIndex, final int endIndex, List<String> tagNameList) {
+        StringBuilder sb = new StringBuilder("SELECT last_row(*) FROM stag_double where tbname in ('");
+        sb.append(tagNameList.get(startIndex));
+        sb.append("'");
+        for (int i = startIndex + 1; i < endIndex; i++) {
+            sb.append(",'");
+            sb.append(tagNameList.get(i));
+            sb.append("'");
+        }
+        sb.append(") GROUP BY tbname");
+        return sb.toString();
+    }
+
+    @Override
+    public int writeDoubleLatest(List<TsPointData> list) throws Exception {
+        boolean flag = false;
+        Connection connection = config.getConnect();
+
+        try {
+            Statement st = connection.createStatement();
+
+            StringBuilder sb = new StringBuilder();
+            sb.append("insert into ");
+            for (TsPointData obj : list) {
+                long time = obj.getTsData().getTs();
+                String point = obj.getTagName();
+                double value = obj.getTsData().getDoubleValue().get();
+
+                sb.append("`").append(point).append("` USING stag_double (organization) TAGS(")
+                        .append(obj.getTagName().substring(0,obj.getTagName().indexOf(".")))
+                        .append(") values (").append(time).append(",").append(value).append(")");
+            }
+            sb.append(";");
+            flag = st.execute(sb.toString());
+
+        } catch (Exception e) {
+        } finally {
+            if (connection != null)
+                connection.close();
+        }
+
+        return flag == true ? list.size() : 0;
+    }
+
+    @Override
+    public int writeStringLatest(List<TsPointData> list) throws Exception {
+        boolean flag = false;
+        Connection connection = config.getConnect();
+
+        try {
+            Statement st = connection.createStatement();
+
+            StringBuilder sb = new StringBuilder();
+            sb.append("insert into ");
+            for (TsPointData obj : list) {
+                long time = obj.getTsData().getTs();
+                String point = obj.getTagName();
+                String value = obj.getTsData().getStringValue().get();
+                sb.append(TaosCovertUtil.coverStationPrefix(point)).append(".");
+                sb.append(point).append(" values (");
+                sb.append(time).append(",'").append(value).append("') ");
+            }
+
+            flag = st.execute(sb.toString());
+
+        } catch (Exception e) {
+        } finally {
+            if (connection != null)
+                connection.close();
+        }
+
+        return flag == true ? list.size() : 0;
+    }
+
+    @Override
+    public int writeBooleanLatest(List<TsPointData> list) throws Exception {
+        boolean flag = false;
+        Connection connection = config.getConnect();
+
+        try {
+            Statement st = connection.createStatement();
+
+            StringBuilder sb = new StringBuilder();
+            sb.append("insert into ");
+            for (TsPointData obj : list) {
+                long time = obj.getTsData().getTs();
+                String point = obj.getTagName();
+                boolean value = obj.getTsData().getBooleanValue().get();
+                sb.append(TaosCovertUtil.coverStationPrefix(point)).append(".");
+                sb.append(point).append(" values (");
+                sb.append(time).append(",").append(value).append(")");
+            }
+
+            flag = st.execute(sb.toString());
+
+        } catch (Exception e) {
+        } finally {
+            if (connection != null)
+                connection.close();
+        }
+
+        return flag == true ? list.size() : 0;
+    }
+
+    @Override
+    public int writeLongLatest(List<TsPointData> list) throws Exception {
+        boolean flag = false;
+        Connection connection = config.getConnect();
+
+        try {
+            Statement st = connection.createStatement();
+
+            StringBuilder sb = new StringBuilder();
+            sb.append("insert into ");
+            for (TsPointData obj : list) {
+                long time = obj.getTsData().getTs();
+                String point = obj.getTagName();
+                long value = obj.getTsData().getLongValue().get();
+                sb.append(TaosCovertUtil.coverStationPrefix(point)).append(".");
+                sb.append(point).append(" values (");
+                sb.append(time).append(",").append(value).append(")");
+            }
+
+            flag = st.execute(sb.toString());
+
+        } catch (Exception e) {
+        } finally {
+            if (connection != null)
+                connection.close();
+        }
+
+        return flag == true ? list.size() : 0;
+    }
+
+    @Override
+    public int writeBlobLatest(List<TsPointData> list) throws Exception {
+        return 0;
+    }
+
+    @Override
+    public int writeCoordinateLatest(List<TsPointData> list) throws Exception {
+        return 0;
+    }
+
+    /**
+     * 由于taos数据库中全部是DOUBLE类型，故不需要区分类型
+     *
+     * @param dataList
+     * @return
+     * @throws Exception
+     */
+    @Override
+    public boolean writeLatest(List<TsPointData> dataList) throws Exception {
+        int writeCount = 0;
+
+        Map<TsDataType, List<TsPointData>> pointGroup = dataList.stream().collect(Collectors.groupingBy(TsPointData::findDataType));
+        for (Map.Entry<TsDataType, List<TsPointData>> entry : pointGroup.entrySet()) {
+            List<TsPointData> pointDataList = entry.getValue();
+            int count = 0;
+            if(entry.getKey()==TsDataType.DOUBLE){
+                count = writeDoubleLatest(pointDataList);
+            }else if(entry.getKey()==TsDataType.BOOLEAN){
+                count = writeIntLatest(pointDataList);
+            }
+            writeCount = writeCount + count;
+        }
+        return writeCount > 0 ? true : false;
+    }
+
+    public int writeIntLatest(List<TsPointData> list) throws SQLException {
+        boolean flag = false;
+        Connection connection = config.getConnect();
+
+        try {
+            Statement st = connection.createStatement();
+
+            StringBuilder sb = new StringBuilder();
+            sb.append("insert into ");
+            for (TsPointData obj : list) {
+                long time = obj.getTsData().getTs();
+                String point = obj.getTagName();
+                int value = obj.getTsData().getBooleanValue().get()?1:0;
+
+                sb.append("`").append(point).append("` USING stag_double (organization) TAGS(")
+                        .append(obj.getTagName().substring(0,obj.getTagName().indexOf(".")))
+                        .append(") values (").append(time).append(",").append(value).append(")");
+            }
+            sb.append(";");
+            flag = st.execute(sb.toString());
+
+        } catch (Exception e) {
+        } finally {
+            if (connection != null)
+                connection.close();
+        }
+
+        return flag == true ? list.size() : 0;
+    }
+
+    public Map<String, TsData> getDoubleTsDataSnapshots(String... tagNames) throws Exception {
+        Map<String, TsData> tsDataMap = new HashMap<>();
+
+        //存储线程的返回值
+        List<Future<Map<String, TsData>>> results = new LinkedList<>();
+
+        for (String tag : tagNames) {
+            // String point = TaosCovertUtil.coverStationPrefix(tag) + "." + tag.replace(".", "_");
+            String  sql = "select last_row(*) from gdnxxny.`" + tag +"`";
+
+            TaskCallable task = new TaskCallable(config.getConnect(), sql, 0, tag, TsDataType.DOUBLE);
+            Future<Map<String, TsData>> submit = taskConfig.getExecutor().submit(task);
+            results.add(submit);
+        }
+
+        //返回结果
+        for (int i = 0; i < results.size(); i++) {
+            tsDataMap.putAll(results.get(i).get());
+        }
+
+        return tsDataMap;
+    }
+
+    public Map<String, TsData> getLongTsDataSnapshots(String... tagNames) throws Exception {
+        Map<String, TsData> tsDataMap = new HashMap<>();
+
+        //存储线程的返回值
+        List<Future<Map<String, TsData>>> results = new LinkedList<>();
+
+        for (String tag : tagNames) {
+            // String point = TaosCovertUtil.coverStationPrefix(tag) + "." + tag.replace(".", "_");
+            String sql = "select last_row(*) from " + tag;
+            TaskCallable task = new TaskCallable(config.getConnect(), sql, 0, tag, TsDataType.LONG);
+            Future<Map<String, TsData>> submit = taskConfig.getExecutor().submit(task);
+            results.add(submit);
+        }
+
+        //返回结果
+        for (int i = 0; i < results.size(); i++) {
+            tsDataMap.putAll(results.get(i).get());
+        }
+
+        return tsDataMap;
+    }
+
+    public Map<String, TsData> getBooleanTsDataSnapshots(String... tagNames) throws Exception {
+        Map<String, TsData> tsDataMap = new HashMap<>();
+
+        //存储线程的返回值
+        List<Future<Map<String, TsData>>> results = new LinkedList<>();
+
+        for (String tag : tagNames) {
+//            String point = TaosCovertUtil.coverStationPrefix(tag) + "." + tag.replace(".", "_");
+            String sql = "select last_row(*) from gdnxxny.`" + tag +"`";
+            log.info("1->" + sql);
+            TaskCallable task = new TaskCallable(config.getConnect(), sql, 0, tag, TsDataType.BOOLEAN);
+            Future<Map<String, TsData>> submit = taskConfig.getExecutor().submit(task);
+            results.add(submit);
+        }
+
+        //返回结果
+        for (int i = 0; i < results.size(); i++) {
+            tsDataMap.putAll(results.get(i).get());
+        }
+
+        return tsDataMap;
+    }
+
+}

timeseries/dao-taos/src/main/java/com/gyee/wisdom/dao/taoscz/TaskCallable.java

@@ -0,0 +1,86 @@
+package com.gyee.wisdom.dao.taoscz;
+
+import com.gyee.wisdom.common.data.timeseries.*;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.util.StringUtils;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+@Slf4j
+public class TaskCallable implements Callable {
+
+    private Connection connection;
+    private String sql;
+    private long time;
+    private String tagName;
+    private TsDataType type;
+
+    public TaskCallable(Connection connection, String sql, long time, String tagName, TsDataType type) {
+        this.connection = connection;
+        this.sql = sql;
+        this.time = time;
+        this.tagName = tagName;
+        this.type = type;
+    }
+
+
+    @Override
+    public Map<String, TsData> call() throws SQLException {
+        TsData tsData = null;
+        Statement st = null;
+        ResultSet rs = null;
+        Map<String, TsData> result = new HashMap<>();
+
+        if (!StringUtils.hasText(tagName))
+            return result;
+
+        try {
+            st = this.connection.createStatement();
+            rs = st.executeQuery(this.sql);
+
+            if (rs.next()) {
+                if (this.time <= 0)
+                    this.time = rs.getLong(1);
+
+                if (this.type == TsDataType.DOUBLE)
+                    tsData = new DoubleTsData(this.time, (short) 0, rs.getDouble(2));
+                else if (this.type == TsDataType.BOOLEAN)
+                    tsData = new BooleanTsData(this.time, (short) 0, rs.getBoolean(2));
+                else if (this.type == TsDataType.STRING)
+                    tsData = new StringTsData(this.time, (short) 0, rs.getString(2));
+                else if (this.type == TsDataType.LONG)
+                    tsData = new LongTsData(this.time, (short) 0, rs.getLong(2));
+            } else {
+                if (this.type == TsDataType.DOUBLE)
+                    tsData = new DoubleTsData(this.time, (short) 0, 0);
+                else if (this.type == TsDataType.BOOLEAN)
+                    tsData = new BooleanTsData(this.time, (short) 0, false);
+                else if (this.type == TsDataType.STRING)
+                    tsData = new StringTsData(this.time, (short) 0, "0");
+                else if (this.type == TsDataType.LONG)
+                    tsData = new LongTsData(this.time, (short) 0, 0);
+            }
+
+            result.put(this.tagName, tsData);
+
+        } catch (Exception e) {
+            log.error(e.getMessage());
+        } finally {
+            if (rs != null)
+                rs.close();
+            if (st != null)
+                st.close();
+            if (this.connection != null)
+                this.connection.close();
+        }
+
+        return result;
+    }
+
+}

timeseries/dao-taos/src/main/java/com/gyee/wisdom/dao/taoscz/TaskCallableList.java

@@ -0,0 +1,93 @@
+package com.gyee.wisdom.dao.taoscz;
+
+import com.gyee.wisdom.common.data.timeseries.*;
+import lombok.extern.slf4j.Slf4j;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+
+@Slf4j
+public class TaskCallableList implements Callable {
+
+    private Connection connection;
+    private String sql;
+    private TsDataType type;
+    private long startTs;
+
+    public TaskCallableList(Connection connection, String sql, TsDataType type, long startTs) {
+        this.connection = connection;
+        this.sql = sql;
+        this.type = type;
+        this.startTs = startTs;
+    }
+
+
+    @Override
+    public List<TsData> call() throws SQLException {
+        Statement st = null;
+        ResultSet rs = null;
+
+        List<TsData> tsDataList = new ArrayList<>();
+        try {
+            st = this.connection.createStatement();
+            rs = st.executeQuery(this.sql);
+
+            long timeShift = 0;
+            while (rs.next()) {
+                long ts = rs.getLong(1);
+                if (timeShift==0) {
+                    timeShift = this.startTs - ts;
+                }
+                ts += timeShift;
+                TsData tsData = null;
+                if (this.type == TsDataType.DOUBLE) tsData = new DoubleTsData(ts, (short) 0, rs.getDouble(2));
+                else if (this.type == TsDataType.BOOLEAN) tsData = new BooleanTsData(ts, (short) 0, rs.getBoolean(2));
+                else if (this.type == TsDataType.STRING) tsData = new StringTsData(ts, (short) 0, rs.getString(2));
+                else if (this.type == TsDataType.LONG) tsData = new LongTsData(ts, (short) 0, rs.getLong(2));
+
+                if (tsData == null){
+                    if (this.type == TsDataType.DOUBLE)
+                        tsData = new DoubleTsData(System.currentTimeMillis(), (short) 0, 0);
+                    else if (this.type == TsDataType.BOOLEAN)
+                        tsData = new BooleanTsData(System.currentTimeMillis(), (short) 0, false);
+                    else if (this.type == TsDataType.STRING)
+                        tsData = new StringTsData(System.currentTimeMillis(), (short) 0, "0");
+                    else if (this.type == TsDataType.LONG)
+                        tsData = new LongTsData(System.currentTimeMillis(), (short) 0, 0);
+                }
+
+                tsDataList.add(tsData);
+            }
+
+        } catch (Exception e) {
+            log.error(e.getMessage());
+            log.error(this.sql);
+            create(this.sql, st);
+        } finally {
+            if (rs != null)
+                rs.close();
+            if (st != null)
+                st.close();
+            if (this.connection != null)
+                this.connection.close();
+        }
+
+        return tsDataList;
+    }
+
+    private void create(String sql, Statement st) throws SQLException {
+        String name = sql.substring(sql.lastIndexOf("from") + 5).trim();
+
+        String dbname = name.substring(0, name.indexOf("."));
+        String stbname = name.substring(name.indexOf(".") + 1, name.indexOf("_"));
+        String ztbname = name.substring(name.indexOf(".") + 1);
+
+        st.execute("CREATE TABLE IF NOT EXISTS " + dbname + "." + ztbname + " USING " +
+                dbname + "." + stbname + " TAGS(\"" + ztbname + "\")");
+    }
+}

timeseries/dao-taos/src/main/java/com/gyee/wisdom/dao/taoscz/ThreadPoolTaskConfig.java

@@ -0,0 +1,56 @@
+package com.gyee.wisdom.dao.taoscz;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.concurrent.ThreadPoolExecutor;
+
+@Configuration
+public class ThreadPoolTaskConfig {
+    /**
+     *   默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，
+     *	当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中；
+     *  当队列满了，就继续创建线程，当线程数量大于等于maxPoolSize后，开始使用拒绝策略拒绝
+     */
+
+    /** 核心线程数（默认线程数） */
+    private static final int corePoolSize = 100;
+    /** 最大线程数 */
+    private static final int maxPoolSize = 500;
+    /** 允许线程空闲时间（单位：默认为秒） */
+    private static final int keepAliveTime = 60;
+    /** 缓冲队列大小 */
+    private static final int queueCapacity = 500;
+    /** 允许等待最长时间 */
+    private static final int awaitTime = 10;
+    /** 线程池名前缀 */
+    private static final String threadNamePrefix = "GYEE-Thread-";
+
+    private ThreadPoolTaskExecutor executor;
+
+    public ThreadPoolTaskExecutor getExecutor(){
+        if (executor == null)
+            executor = taskExecutor();
+
+        return executor;
+    }
+
+
+    private ThreadPoolTaskExecutor taskExecutor(){
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(corePoolSize);
+        executor.setMaxPoolSize(maxPoolSize);
+        executor.setQueueCapacity(queueCapacity);
+        executor.setKeepAliveSeconds(keepAliveTime);
+        executor.setThreadNamePrefix(threadNamePrefix);
+        executor.setAwaitTerminationSeconds(awaitTime);
+
+        // 线程池对拒绝任务的处理策略
+        // CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+        // 初始化
+        executor.initialize();
+        return executor;
+    }
+
+}

timeseries/dao-taoscz/build.gradle

@@ -21,8 +21,8 @@ dependencies {
     implementation project(":common:data")
     implementation project(":timeseries:dao-interface")
     implementation fileTree(dir: 'src/main/lib', include: '*.jar')
-    implementation("$bootGroup:spring-boot-starter-log4j2")
-    implementation ("com.taosdata.jdbc:taos-jdbcdriver:$taosVersion"){
+//    implementation("$bootGroup:spring-boot-starter-log4j2")
+    implementation ("com.taosdata.jdbc:taos-jdbcdriver:$taosVersion3"){
         exclude group: 'com.alibaba', module: 'fastjson'
     }
 }

timeseries/data-adapter/build.gradle

@@ -34,15 +34,15 @@ dependencies {
     //compile("$bootGroup:spring-boot-starter-session")
     implementation("$bootGroup:spring-boot-starter-websocket")
     implementation("$bootGroup:spring-boot-starter-undertow")
-    implementation("org.apache.logging.log4j:log4j-core:$log4jVersion")
-    implementation("org.apache.logging.log4j:log4j-jul:$log4jVersion")
-    implementation("org.apache.logging.log4j:log4j-api:$log4jVersion")
-    implementation("org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion")
+//    implementation("org.apache.logging.log4j:log4j-core:$log4jVersion")
+//    implementation("org.apache.logging.log4j:log4j-jul:$log4jVersion")
+//    implementation("org.apache.logging.log4j:log4j-api:$log4jVersion")
+//    implementation("org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion")
     //compile('org.springframework.boot:spring-boot-starter-data-redis')
     //implementation "org.springframework.boot:spring-boot-starter-redis:1.3.5.RELEASE"
     //compile('io.moquette:moquette-broker:0.10')
     implementation('com.fasterxml.jackson.datatype:jackson-datatype-jdk8')
-    implementation ("com.taosdata.jdbc:taos-jdbcdriver:$taosVersion")
+    //implementation ("com.taosdata.jdbc:taos-jdbcdriver:$taosVersion")
     implementation 'org.postgresql:postgresql:42.4.0'
 
 }

timeseries/data-adapter/src/main/resources/application-nx.yaml

@@ -1,5 +1,5 @@
 server:
-  port: 8012
+  port: 8011
   max-http-header-size: 128KB
 
 
@@ -104,12 +104,16 @@ hadoop:
 
 #taos数据库
 taos:
-  server_ip: 192.168.2.252
-  server_port: 6030
+  server_ip: 18.6.30.71
+  #server_ip: 192.168.1.67
+  #restful
+  server_port: 6041
+  #jni
+  #server_port: 6030
   user_name: root
   password: taosdata
-  pool_size: 10
-  max_pool_size: 100
+  pool_size: 500
+  max_pool_size: 2000
 
 #适配器链接taos数据库配置信息
 taoscz:

timeseries/data-adapter/src/main/resources/application.yaml

@@ -36,7 +36,7 @@ spring:
           test-on-borrow: false
           test-on-return: false
   profiles:
-    active: nxxnj
+    active: nx
 #  redis:
 #    database: 2
 #    host: 192.168.2.5
@@ -69,7 +69,7 @@ knife4j:
     type: Simple
 
 timeseries:
-  db-type: taoscz #"${DATABASE_TYPE:sql}" # cassandra/kairosDB/hbase/opentsDB/influxDB/TiDB
+  db-type: taos #"${DATABASE_TYPE:sql}" # cassandra/kairosDB/hbase/opentsDB/influxDB/TiDB
   #db-type: hadoop #"${DATABASE_TYPE:sql}" # cassandra/kairosDB/hbase/opentsDB/influxDB/TiDB
 #golden 数据库信息
 golden: