动态支持MySQL/Oracle

config/application-standone.yml

+from-db:
+  url: jdbc:oracle:thin:@192.168.10.236:1521:orcl
+  user: C##FMMP
+  password: vcare~1(^_^)
+to-db:
+  url: jdbc:mysql://192.168.10.214:3306/fmmp?serverTimezone=Asia/Shanghai
+  user: fmmp
+  password: E_wsq@163.com
+export:
+  folder: ./logs/

config/application.yml

 profile:
 #  active: mysql
-   active: oracle
+#   active: oracle
+   active: standone

pom.xml

             <version>1.0.2</version>
         </dependency>
 
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <version>5.5.2</version>
+            <scope>test</scope>
+        </dependency>
+
+
     </dependencies>
 
 

src/main/java/com/vcarecity/cvs/FileExporterApp.java

         }
         Class<?> tableClass = ReflectionUtil.getTableClass(table);
 
-        String basePath = table;
-        final AppProperties.Export export = appProperties.getExport();
-        if (export != null) {
-            basePath = export.getFolder() + table;
+        boolean isMysql;
+        Object properties;
+
+        if (isImportToMysql(args)) {
+            properties = "mysql";
+            isMysql = true;
+        } else {
+
+            String basePath = table;
+            final AppProperties.Export export = appProperties.getExport();
+            if (export != null) {
+                basePath = export.getFolder() + table;
+            }
+            properties = basePath;
+            isMysql = false;
         }
 
-        injector = Guice.createInjector(new PropertiesModule(appProperties), new SQLModule());
+
+        injector = Guice.createInjector(new PropertiesModule(appProperties), new SQLModule(isMysql));
 
         final SQLStarter instance = injector.getInstance(SQLStarter.class);
-        instance.exportData(table, basePath, tableClass);
+        instance.exportDataAndConvert(table, properties, tableClass);
     }
 
