动态支持MySQL/Oracle

config/application-mysql.yml

-database:
+from-db:
+  url: jdbc:mysql:replication://192.168.10.210:6446,192.168.10.210:6447/fmmp?serverTimezone=Asia/Shanghai
+  user: root
+  password: R00T@mysql
+to-db:
   url: jdbc:mysql:replication://192.168.10.210:6446,192.168.10.210:6447/fmmp?serverTimezone=Asia/Shanghai
   user: root
   password: R00T@mysql

config/application-oracle.yml

-database:
+from-db:
   url: jdbc:oracle:thin:@192.168.10.236:1521:orcl
   user: C##FMMP
   password: vcare~1(^_^)
+to-db:
+  url: jdbc:mysql:replication://192.168.10.210:6446,192.168.10.210:6447/fmmp?serverTimezone=Asia/Shanghai
+  user: root
+  password: R00T@mysql
 export:
   folder: ./logs/

config/application.yml

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

import.sh

+#/bin/bash
+
+echo "test"

pom.xml

             <artifactId>reflections</artifactId>
             <version>0.9.11</version>
         </dependency>
-
-        <dependency>
-            <groupId>org.apache.shardingsphere</groupId>
-            <artifactId>sharding-jdbc-core</artifactId>
-            <version>4.0.0-RC2</version>
-        </dependency>
         <dependency>
             <groupId>javax.persistence</groupId>
             <artifactId>persistence-api</artifactId>

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

 import com.vcarecity.cvs.starter.SQLStarter;
 import me.yuxiaoyao.config.loader.ConfigLoader;
 
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
 /**
  * @author Kerry on 19/11/27
  */
 public class FileExporterApp {
 
     public static Injector injector;
+    public static Map<String, String[]> TABLE_COLUMN = new ConcurrentHashMap<>();
 
 
     public static void main(String[] args) throws Exception {
 
 
     private static void jdbcArgs(String[] args, AppProperties properties) {
-        final DbProperties database = properties.getDatabase();
+        final DbProperties database = properties.getFromDb();
         for (String arg : args) {
             if (arg.startsWith("jdbc.url=")) {
                 database.setUrl(arg.substring("jdbc.url=".length()).trim());
                 database.setPassword(arg.substring("jdbc.password=".length()).trim());
             }
         }
-        properties.setDatabase(database);
+        properties.setFromDb(database);
     }
 }

src/main/java/com/vcarecity/cvs/core/ColumnUpdateMapper.java

+package com.vcarecity.cvs.core;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * @author Kerry on 19/11/29
+ */
+
+public class ColumnUpdateMapper {
+
+
+    @AllArgsConstructor
+    @NoArgsConstructor
+    @Data
+    private static class ColumnMapper {
+        private String origin;
+        private String update;
+    }
+
+    private static List<ColumnMapper> manualMapper;
+
+    private static List<String> MYSQL_KEY_WORD;
+
+    static {
+        manualMapper = new ArrayList<>();
+        manualMapper.add(new ColumnMapper("POSITION", "POSITIONS"));
+
+
+        Set<String> keyWord = new HashSet<>();
+        keyWord.add("POSITION");
+
+        MYSQL_KEY_WORD = new ArrayList<>(keyWord);
+    }
+
+
+    /**
+     * Oracle数据库的字段可能跟 MySQL 不一样,统一这里处理
+     *
+     * @param table
+     * @param headers
+     * @return
+     */
+    public static String[] updateColumnName(String table, String[] headers) {
+        // common mapper
+        for (int i = 0; i < headers.length; i++) {
+            final String h = headers[i].toUpperCase();
+            for (ColumnMapper columnMapper : manualMapper) {
+                if (columnMapper.getOrigin().equalsIgnoreCase(h)) {
+                    headers[i] = columnMapper.getUpdate();
+                    break;
+                }
+            }
+        }
+
+        // 关键字先处理下
+        for (int i = 0; i < headers.length; i++) {
+            final String h = headers[i].toUpperCase();
+            for (String s : MYSQL_KEY_WORD) {
+                if (s.equalsIgnoreCase(h)) {
+                    headers[i] = "`" + headers[i] + "`";
+                    break;
+                }
+            }
+        }
+
+        return headers;
+    }
+
+}

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

+package com.vcarecity.cvs.event;
+
+/**
+ * @author Kerry on 19/11/29
+ */
+
+public interface ExportEventListener {
+
+    /**
+     * 导出之前
+     *
+     * @param filename
+     */
+    void beforeExport(String filename);
+
+    /**
+     * 导出成功
+     *
+     * @param filename
+     * @param table
+     * @param header
+     */
+    void exportSuccess(String filename, String table, String[] header);
+
+    /**
+     * 异常
+     *
+     * @param throwable
+     */
+    void exceptionCatch(Throwable throwable);
+
+
+}

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;
+
+/**
+ * @author Kerry on 19/11/29
+ */
+
+@Slf4j
+public class MySQLImportEvent implements ExportEventListener {
+
+    private AtomicInteger callIndex = new AtomicInteger(0);
+
+    private final DbProperties dbProperties;
+
+    @Inject
+    public MySQLImportEvent(AppProperties properties) {
+        this.dbProperties = properties.getToDb();
+    }
+
+    @Override
+    public void beforeExport(String filename) {
+        logger.info("MySQLImportEvent.beforeExport. filename = {}", filename);
+    }
+
+    @Override
+    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();
+
+        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();
+    }
+
+
+    @Override
+    public void exceptionCatch(Throwable throwable) {
+        logger.info("MySQLImportEvent.exceptionCatch. throwable = {}", throwable);
+    }
+}

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

 import com.google.inject.Provides;
 import com.google.inject.Singleton;
 import com.google.inject.assistedinject.FactoryModuleBuilder;
