diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java index 185423ee..85f2f62c 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java @@ -1,10 +1,11 @@ package com.alibaba.otter.canal.client.adapter.rdb; -import java.io.File; import java.sql.Connection; -import java.sql.ResultSetMetaData; import java.sql.SQLException; -import java.util.*; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import javax.sql.DataSource; @@ -14,8 +15,8 @@ import org.slf4j.LoggerFactory; import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.client.adapter.OuterAdapter; -import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.rdb.config.ConfigLoader; +import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.rdb.service.RdbEtlService; import com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService; import com.alibaba.otter.canal.client.adapter.support.*; @@ -34,10 +35,6 @@ public class RdbAdapter implements OuterAdapter { @Override public void init(OuterAdapterConfig configuration) { - System.out.println("xxxxx: " + this.getClass().getClassLoader().getResource("").getPath()); - File file = new File(this.getClass().getClassLoader().getResource("").getPath() + "rdb" + File.separator); - System.out.println(file.getAbsolutePath()); - Map rdbMappingTmp = ConfigLoader.load(); // 过滤不匹配的key的配置 rdbMappingTmp.forEach((key, mappingConfig) -> { @@ -73,55 +70,6 @@ public class RdbAdapter implements OuterAdapter { logger.error("ERROR ## failed to initial datasource: " + properties.get("jdbc.url"), e); } - rdbMapping.values().forEach(config -> { - try { - MappingConfig.DbMapping dbMapping = config.getDbMapping(); - // 从源表加载所有字段名 - if (dbMapping.getAllColumns() == null) { - synchronized (RdbSyncService.class) { - if (dbMapping.getAllColumns() == null) { - DataSource srcDS = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); - Connection srcConn = srcDS.getConnection(); - String srcMetaSql = "SELECT * FROM " + dbMapping.getDatabase() + "." + dbMapping.getTable() - + " WHERE 1=2 "; - List srcColumns = new ArrayList<>(); - Util.sqlRS(srcConn, srcMetaSql, rs -> { - try { - ResultSetMetaData rmd = rs.getMetaData(); - int cnt = rmd.getColumnCount(); - for (int i = 1; i <= cnt; i++) { - srcColumns.add(rmd.getColumnName(i).toLowerCase()); - } - } catch (SQLException e) { - logger.error(e.getMessage(), e); - } - }); - Map columnsMap = new LinkedHashMap<>(); - - for (String srcColumn : srcColumns) { - String targetColumn = srcColumn; - if (dbMapping.getTargetColumns() != null) { - for (Map.Entry entry : dbMapping.getTargetColumns().entrySet()) { - String targetColumnName = entry.getKey(); - String srcColumnName = entry.getValue(); - - if (srcColumnName != null - && srcColumnName.toLowerCase().equals(srcColumn.toUpperCase())) { - targetColumn = targetColumnName; - } - } - } - columnsMap.put(targetColumn, srcColumn); - } - dbMapping.setAllColumns(columnsMap); - } - } - } - } catch (SQLException e) { - logger.error(e.getMessage(), e); - } - }); - rdbSyncService = new RdbSyncService(dataSource); } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java index 95623ea4..40389f9a 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java @@ -79,7 +79,7 @@ public class MappingConfig { private int readBatch = 5000; private int commitBatch = 5000; // etl等批量提交大小 - private volatile Map allColumns; // mapAll为true,自动设置改字段 +// private volatile Map allColumns; // mapAll为true,自动设置改字段 public String getDatabase() { return database; @@ -161,12 +161,12 @@ public class MappingConfig { this.commitBatch = commitBatch; } - public Map getAllColumns() { - return allColumns; - } - - public void setAllColumns(Map allColumns) { - this.allColumns = allColumns; - } +// public Map getAllColumns() { +// return allColumns; +// } +// +// public void setAllColumns(Map allColumns) { +// this.allColumns = allColumns; +// } } } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java index 321c9f21..3cf4976a 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java @@ -13,6 +13,7 @@ import java.util.concurrent.atomic.AtomicLong; import javax.sql.DataSource; +import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -91,8 +92,8 @@ public class RdbEtlService { } else { sqlFinal = sql + " LIMIT " + offset + "," + cnt; } - Future future = executor.submit( - () -> executeSqlImport(params, srcDS, targetDS, sqlFinal, dbMapping, successCount, errMsg)); + Future future = executor + .submit(() -> executeSqlImport(srcDS, targetDS, sqlFinal, dbMapping, successCount, errMsg)); futures.add(future); } @@ -102,7 +103,7 @@ public class RdbEtlService { executor.shutdown(); } else { - executeSqlImport(params, srcDS, targetDS, sql.toString(), dbMapping, successCount, errMsg); + executeSqlImport(srcDS, targetDS, sql.toString(), dbMapping, successCount, errMsg); } logger.info( @@ -160,8 +161,8 @@ public class RdbEtlService { /** * 执行导入 */ - private static boolean executeSqlImport(List params, DataSource srcDS, DataSource targetDS, String sql, - DbMapping dbMapping, AtomicLong successCount, List errMsg) { + private static boolean executeSqlImport(DataSource srcDS, DataSource targetDS, String sql, DbMapping dbMapping, + AtomicLong successCount, List errMsg) { try { Util.sqlRS(srcDS, sql, rs -> { int idx = 1; @@ -172,16 +173,18 @@ public class RdbEtlService { Map columnType = new LinkedHashMap<>(); ResultSetMetaData rsd = rs.getMetaData(); int columnCount = rsd.getColumnCount(); + List columns = new ArrayList<>(); for (int i = 1; i <= columnCount; i++) { columnType.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i)); + columns.add(rsd.getColumnName(i)); } - Map columnsMap; - if (dbMapping.isMapAll()) { - columnsMap = dbMapping.getAllColumns(); - } else { - columnsMap = dbMapping.getTargetColumns(); - } + Map columnsMap = SyncUtil.getColumnsMap(dbMapping, columns); + // if (dbMapping.isMapAll()) { + // columnsMap = dbMapping.getAllColumns(); + // } else { + // columnsMap = dbMapping.getTargetColumns(); + // } StringBuilder insertSql = new StringBuilder(); insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" ("); diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index 9a01fbf7..def43637 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java @@ -10,11 +10,9 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import java.util.function.Consumer; import javax.sql.DataSource; -import com.alibaba.otter.canal.client.adapter.support.Util; import org.joda.time.DateTime; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -23,8 +21,9 @@ import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig.DbMapping; -import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; +import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil; import com.alibaba.otter.canal.client.adapter.support.Dml; +import com.alibaba.otter.canal.client.adapter.support.Util; /** * RDB同步操作业务 @@ -84,12 +83,7 @@ public class RdbSyncService { Connection conn = dataSource.getConnection(); conn.setAutoCommit(false); try { - Map columnsMap; - if (dbMapping.isMapAll()) { - columnsMap = dbMapping.getAllColumns(); - } else { - columnsMap = dbMapping.getTargetColumns(); - } + Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data.get(0)); StringBuilder insertSql = new StringBuilder(); insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" ("); @@ -181,12 +175,7 @@ public class RdbSyncService { conn.setAutoCommit(false); try { - Map columnsMap; - if (dbMapping.isMapAll()) { - columnsMap = dbMapping.getAllColumns(); - } else { - columnsMap = dbMapping.getTargetColumns(); - } + Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data.get(0)); Map ctype = getTargetColumnType(conn, config); @@ -525,4 +514,5 @@ public class RdbSyncService { logger.error(e.getMessage(), e); } } + } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java new file mode 100644 index 00000000..5e2c38bb --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java @@ -0,0 +1,39 @@ +package com.alibaba.otter.canal.client.adapter.rdb.support; + +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.Map; + +import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; + +public class SyncUtil { + + public static Map getColumnsMap(MappingConfig.DbMapping dbMapping, Map data) { + return getColumnsMap(dbMapping, data.keySet()); + } + + public static Map getColumnsMap(MappingConfig.DbMapping dbMapping, Collection columns) { + Map columnsMap; + if (dbMapping.isMapAll()) { + columnsMap = new LinkedHashMap<>(); + for (String srcColumn : columns) { + boolean flag = true; + if (dbMapping.getTargetColumns() != null) { + for (Map.Entry entry : dbMapping.getTargetColumns().entrySet()) { + if (srcColumn.equals(entry.getValue())) { + columnsMap.put(entry.getKey(), srcColumn); + flag = false; + break; + } + } + } + if (flag) { + columnsMap.put(srcColumn, srcColumn); + } + } + } else { + columnsMap = dbMapping.getTargetColumns(); + } + return columnsMap; + } +}