+
+    private static boolean isImportToMysql(String[] args) {
+        for (String arg : args) {
+            if (arg.startsWith("--type=")) {
+                return !"csv".equalsIgnoreCase(arg.substring("--type=".length()));
+            }
+        }
+        return true;
+    }
+
+
     private static String getTableByArgs(String[] args) {
         for (String arg : args) {
             if (arg.startsWith("--table=")) {

src/main/java/com/vcarecity/cvs/event/MySQLImportEvent.java

 package com.vcarecity.cvs.event;
 
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.inject.Inject;
-import com.vcarecity.cvs.core.ColumnUpdateMapper;
 import com.vcarecity.cvs.properties.AppProperties;
 import com.vcarecity.cvs.properties.DbProperties;
-import com.vcarecity.cvs.shell.MySQLImporter;
 import lombok.extern.slf4j.Slf4j;
 
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
     public void exportSuccess(String filename, String table, String[] header) {
         logger.info("MySQLImportEvent.exportSuccess. filename = {}", filename);
 
-        ThreadFactory factory = new ThreadFactoryBuilder().setNameFormat("export-" + (callIndex.incrementAndGet()) + "+-%d").build();
+        //ThreadFactory factory = new ThreadFactoryBuilder().setNameFormat("export-" + (callIndex.incrementAndGet()) + "+-%d").build();
 
-        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
-                new LinkedBlockingQueue<>(1024), factory, new ThreadPoolExecutor.AbortPolicy());
+        //ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
+        //        new LinkedBlockingQueue<>(1024), factory, new ThreadPoolExecutor.AbortPolicy());
 
         // 列名不一样时,转换
-        final String[] newHeader = ColumnUpdateMapper.updateColumnName(table, header);
-        executor.execute(new MySQLImporter(dbProperties, filename, table, newHeader));
-        executor.shutdown();
+        //final String[] newHeader = ColumnUpdateMapper.updateColumnName(table, header);
+        //executor.execute(new MySQLImporter(dbProperties, filename, table, newHeader));
+        //executor.shutdown();
     }
 
 

src/main/java/com/vcarecity/cvs/factory/ResultHandlerFactory.java

      * @param properties
      * @return
      */
-    ResultHandlerService createResultHandler(String properties);
+    ResultHandlerService createResultHandler(Object properties);
 
 }

src/main/java/com/vcarecity/cvs/module/SQLModule.java

 import com.vcarecity.cvs.properties.DbProperties;
 import com.vcarecity.cvs.service.ResultHandlerService;
 import com.vcarecity.cvs.service.SQLQueryService;
-import com.vcarecity.cvs.service.impl.CSVResultHandlerServiceImpl;
-import com.vcarecity.cvs.service.impl.MySQLQueryServiceImpl;
-import com.vcarecity.cvs.service.impl.OracleQueryServiceImpl;
-import com.zaxxer.hikari.HikariConfig;
+import com.vcarecity.cvs.service.impl.*;
+import com.vcarecity.cvs.util.CreateDatasource;
 import com.zaxxer.hikari.HikariDataSource;
 
 /**
  */
 
 public class SQLModule extends AbstractModule {
+    private final boolean isMysql;
+
+    public SQLModule(boolean isMysql) {
+        this.isMysql = isMysql;
+    }
+
     @Override
     protected void configure() {
 
         bind(ExportEventListener.class).to(MySQLImportEvent.class);
 
+        Class<? extends ResultHandlerService> resultClass;
+        if (isMysql) {
+            // resultClass = MySQLResultStatHandlerServiceImpl.class;
+            resultClass = MySQLResultHandlerServiceImpl.class;
+        } else {
+            resultClass = CSVResultHandlerServiceImpl.class;
+        }
+
         install(new FactoryModuleBuilder()
-                .implement(ResultHandlerService.class, CSVResultHandlerServiceImpl.class)
+                .implement(ResultHandlerService.class, resultClass)
                 .build(ResultHandlerFactory.class));
 
     }
         // @see https://github.com/brettwooldridge/HikariCP/wiki/MySQL-Configuration
 
         final DbProperties database = properties.getFromDb();
-        HikariConfig config = new HikariConfig();
-        config.setJdbcUrl(database.getUrl());
-        config.setUsername(database.getUser());
-        config.setPassword(database.getPassword());
-
-        config.addDataSourceProperty("cachePrepStmts", "true");
-        config.addDataSourceProperty("prepStmtCacheSize", "250");
-        config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
-        config.addDataSourceProperty("useServerPrepStmts", "true");
-        config.addDataSourceProperty("useLocalSessionState", "true");
-        config.addDataSourceProperty("rewriteBatchedStatements", "true");
-        config.addDataSourceProperty("cacheResultSetMetadata", "true");
-        config.addDataSourceProperty("cacheServerConfiguration", "true");
-        config.addDataSourceProperty("elideSetAutoCommits", "true");
-        config.addDataSourceProperty("maintainTimeStats", "false");
-
-        final HikariDataSource hikariDataSource = new HikariDataSource(config);
+        final HikariDataSource hikariDataSource = CreateDatasource.createDataSource(database);
         Runtime.getRuntime().addShutdownHook(new Thread(hikariDataSource::close));
         return hikariDataSource;
     }

src/main/java/com/vcarecity/cvs/script/BuildPartitionSQL.java

 
     public static void main(String[] args) {
         BuildPartitionSQL buildPartitionSQL = new BuildPartitionSQL();
-        buildPartitionSQL.partitionFirstSql("T_CHECK_RECORD_1", "stamp", "2016-08-23", "2019-11-21", PartitionType.MONTH);
+        buildPartitionSQL.partitionFirstSql("T_ALARM_LAST", "stamp", "2016-03-29", "2020-01-01", PartitionType.MONTH);
     }
 
 

src/main/java/com/vcarecity/cvs/service/impl/AbstractSQLQueryService.java

 
         List<T> list = resultMapper(table, resultSet, clazz);
 
-        logger.debug("query success. size = {}. usageTime = {} /ms", list.size(), (System.currentTimeMillis() - startTime));
+        logger.info("QUERY SUCCESS. size = {}. usageTime = {} /ms", list.size(), (System.currentTimeMillis() - startTime));
 
         resultSet.close();
         preparedStatement.close();

src/main/java/com/vcarecity/cvs/service/impl/CSVResultHandlerServiceImpl.java

     private final ExportEventListener eventListener;
 
     @Inject
-    public CSVResultHandlerServiceImpl(ExportEventListener eventListener, @Assisted String filePattern) {
+    public CSVResultHandlerServiceImpl(ExportEventListener eventListener,
+                                       @Assisted Object filePattern) {
         this.eventListener = eventListener;
-        this.filePattern = filePattern;
+        this.filePattern = (String) filePattern;
         this.csvFormat = CSVFormat.MYSQL;
         this.counter = 0;
     }

src/main/java/com/vcarecity/cvs/service/impl/MySQLResultHandlerServiceImpl.java

 
 import com.google.inject.Inject;
 import com.google.inject.assistedinject.Assisted;
-import com.vcarecity.cvs.event.ExportEventListener;
+import com.vcarecity.cvs.core.ColumnUpdateMapper;
+import com.vcarecity.cvs.properties.AppProperties;
 import com.vcarecity.cvs.service.ResultHandlerService;
+import com.vcarecity.cvs.util.CreateDatasource;
 import com.zaxxer.hikari.HikariDataSource;
 import lombok.extern.slf4j.Slf4j;
 
 import java.sql.Connection;
-import java.time.LocalDateTime;
+import java.sql.PreparedStatement;
+import java.util.Arrays;
 import java.util.List;
 
 /**
 @Slf4j
 public class MySQLResultHandlerServiceImpl implements ResultHandlerService {
 
-    private final HikariDataSource dataSource;
-    private final ExportEventListener eventListener;
     private final String sql;
+    private final HikariDataSource dataSource;
 
     @Inject
-    public MySQLResultHandlerServiceImpl(HikariDataSource dataSource, ExportEventListener eventListener, @Assisted String sql) {
-        this.dataSource = dataSource;
-        this.eventListener = eventListener;
-        this.sql = sql;
+    public MySQLResultHandlerServiceImpl(AppProperties properties, @Assisted Object sql) {
+        this.sql = (String) sql;
+        this.dataSource = CreateDatasource.createDataSource(properties.getToDb());
     }
 
 
     @Override
     public <T> void resultHandler(String table, String[] header, List<T> dataList) throws Exception {
-        logger.debug("start write to mysql size = {}, on = {}", dataList.size(), LocalDateTime.now());
+
+        final String sql = getSql(table, header);
+
+
         long startTime = System.currentTimeMillis();
+
+        logger.debug("{},{}", dataList.size(), sql);
         final Connection connection = dataSource.getConnection();
+        try {
+            connection.setAutoCommit(false);
+            PreparedStatement ps = connection.prepareStatement(sql);
+
+            int index = 0;
+            for (T t : dataList) {
+                index++;
+                if (t.getClass().isArray()) {
+                    Object[] rows = (Object[]) t;
+                    for (int i = 0; i < rows.length; i++) {
+                        ps.setObject(i + 1, rows[i]);
+                    }
+                } else if (List.class.isAssignableFrom(t.getClass())) {
+                    List rows = (List) t;
+                    for (int i = 0; i < rows.size(); i++) {
+                        ps.setObject(i + 1, rows.get(i));
+                    }
+                }
+                ps.addBatch();
+                //if (index % 1000 == 0) {
+                //    ps.executeBatch();
+                //    connection.commit();
+                //}
+            }
+            ps.executeBatch();
+            connection.commit();
+            ps.close();
+        } finally {
+            connection.close();
+        }
+        logger.info("WRITE SUCCESS to mysql usage time {} /ms, size = {}", System.currentTimeMillis() - startTime, dataList.size());
+    }
+
+
+    private static String getSql(String table, String[] headers) {
+
+        final String[] newHeaders = ColumnUpdateMapper.updateColumnName(table, headers);
+        final String column = String.join(",", newHeaders);
 
+        String[] fill = new String[headers.length];
+        Arrays.fill(fill, "?");
+        final String join = String.join(",", fill);
 
-        connection.close();
+        return "INSERT INTO " + table + "(" + column + ") VALUES(" + join + ")";
 
-        logger.debug("write to mysql usage time {} /ms", System.currentTimeMillis() - startTime);
     }
 }

src/main/java/com/vcarecity/cvs/service/impl/MySQLResultStatHandlerServiceImpl.java

+package com.vcarecity.cvs.service.impl;
+
+import com.google.inject.Inject;
+import com.google.inject.assistedinject.Assisted;
+import com.vcarecity.cvs.core.ColumnUpdateMapper;
+import com.vcarecity.cvs.properties.AppProperties;
+import com.vcarecity.cvs.service.ResultHandlerService;
+import com.vcarecity.cvs.util.CreateDatasource;
+import com.zaxxer.hikari.HikariDataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import java.sql.Connection;
+import java.sql.Date;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.util.List;
+
+/**
+ * @author Kerry on 19/11/21
+ */
+
+
+@Slf4j
+public class MySQLResultStatHandlerServiceImpl implements ResultHandlerService {
+
+
+    private final String sql;
+    private final HikariDataSource dataSource;
+
+    @Inject
+    public MySQLResultStatHandlerServiceImpl(AppProperties properties, @Assisted Object sql) {
+        this.sql = (String) sql;
+        this.dataSource = CreateDatasource.createDataSource(properties.getToDb());
+    }
+
+
+    @Override
+    public <T> void resultHandler(String table, String[] header, List<T> dataList) throws Exception {
+
+
+        long startTime = System.currentTimeMillis();
+
+        StringBuffer sb = new StringBuffer();
+
+        final String[] newHeaders = ColumnUpdateMapper.updateColumnName(table, header);
+        final String column = String.join(",", newHeaders);
+
+
+        sb.append("INSERT INTO ").append(table).append("(").append(String.join(",", column)).append(") VALUES ");
+
+        for (T t : dataList) {
+            if (t.getClass().isArray()) {
+                Object[] rows = (Object[]) t;
+                sb.append("(");
+                for (int i = 0; i < header.length; i++) {
+
+                    Object row = rows[i];
+                    if (row instanceof String) {
+                        sb.append("'").append(row).append("'");
+                    } else if (row instanceof Date) {
+                        sb.append("'").append(row).append("'");
+                    } else if (row instanceof Timestamp) {
+                        sb.append("'").append(row).append("'");
+                    } else {
+                        sb.append(row);
+                    }
+                    if (i != header.length - 1) {
+                        sb.append(",");
+                    }
+                }
+                sb.append("),");
+            }
+        }
+        final StringBuffer stringBuffer = sb.deleteCharAt(sb.length() - 1);
+
+        // logger.debug("{}", stringBuffer.toString());
+
+        final Connection connection = dataSource.getConnection();
+
+        final Statement statement = connection.createStatement();
+        statement.execute(stringBuffer.toString());
+        statement.close();
+        connection.close();
+
+        logger.info("WRITE SUCCESS to mysql usage time {} /ms, size = {}", System.currentTimeMillis() - startTime, dataList.size());
+    }
+
+
+}

src/main/java/com/vcarecity/cvs/starter/SQLStarter.java

 package com.vcarecity.cvs.starter;
 
+import com.vcarecity.cvs.core.ColumnUpdateMapper;
 import com.vcarecity.cvs.factory.ResultHandlerFactory;
 import com.vcarecity.cvs.service.ResultHandlerService;
 import com.vcarecity.cvs.service.SQLQueryService;
      * @param properties
      * @throws Exception
      */
-    public <T> void exportData(String table, String properties, Class<T> cls) throws Exception {
+    public <T> void exportDataAndConvert(String table, Object properties, Class<T> cls) throws Exception {
 
         if (cls == null) {
             //noinspection unchecked
         int page = 1;
 
         do {
+            // query
             List<T> result = queryService.queryByPage(table, page, this.pageCount, cls);
             final String[] headers = TABLE_COLUMN.get(table);
+            // convert or save
             resultHandler.resultHandler(table, headers, result);
             currentSize = result.size();
 
             ((CSVResultHandlerServiceImpl) resultHandler).close();
         }
 
+
+        final String[] headers = TABLE_COLUMN.get(table);
+        final String[] mysqlHeader = ColumnUpdateMapper.updateColumnName(table, headers);
+
+        logger.info("mysql header: {}", String.join(",", mysqlHeader));
+
         logger.info("stop query database...{} /ms.", (System.currentTimeMillis() - startTime));
     }
 

src/main/java/com/vcarecity/cvs/util/CreateDatasource.java

+package com.vcarecity.cvs.util;
+
+import com.vcarecity.cvs.properties.DbProperties;
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+
+/**
+ * @author Kerry on 19/12/02
+ */
+
+public class CreateDatasource {
+
+
+    public static HikariDataSource createDataSource(DbProperties database) {
+
+        HikariConfig config = new HikariConfig();
+        config.setJdbcUrl(database.getUrl());
+        config.setUsername(database.getUser());
+        config.setPassword(database.getPassword());
+
+        config.addDataSourceProperty("cachePrepStmts", "true");
+        config.addDataSourceProperty("prepStmtCacheSize", "250");
+        config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
+        config.addDataSourceProperty("useServerPrepStmts", "true");
+        config.addDataSourceProperty("useLocalSessionState", "true");
+        config.addDataSourceProperty("rewriteBatchedStatements", "true");
+        config.addDataSourceProperty("cacheResultSetMetadata", "true");
+        config.addDataSourceProperty("cacheServerConfiguration", "true");
+        config.addDataSourceProperty("elideSetAutoCommits", "true");
+        config.addDataSourceProperty("maintainTimeStats", "false");
+
+        return new HikariDataSource(config);
+    }
+
+}

src/main/resources/logback-test.xml

 
     <logger name="com.vcarecity" level="INFO" additivity="false">
         <appender-ref ref="STDOUT"/>
+        <appender-ref ref="FILE_APPENDER"/>
     </logger>
 
     <root level="INFO">

src/test/java/com/vcarecity/cvs/service/impl/MySQLResultHandlerServiceImplTest.java

+package com.vcarecity.cvs.service.impl;
+
+import java.util.Arrays;
+
+class MySQLResultHandlerServiceImplTest {
+
+    @org.junit.jupiter.api.Test
+    void resultHandler() {
+        String[] h = new String[3];
+        StringBuffer sb = new StringBuffer();
+
+        Object[] rows = new Object[]{1, "test", null};
+
+        for (int i = 0; i < rows.length; i++) {
+            Object row = rows[i];
+            if (row instanceof String) {
+                sb.append("'").append(row).append("'");
+            } else {
+                sb.append(row);
+            }
+
+            if (i != rows.length - 1) {
+                sb.append(",");
+            }
+        }
+
+
+        System.out.println(sb.toString());
+
+
+        Arrays.fill(h, "%s");
+        final String join = String.join(",", h);
+
+
+        final String value = String.format(join, rows);
+
+        System.out.println(join);
+        System.out.println(value);
+    }
+}