+import com.vcarecity.cvs.event.ExportEventListener;
+import com.vcarecity.cvs.event.MySQLImportEvent;
 import com.vcarecity.cvs.factory.ResultHandlerFactory;
 import com.vcarecity.cvs.properties.AppProperties;
 import com.vcarecity.cvs.properties.DbProperties;
     @Override
     protected void configure() {
 
+        bind(ExportEventListener.class).to(MySQLImportEvent.class);
+
         install(new FactoryModuleBuilder()
                 .implement(ResultHandlerService.class, CSVResultHandlerServiceImpl.class)
                 .build(ResultHandlerFactory.class));
     public HikariDataSource dataSource(AppProperties properties) {
         // @see https://github.com/brettwooldridge/HikariCP/wiki/MySQL-Configuration
 
-        final DbProperties database = properties.getDatabase();
+        final DbProperties database = properties.getFromDb();
         HikariConfig config = new HikariConfig();
         config.setJdbcUrl(database.getUrl());
         config.setUsername(database.getUser());

src/main/java/com/vcarecity/cvs/properties/AppProperties.java

 package com.vcarecity.cvs.properties;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
 import lombok.Data;
 
 /**
 
 @Data
 public class AppProperties {
-    private DbProperties database;
+
+    @JsonProperty("from-db")
+    private DbProperties fromDb;
+
+    @JsonProperty("to-db")
+    private DbProperties toDb;
+
     private Export export;
 
     @Data
     public static class Export {
         private String folder;
     }
+
+
 }

src/main/java/com/vcarecity/cvs/service/ResultHandlerService.java

     /**
      * ins
      *
+     * @param table
+     * @param header
      * @param dataList
      * @throws Exception
      */
-    <T> void resultHandler(List<T> dataList) throws Exception;
+    <T> void resultHandler(String table, String[] header, List<T> dataList) throws Exception;
 
 
 }

src/main/java/com/vcarecity/cvs/service/SQLQueryService.java

 
 public interface SQLQueryService {
 
-
     /**
      * query by page
      *

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

 import com.vcarecity.cvs.core.AutoMapper;
 import com.vcarecity.cvs.service.SQLQueryService;
 import com.zaxxer.hikari.HikariDataSource;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
 import java.sql.*;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static com.vcarecity.cvs.FileExporterApp.TABLE_COLUMN;
 
 /**
  * @author Kerry on 19/11/27
 @Slf4j
 public abstract class AbstractSQLQueryService implements SQLQueryService {
 
+    private boolean isFirst = true;
+
+
     protected final HikariDataSource dataSource;
 
     protected AbstractSQLQueryService(HikariDataSource dataSource) {
 
         ResultSet resultSet = preparedStatement.executeQuery();
 
-        List<T> list = resultMapper(resultSet, clazz);
+        List<T> list = resultMapper(table, resultSet, clazz);
 
         logger.debug("query success. size = {}. usageTime = {} /ms", list.size(), (System.currentTimeMillis() - startTime));
 
      */
     protected abstract PreparedStatement getPreparedStatement(Connection connection, String sql, int start, int pageCount) throws SQLException;
 
-    public <T> List<T> resultMapper(ResultSet resultSet, Class<T> clazz) throws SQLException {
+    public <T> List<T> resultMapper(String table, ResultSet resultSet, Class<T> clazz) throws SQLException {
         if (clazz == Object.class) {
             //noinspection unchecked
-            return (List<T>) resultSetToObject(resultSet);
+            return (List<T>) resultSetToObject(table, resultSet);
         }
         return autoMapper(resultSet, clazz);
     }
 
-    public List<Object[]> resultSetToObject(ResultSet resultSet) throws SQLException {
+    public List<Object[]> resultSetToObject(String table, ResultSet resultSet) throws SQLException {
         final ResultSetMetaData metaData = resultSet.getMetaData();
         final int columnCount = metaData.getColumnCount() - 1;
 
 
-        int skipRnIndex = -1;
-        int index = 0;
-        String[] header = new String[columnCount];
-        for (int i = 0; i < columnCount; i++) {
-            String s = metaData.getColumnName(i + 1);
-            if ("RN".equalsIgnoreCase(s)) {
-                skipRnIndex = i;
-            } else {
-                header[index++] = s;
-            }
-        }
+        final List<ColumnIndexName> columnWithObject = getColumnWithObject(metaData);
+        final Set<Integer> skipIndex = columnWithObject.stream().map(ColumnIndexName::getIndex).collect(Collectors.toSet());
 
-        logger.info("header = ({})", String.join(",", header));
+        final String[] header = columnWithObject.stream().map(ColumnIndexName::getName).toArray(String[]::new);
+
+
+        TABLE_COLUMN.put(table, header);
+
+
+        if (isFirst) {
+            final String join = String.join(",", header);
+            logger.info("header = ({})", join);
+            isFirst = false;
+        }
 
         List<Object[]> data = new ArrayList<>();
         while (resultSet.next()) {
             Object[] objects = new Object[columnCount];
             int cIndex = 0;
             for (int i = 0; i < columnCount; i++) {
-                if (i == skipRnIndex) {
+                if (!skipIndex.contains(i)) {
                     continue;
                 }
                 objects[cIndex++] = resultSet.getObject(i + 1);
         return data;
     }
 
+
+    @NoArgsConstructor
+    @AllArgsConstructor
+    @Data
+    public static class ColumnIndexName {
+        private int index;
+        private String name;
+    }
+
+    /**
+     * @param metaData
+     * @return
+     */
+    protected List<ColumnIndexName> getColumnWithObject(ResultSetMetaData metaData) throws SQLException {
+        final int columnCount = metaData.getColumnCount();
+        List<ColumnIndexName> res = new ArrayList<>(columnCount);
+        for (int i = 0; i < columnCount; i++) {
+            String name = metaData.getColumnName(i + 1);
+            res.add(new ColumnIndexName(i, name));
+        }
+        return res;
+    }
+
     public <T> List<T> autoMapper(ResultSet resultSet, Class<T> clazz) throws SQLException {
         List<T> list = new ArrayList<>();
         while (resultSet.next()) {
         }
         return list;
     }
+
 }

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

 import com.vcarecity.cvs.core.AutoMapper;
 import com.vcarecity.cvs.core.JavaMapColumn;
 import com.vcarecity.cvs.core.ReflectionUtil;
+import com.vcarecity.cvs.event.ExportEventListener;
 import com.vcarecity.cvs.service.ResultHandlerService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.csv.CSVFormat;
     private int counter;
     private int fileNameIndex = 0;
 
+    private File lastFile;
+
+    private String table;
+    private String[] lastHeader;
 
     /**
      * 分割文件行数
      */
     private static final int SPLIT_LINE_NUM = 200_0000;
 
+    private final ExportEventListener eventListener;
 
     @Inject
-    public CSVResultHandlerServiceImpl(@Assisted String filePattern) {
+    public CSVResultHandlerServiceImpl(ExportEventListener eventListener, @Assisted String filePattern) {
+        this.eventListener = eventListener;
         this.filePattern = filePattern;
         this.csvFormat = CSVFormat.MYSQL;
         this.counter = 0;
     private File getFile() {
         final int index = fileNameIndex;
         fileNameIndex++;
-        return new File(filePattern + "-" + +index + ".csv");
+
+        lastFile = new File(filePattern + "-" + +index + ".csv");
+        // call back
+        eventListener.beforeExport(lastFile.getAbsolutePath());
+
+        return lastFile;
     }
 
 
         synchronized (this) {
             if (counter % SPLIT_LINE_NUM == 0) {
                 try {
+                    if (lastFile != null) {
+                        eventListener.exportSuccess(this.lastFile.getAbsolutePath(), table, lastHeader);
+                    }
                     if (bufferedWriter != null) {
                         bufferedWriter.close();
                     }
                     writer = new CSVPrinter(bufferedWriter, csvFormat);
                 } catch (IOException e) {
                     e.printStackTrace();
+                    eventListener.exceptionCatch(e);
                 }
             }
         }
     }
 
     @Override
-    public <T> void resultHandler(List<T> dataList) throws Exception {
+    public <T> void resultHandler(String table, String[] header, List<T> dataList) throws Exception {
+        this.table = table;
+        this.lastHeader = header;
         checkFile();
 
         logger.debug("start write to csv file size = {}", dataList.size());
 
         long startTime = System.currentTimeMillis();
 
-
         Class<?> cls = Object.class;
 
         for (T t : dataList) {
     }
 
     public void close() throws IOException {
+        eventListener.exportSuccess(this.lastFile.getAbsolutePath(), table, this.lastHeader);
         bufferedWriter.close();
         writer.close();
     }
         return localTime.format(TIME_FORMATTER);
     }
 
+
 }

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.service.ResultHandlerService;
 import com.zaxxer.hikari.HikariDataSource;
 import lombok.extern.slf4j.Slf4j;
 public class MySQLResultHandlerServiceImpl implements ResultHandlerService {
 
     private final HikariDataSource dataSource;
+    private final ExportEventListener eventListener;
     private final String sql;
 
     @Inject
-    public MySQLResultHandlerServiceImpl(HikariDataSource dataSource, @Assisted String sql) {
+    public MySQLResultHandlerServiceImpl(HikariDataSource dataSource, ExportEventListener eventListener, @Assisted String sql) {
         this.dataSource = dataSource;
+        this.eventListener = eventListener;
         this.sql = sql;
     }
 
 
     @Override
-    public <T> void resultHandler(List<T> dataList) throws Exception {
+    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());
         long startTime = System.currentTimeMillis();
         final Connection connection = dataSource.getConnection();

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

 
 import java.sql.Connection;
 import java.sql.PreparedStatement;
+import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
 
 /**
  * @author Kerry on 19/11/25
         preparedStatement.setInt(2, (page - 1) * pageCount);
         return preparedStatement;
     }
+
+    @Override
+    protected List<ColumnIndexName> getColumnWithObject(ResultSetMetaData metaData) throws SQLException {
+        final int columnCount = metaData.getColumnCount();
+        List<ColumnIndexName> res = new ArrayList<>(columnCount);
+        for (int i = 0; i < columnCount; i++) {
+            String name = metaData.getColumnName(i + 1);
+            if ("RN".equalsIgnoreCase(name)) {
+                continue;
+            }
+            res.add(new ColumnIndexName(i, name));
+        }
+        return res;
+    }
 }

src/main/java/com/vcarecity/cvs/shell/MySQLImporter.java

+package com.vcarecity.cvs.shell;
+
+import com.vcarecity.cvs.properties.DbProperties;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.File;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+
+/**
+ * @author Kerry on 19/11/29
+ */
+
+@Slf4j
+public class MySQLImporter implements Runnable {
+
+    private final DbProperties dbProperties;
+
+    private final String filepath;
+    private final String table;
+    private final String[] header;
+
+    private static final String LOAD_DATA_SQL_TEP = "LOAD DATA INFILE '%s' " +
+            "INTO TABLE %s " +
+            "FIELDS " +
+            "TERMINATED BY '\\t' " +
+            "ENCLOSED BY '\\\"' " +
+            "ESCAPED BY '\\\\' " +
+            "LINES TERMINATED BY '\\n' (%s)";
+
+    public MySQLImporter(DbProperties dbProperties, String filepath, String table, String[] header) {
+        this.dbProperties = dbProperties;
+        this.filepath = filepath;
+        this.table = table;
+        this.header = header;
+    }
+
+    @Override
+    public void run() {
+        File file = new File(filepath);
+        if (!file.exists()) {
+            return;
+        }
+
+        final String sql = String.format(LOAD_DATA_SQL_TEP, filepath, table, String.join(",", this.header));
+
+        logger.info("sql = {}", sql);
+        try {
+            importData(sql);
+        } catch (SQLException e) {
+            e.printStackTrace();
+        }
+    }
+
+    private void execShell(String sql) {
+        //Runtime.getRuntime().exec()
+    }
+
+
+    private void importFromLocal(String sql) {
+        //if (statement.isWrapperFor(JdbcStatement.class)) {
+        //    final JdbcStatement unwrap = statement.unwrap(JdbcStatement.class);
+        //    unwrap.setLocalInfileInputStream();
+        //}
+    }
+
+
+    private void importData(String sql) throws SQLException {
+        try (final Connection connection = getConnection();
+             final PreparedStatement statement = connection.prepareStatement(sql)) {
+            statement.executeUpdate();
+        }
+    }
+
+    private Connection getConnection() throws SQLException {
+        return DriverManager.getConnection(dbProperties.getUrl(), dbProperties.getUser(), dbProperties.getPassword());
+    }
+
+}

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

 import javax.inject.Inject;
 import java.util.List;
 
+import static com.vcarecity.cvs.FileExporterApp.TABLE_COLUMN;
+
 /**
  * @author Kerry on 19/11/20
  */
             //noinspection unchecked
             cls = (Class<T>) Object.class;
         }
+
         ResultHandlerService resultHandler = resultHandlerFactory.createResultHandler(properties);
 
 
 
         do {
             List<T> result = queryService.queryByPage(table, page, this.pageCount, cls);
-            resultHandler.resultHandler(result);
+            final String[] headers = TABLE_COLUMN.get(table);
+            resultHandler.resultHandler(table, headers, result);
             currentSize = result.size();
 
             page++;

src/main/resources/logback-test.xml

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