From 8ae192f0a91f331e3343986cd91028c391132426 Mon Sep 17 00:00:00 2001 From: mcy Date: Tue, 6 Nov 2018 09:56:16 +0800 Subject: [PATCH 1/4] =?UTF-8?q?adapter=20canal-client=20=E7=89=88=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client-adapter/common/pom.xml | 2 +- client-adapter/launcher/pom.xml | 2 +- client-adapter/pom.xml | 2 -- 3 files changed, 2 insertions(+), 4 deletions(-) diff --git a/client-adapter/common/pom.xml b/client-adapter/common/pom.xml index f018c490..e62fc386 100644 --- a/client-adapter/common/pom.xml +++ b/client-adapter/common/pom.xml @@ -14,7 +14,7 @@ com.alibaba.otter canal.protocol - ${canal_version} + 1.1.2-SNAPSHOT joda-time diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index c24674bd..c8969223 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -33,7 +33,7 @@ com.alibaba.otter canal.client - ${canal_version} + 1.1.2-SNAPSHOT org.yaml diff --git a/client-adapter/pom.xml b/client-adapter/pom.xml index d18d1c19..4cc7a80d 100644 --- a/client-adapter/pom.xml +++ b/client-adapter/pom.xml @@ -14,8 +14,6 @@ 1.8 1.8 UTF-8 - - 1.1.2-SNAPSHOT From 5ae71121927887b27d42bc9a8d8d10b730932da5 Mon Sep 17 00:00:00 2001 From: mcy Date: Wed, 7 Nov 2018 21:01:50 +0800 Subject: [PATCH 2/4] init rdb sync --- .../adapter/es/config/ESSyncConfigLoader.java | 6 +- .../hbase/config/MappingConfigLoader.java | 9 +- .../launcher/config/AdapterConfig.java | 2 - client-adapter/pom.xml | 1 + client-adapter/rdb/pom.xml | 85 ++++ .../canal/client/adapter/rdb/RdbAdapter.java | 86 ++++ .../adapter/rdb/config/MappingConfig.java | 142 +++++++ .../rdb/config/MappingConfigLoader.java | 102 +++++ .../adapter/rdb/service/RdbSyncService.java | 368 ++++++++++++++++++ ...ba.otter.canal.client.adapter.OuterAdapter | 1 + .../src/main/resources/rdb/mytest_user.yml | 12 + .../adapter/rdb/test/ConfigLoadTest.java | 29 ++ .../canal/client/adapter/rdb/test/DBTest.java | 82 ++++ .../client/adapter/rdb/test/TestConstant.java | 37 ++ .../client/adapter/rdb/test/sync/Common.java | 28 ++ .../adapter/rdb/test/sync/OracleSyncTest.java | 47 +++ .../rdb/src/test/resources/log4j2-test.xml | 13 + .../rdb/src/test/resources/logback-test.xml | 13 + .../src/test/resources/rdb/mytest_user.yml | 14 + 19 files changed, 1068 insertions(+), 9 deletions(-) create mode 100644 client-adapter/rdb/pom.xml create mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java create mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java create mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java create mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java create mode 100644 client-adapter/rdb/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter create mode 100644 client-adapter/rdb/src/main/resources/rdb/mytest_user.yml create mode 100644 client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java create mode 100644 client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java create mode 100644 client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/TestConstant.java create mode 100644 client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java create mode 100644 client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java create mode 100644 client-adapter/rdb/src/test/resources/log4j2-test.xml create mode 100644 client-adapter/rdb/src/test/resources/logback-test.xml create mode 100644 client-adapter/rdb/src/test/resources/rdb/mytest_user.yml diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java index d9b09161..681537cf 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java @@ -42,7 +42,7 @@ public class ESSyncConfigLoader { } public static synchronized void load() { - logger.info("## Start loading mapping config ... "); + logger.info("## Start loading es mapping config ... "); Collection configs = AdapterConfigs.get("es"); if (configs == null) { return; @@ -92,7 +92,7 @@ public class ESSyncConfigLoader { esSyncConfig.put(c, config); } - logger.info("## Mapping config loaded"); + logger.info("## ES mapping config loaded"); } private static String readConfigContent(String config) { @@ -113,7 +113,7 @@ public class ESSyncConfigLoader { in.read(bytes); return new String(bytes, StandardCharsets.UTF_8); } catch (IOException e) { - throw new RuntimeException("Read yml config error ", e); + throw new RuntimeException("Read es mapping config error ", e); } finally { try { if (in != null) { diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java index 5b6cf2bd..f2ee6994 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java @@ -4,6 +4,7 @@ import java.io.File; import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; +import java.nio.charset.StandardCharsets; import java.util.Collection; import java.util.LinkedHashMap; import java.util.Map; @@ -33,7 +34,7 @@ public class MappingConfigLoader { * @return 配置名/配置文件名--对象 */ public static Map load() { - logger.info("## Start loading mapping config ... "); + logger.info("## Start loading hbase mapping config ... "); Map result = new LinkedHashMap<>(); @@ -116,7 +117,7 @@ public class MappingConfigLoader { result.put(c, config); } - logger.info("## Mapping config loaded"); + logger.info("## Hbase mapping config loaded"); return result; } @@ -136,9 +137,9 @@ public class MappingConfigLoader { byte[] bytes = new byte[in.available()]; in.read(bytes); - return new String(bytes, "UTF-8"); + return new String(bytes, StandardCharsets.UTF_8); } catch (IOException e) { - throw new RuntimeException("Read ds-config.yml or hbase-mappings.conf error. ", e); + throw new RuntimeException("Read hbase mapping config error. ", e); } finally { try { if (in != null) { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java index f3e265f8..0f18c996 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java @@ -55,8 +55,6 @@ public class AdapterConfig { ds.setMaxWait(60000); ds.setTimeBetweenEvictionRunsMillis(60000); ds.setMinEvictableIdleTimeMillis(300000); - ds.setPoolPreparedStatements(false); - ds.setMaxPoolPreparedStatementPerConnectionSize(20); ds.setValidationQuery("select 1"); try { ds.init(); diff --git a/client-adapter/pom.xml b/client-adapter/pom.xml index 4cc7a80d..3c1dcb55 100644 --- a/client-adapter/pom.xml +++ b/client-adapter/pom.xml @@ -22,6 +22,7 @@ hbase elasticsearch launcher + rdb diff --git a/client-adapter/rdb/pom.xml b/client-adapter/rdb/pom.xml new file mode 100644 index 00000000..951579a9 --- /dev/null +++ b/client-adapter/rdb/pom.xml @@ -0,0 +1,85 @@ + + + + canal.client-adapter + com.alibaba.otter + 1.1.2-SNAPSHOT + + 4.0.0 + com.alibaba.otter + client-adapter.rdb + jar + canal client adapter rdb module for otter ${project.version} + + + + com.alibaba.otter + client-adapter.common + ${project.version} + provided + + + org.yaml + snakeyaml + 1.19 + provided + + + + mysql + mysql-connector-java + 5.1.40 + test + + + org.postgresql + postgresql + 42.1.4 + test + + + com.oracle + ojdbc6 + 11.2.0.4.0-atlassian-hosted + test + + + com.microsoft.sqlserver + mssql-jdbc + 7.0.0.jre8 + test + + + junit + junit + 4.12 + test + + + + + + + org.apache.maven.plugins + maven-assembly-plugin + 2.4 + + + jar-with-dependencies + + + + + make-assembly + package + + single + + + + + + + \ No newline at end of file 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 new file mode 100644 index 00000000..d8fb9bcb --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java @@ -0,0 +1,86 @@ +package com.alibaba.otter.canal.client.adapter.rdb; + +import java.sql.SQLException; +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +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.MappingConfigLoader; +import com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService; +import com.alibaba.otter.canal.client.adapter.support.Dml; +import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; +import com.alibaba.otter.canal.client.adapter.support.SPI; + +@SPI("rdb") +public class RdbAdapter implements OuterAdapter { + + private static Logger logger = LoggerFactory.getLogger(RdbAdapter.class); + + private static volatile Map rdbMapping = null; // 文件名对应配置 + private static volatile Map mappingConfigCache = null; // 库名-表名对应配置 + + private DruidDataSource dataSource; + + private RdbSyncService rdbSyncService; + + @Override + public void init(OuterAdapterConfig configuration) { + if (mappingConfigCache == null) { + synchronized (MappingConfig.class) { + if (mappingConfigCache == null) { + rdbMapping = MappingConfigLoader.load(); + mappingConfigCache = new HashMap<>(); + for (MappingConfig mappingConfig : rdbMapping.values()) { + mappingConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." + + mappingConfig.getDbMapping().getDatabase() + "." + + mappingConfig.getDbMapping().getTable(), + mappingConfig); + } + } + } + } + + Map properties = configuration.getProperties(); + dataSource = new DruidDataSource(); + dataSource.setDriverClassName(properties.get("jdbc.driverClassName")); + dataSource.setUrl(properties.get("jdbc.url")); + dataSource.setUsername(properties.get("jdbc.username")); + dataSource.setPassword(properties.get("jdbc.password")); + dataSource.setInitialSize(1); + dataSource.setMinIdle(1); + dataSource.setMaxActive(2); + dataSource.setMaxWait(60000); + dataSource.setTimeBetweenEvictionRunsMillis(60000); + dataSource.setMinEvictableIdleTimeMillis(300000); + + try { + dataSource.init(); + } catch (SQLException e) { + logger.error("ERROR ## failed to initial datasource: " + properties.get("jdbc.url"), e); + } + + rdbSyncService = new RdbSyncService(dataSource); + } + + @Override + public void sync(Dml dml) { + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String database = dml.getDatabase(); + String table = dml.getTable(); + MappingConfig config = mappingConfigCache.get(destination + "." + database + "." + table); + rdbSyncService.sync(config, dml); + } + + @Override + public void destroy() { + if (dataSource != null) { + dataSource.close(); + } + } +} 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 new file mode 100644 index 00000000..d0709db6 --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java @@ -0,0 +1,142 @@ +package com.alibaba.otter.canal.client.adapter.rdb.config; + +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Set; + +/** + * RDB表映射配置 + * + * @author rewerma 2018-11-07 下午02:41:34 + * @version 1.0.0 + */ +public class MappingConfig { + + private String dataSourceKey; // 数据源key + + private String destination; // canal实例或MQ的topic + + private DbMapping dbMapping; // db映射配置 + + public String getDataSourceKey() { + return dataSourceKey; + } + + public void setDataSourceKey(String dataSourceKey) { + this.dataSourceKey = dataSourceKey; + } + + public DbMapping getDbMapping() { + return dbMapping; + } + + public void setDbMapping(DbMapping dbMapping) { + this.dbMapping = dbMapping; + } + + public String getDestination() { + return destination; + } + + public void setDestination(String destination) { + this.destination = destination; + } + + public void validate() { + if (dbMapping.database == null || dbMapping.database.isEmpty()) { + throw new NullPointerException("dbMapping.database"); + } + if (dbMapping.table == null || dbMapping.table.isEmpty()) { + throw new NullPointerException("dbMapping.table"); + } + if (dbMapping.targetTable == null || dbMapping.targetTable.isEmpty()) { + throw new NullPointerException("dbMapping.targetTable"); + } + } + + public static class DbMapping { + + private String database; // 数据库名或schema名 + private String table; // 表面名 + private boolean mapAll = false; // 映射所有字段 + private String targetTable; // 目标表名 + private Map targetColums; // 目标表字段映射 + private String etlCondition; // etl条件sql + + private Set families = new LinkedHashSet<>(); // column family列表 + private int readBatch = 5000; + private int commitBatch = 5000; // etl等批量提交大小 + + public String getDatabase() { + return database; + } + + public void setDatabase(String database) { + this.database = database; + } + + public String getTable() { + return table; + } + + public void setTable(String table) { + this.table = table; + } + + public boolean isMapAll() { + return mapAll; + } + + public void setMapAll(boolean mapAll) { + this.mapAll = mapAll; + } + + public String getTargetTable() { + return targetTable; + } + + public void setTargetTable(String targetTable) { + this.targetTable = targetTable; + } + + public Map getTargetColums() { + return targetColums; + } + + public void setTargetColums(Map targetColums) { + this.targetColums = targetColums; + } + + public String getEtlCondition() { + return etlCondition; + } + + public void setEtlCondition(String etlCondition) { + this.etlCondition = etlCondition; + } + + public Set getFamilies() { + return families; + } + + public void setFamilies(Set families) { + this.families = families; + } + + public int getReadBatch() { + return readBatch; + } + + public void setReadBatch(int readBatch) { + this.readBatch = readBatch; + } + + public int getCommitBatch() { + return commitBatch; + } + + public void setCommitBatch(int commitBatch) { + this.commitBatch = commitBatch; + } + } +} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java new file mode 100644 index 00000000..8d3613a8 --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java @@ -0,0 +1,102 @@ +package com.alibaba.otter.canal.client.adapter.rdb.config; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.Map; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.yaml.snakeyaml.Yaml; + +import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; + +/** + * RDB表映射配置加载器 + * + * @author rewerma 2018-11-07 下午02:41:34 + * @version 1.0.0 + */ +public class MappingConfigLoader { + + private static Logger logger = LoggerFactory.getLogger(MappingConfigLoader.class); + + private static final String BASE_PATH = "rdb"; + + /** + * 加载HBase表映射配置 + * + * @return 配置名/配置文件名--对象 + */ + public static Map load() { + logger.info("## Start loading rdb mapping config ... "); + + Map result = new LinkedHashMap<>(); + + Collection configs = AdapterConfigs.get("rdb"); + if (configs == null) { + return result; + } + for (String c : configs) { + if (c == null) { + continue; + } + c = c.trim(); + if (c.equals("") || c.startsWith("#")) { + continue; + } + + String configContent = null; + + if (c.endsWith(".yml")) { + configContent = readConfigContent(BASE_PATH + "/" + c); + } + + MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class); + + try { + config.validate(); + } catch (Exception e) { + throw new RuntimeException("ERROR Config: " + c + " " + e.getMessage(), e); + } + result.put(c, config); + } + + logger.info("## Rdb mapping config loaded"); + return result; + } + + public static String readConfigContent(String config) { + InputStream in = null; + try { + // 先取本地文件，再取类路径 + File configFile = new File("../config/" + config); + if (configFile.exists()) { + in = new FileInputStream(configFile); + } else { + in = MappingConfigLoader.class.getClassLoader().getResourceAsStream(config); + } + if (in == null) { + throw new RuntimeException("Rdb mapping config file not found."); + } + + byte[] bytes = new byte[in.available()]; + in.read(bytes); + return new String(bytes, StandardCharsets.UTF_8); + } catch (IOException e) { + throw new RuntimeException("Read rdb mapping config error. ", e); + } finally { + try { + if (in != null) { + in.close(); + } + } catch (IOException e) { + // ignore + } + } + } +} 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 new file mode 100644 index 00000000..9012b6d5 --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java @@ -0,0 +1,368 @@ +package com.alibaba.otter.canal.client.adapter.rdb.service; + +import java.io.Reader; +import java.io.StringReader; +import java.math.BigDecimal; +import java.nio.charset.StandardCharsets; +import java.sql.*; +import java.util.Date; +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 org.joda.time.DateTime; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.alibaba.fastjson.JSON; +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.Dml; + +/** + * RDB同步操作业务 + * + * @author rewerma 2018-11-7 下午06:45:49 + * @version 1.0.0 + */ +public class RdbSyncService { + + private static Logger logger = LoggerFactory.getLogger(RdbSyncService.class); + + private DataSource dataSource; + + public RdbSyncService(DataSource dataSource){ + this.dataSource = dataSource; + } + + public void sync(MappingConfig config, Dml dml) { + try { + if (config != null) { + String type = dml.getType(); + if (type != null && type.equalsIgnoreCase("INSERT")) { + insert(config, dml); + } else if (type != null && type.equalsIgnoreCase("UPDATE")) { + // update(config, dml); + } else if (type != null && type.equalsIgnoreCase("DELETE")) { + // delete(config, dml); + } + if (logger.isDebugEnabled()) { + logger.debug("DML: {}", JSON.toJSONString(dml)); + } + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + /** + * 插入操作 + * + * @param config 配置项 + * @param dml DML数据 + */ + private void insert(MappingConfig config, Dml dml) throws SQLException { + List> data = dml.getData(); + if (data == null || data.isEmpty()) { + return; + } + + DbMapping dbMapping = config.getDbMapping(); + + int idx = 1; + boolean complete = false; + + Connection conn = dataSource.getConnection(); + + conn.setAutoCommit(false); + + StringBuilder insertSql = new StringBuilder(); + insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" ("); + if (!dbMapping.isMapAll()) { + dbMapping.getTargetColums().forEach((targetColumnName, srcColumnName) -> { + insertSql.append(targetColumnName).append(","); + }); + int len = insertSql.length(); + insertSql.delete(len - 1, len).append(") VALUES ("); + int mapLen = dbMapping.getTargetColums().size(); + for (int i = 0; i < mapLen; i++) { + insertSql.append("?,"); + } + len = insertSql.length(); + insertSql.delete(len - 1, len).append(")"); + } + + PreparedStatement pstmt = conn.prepareStatement(insertSql.toString()); + + for (Map r : data) { + pstmt.clearParameters(); + convertData2DbRow(conn, config, r, pstmt); + + pstmt.execute(); + + if (idx % config.getDbMapping().getCommitBatch() == 0) { + conn.commit(); + complete = true; + } + idx++; + } + if (!complete) { + conn.commit(); + } + conn.close(); + } + + private static void sqlRS(Connection conn, String sql, Consumer consumer) { + Statement stmt = null; + ResultSet rs = null; + try { + stmt = conn.createStatement(); + rs = stmt.executeQuery(sql); + consumer.accept(rs); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } finally { + if (rs != null) { + try { + rs.close(); + rs = null; + } catch (SQLException e) { + // ignore + } + } + if (stmt != null) { + try { + stmt.close(); + stmt = null; + } catch (SQLException e) { + // ignore + } + } + } + } + + private static Map> COLUMNS_TYPE_CACHE = new ConcurrentHashMap<>(); + + private Map getTargetColumnType(Connection conn, MappingConfig config) { + DbMapping dbMapping = config.getDbMapping(); + String cacheKey = config.getDestination() + "." + dbMapping.getDatabase() + "." + dbMapping.getTable(); + Map columnType = COLUMNS_TYPE_CACHE.get(cacheKey); + if (columnType == null) { + synchronized (RdbSyncService.class) { + columnType = COLUMNS_TYPE_CACHE.get(cacheKey); + if (columnType == null) { + columnType = new LinkedHashMap<>(); + final Map columnTypeTmp = columnType; + String sql = "SELECT * FROM " + dbMapping.getTargetTable() + " WHERE 1=2"; + sqlRS(conn, sql, rs -> { + try { + ResultSetMetaData rsd = rs.getMetaData(); + int columnCount = rsd.getColumnCount(); + for (int i = 1; i <= columnCount; i++) { + columnTypeTmp.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i)); + } + COLUMNS_TYPE_CACHE.put(cacheKey, columnTypeTmp); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + }); + } + } + } + return columnType; + } + + /** + * 新增类型转换 + * + * @param config + * @param data + * @param pstmt + * @throws SQLException + */ + private void convertData2DbRow(Connection conn, MappingConfig config, Map data, + PreparedStatement pstmt) throws SQLException { + DbMapping dbMapping = config.getDbMapping(); + if (dbMapping.isMapAll()) { + + } else { + Map ctype = getTargetColumnType(conn, config); + + int i = 1; + for (Map.Entry entry : dbMapping.getTargetColums().entrySet()) { + String targetClassName = entry.getKey(); + String srcColumnName = entry.getValue(); + if (srcColumnName == null) { + srcColumnName = targetClassName; + } + + Integer type = ctype.get(targetClassName.toLowerCase()); + + Object value = data.get(srcColumnName); + if (value != null) { + if (type == null) { + throw new RuntimeException("No column: " + targetClassName + " found in target db"); + } + switch (type) { + case Types.BIT: + case Types.BOOLEAN: + if (value instanceof Boolean) { + pstmt.setBoolean(i, (Boolean) value); + } else if (value instanceof String) { + boolean v = !value.equals("0"); + pstmt.setBoolean(i, v); + } else if (value instanceof Number) { + boolean v = ((Number) value).intValue() != 0; + pstmt.setBoolean(i, v); + } else { + pstmt.setNull(i, type); + } + break; + case Types.CHAR: + case Types.NCHAR: + case Types.VARCHAR: + case Types.LONGVARCHAR: + pstmt.setString(i, value.toString()); + break; + case Types.TINYINT: + if (value instanceof Number) { + pstmt.setByte(i, ((Number) value).byteValue()); + } else if (value instanceof String) { + pstmt.setByte(i, Byte.parseByte((String) value)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.SMALLINT: + if (value instanceof Number) { + pstmt.setShort(i, ((Number) value).shortValue()); + } else if (value instanceof String) { + pstmt.setShort(i, Short.parseShort((String) value)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.INTEGER: + if (value instanceof Number) { + pstmt.setInt(i, ((Number) value).intValue()); + } else if (value instanceof String) { + pstmt.setInt(i, Integer.parseInt((String) value)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.BIGINT: + if (value instanceof Number) { + pstmt.setLong(i, ((Number) value).longValue()); + } else if (value instanceof String) { + pstmt.setLong(i, Long.parseLong((String) value)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.DECIMAL: + case Types.NUMERIC: + pstmt.setBigDecimal(i, new BigDecimal(value.toString())); + break; + case Types.REAL: + if (value instanceof Number) { + pstmt.setFloat(i, ((Number) value).floatValue()); + } else if (value instanceof String) { + pstmt.setFloat(i, Float.parseFloat((String) value)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.FLOAT: + case Types.DOUBLE: + if (value instanceof Number) { + pstmt.setDouble(i, ((Number) value).doubleValue()); + } else if (value instanceof String) { + pstmt.setDouble(i, Double.parseDouble((String) value)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.BINARY: + case Types.VARBINARY: + case Types.LONGVARBINARY: + case Types.BLOB: + + if (value instanceof byte[]) { + pstmt.setBytes(i, (byte[]) value); + } else if (value instanceof String) { + pstmt.setBytes(i, ((String) value).getBytes(StandardCharsets.ISO_8859_1)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.CLOB: + if (value instanceof byte[]) { + pstmt.setBytes(i, (byte[]) value); + } else if (value instanceof String) { + Reader clobReader = new StringReader((String) value); + pstmt.setCharacterStream(i, clobReader); + } else { + pstmt.setNull(i, type); + } + break; + case Types.DATE: + if (value instanceof Date) { + pstmt.setDate(i, new java.sql.Date(((Date) value).getTime())); + } else if (value instanceof String) { + String v = (String) value; + if (!v.startsWith("0000-00-00")) { + v = v.trim().replace(" ", "T"); + DateTime dt = new DateTime(v); + pstmt.setDate(i, new java.sql.Date(dt.toDate().getTime())); + } else { + pstmt.setNull(i, type); + } + } else { + pstmt.setNull(i, type); + } + break; + case Types.TIME: + if (value instanceof Date) { + pstmt.setTime(i, new java.sql.Time(((Date) value).getTime())); + } else if (value instanceof String) { + String v = (String) value; + v = "T" + v; + DateTime dt = new DateTime(v); + pstmt.setTime(i, new Time(dt.toDate().getTime())); + } else { + pstmt.setNull(i, type); + } + break; + case Types.TIMESTAMP: + if (value instanceof Date) { + pstmt.setTimestamp(i, new java.sql.Timestamp(((Date) value).getTime())); + } else if (value instanceof String) { + String v = (String) value; + if (!v.startsWith("0000-00-00")) { + v = v.trim().replace(" ", "T"); + DateTime dt = new DateTime(v); + pstmt.setTimestamp(i, new Timestamp(dt.toDate().getTime())); + } else { + pstmt.setNull(i, type); + } + } else { + pstmt.setNull(i, type); + } + break; + default: + pstmt.setObject(i, value, type); + } + + } else { + pstmt.setNull(i, type); + } + i++; + } + } + } +} diff --git a/client-adapter/rdb/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter b/client-adapter/rdb/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter new file mode 100644 index 00000000..0a840275 --- /dev/null +++ b/client-adapter/rdb/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter @@ -0,0 +1 @@ +rdb=com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter \ No newline at end of file diff --git a/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml b/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml new file mode 100644 index 00000000..ff6058b0 --- /dev/null +++ b/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml @@ -0,0 +1,12 @@ +dataSourceKey: defaultDS +destination: example +dbMapping: + database: mytest + table: user + targetTable: mytest.tb_user + commitBatch: 3000 + mapAll: false + columns: + id: + name: + role_id: \ No newline at end of file diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java new file mode 100644 index 00000000..ace59b55 --- /dev/null +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java @@ -0,0 +1,29 @@ +package com.alibaba.otter.canal.client.adapter.rdb.test; + +import java.util.Map; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; +import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfigLoader; +import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; + +public class ConfigLoadTest { + + @Before + public void before() { + AdapterConfigs.put("rdb", "mytest_user.yml"); + // 加载数据源连接池 + DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource); + } + + @Test + public void testLoad() { + Map configMap = MappingConfigLoader.load(); + + Assert.assertFalse(configMap.isEmpty()); + } +} diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java new file mode 100644 index 00000000..54e6c447 --- /dev/null +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java @@ -0,0 +1,82 @@ +package com.alibaba.otter.canal.client.adapter.rdb.test; + +import java.io.BufferedReader; +import java.io.Reader; +import java.math.BigDecimal; +import java.sql.*; + +import org.junit.Test; + +import com.alibaba.druid.pool.DruidDataSource; + +public class DBTest { + + @Test + public void test01() throws SQLException { + DruidDataSource dataSource = new DruidDataSource(); + // dataSource.setDriverClassName("oracle.jdbc.OracleDriver"); + // dataSource.setUrl("jdbc:oracle:thin:@127.0.0.1:49161:XE"); + // dataSource.setUsername("mytest"); + // dataSource.setPassword("m121212"); + + dataSource.setDriverClassName("com.mysql.jdbc.Driver"); + dataSource.setUrl("jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true"); + dataSource.setUsername("root"); + dataSource.setPassword("121212"); + + dataSource.setInitialSize(1); + dataSource.setMinIdle(1); + dataSource.setMaxActive(2); + dataSource.setMaxWait(60000); + dataSource.setTimeBetweenEvictionRunsMillis(60000); + dataSource.setMinEvictableIdleTimeMillis(300000); + + dataSource.init(); + + Connection conn = dataSource.getConnection(); + Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery("select * from user t where 1=2"); + + ResultSetMetaData rsm = rs.getMetaData(); + int cnt = rsm.getColumnCount(); + for (int i = 1; i <= cnt; i++) { + System.out.println(rsm.getColumnName(i) + " " + rsm.getColumnType(i)); + } + + rs.close(); + stmt.close(); + +// PreparedStatement pstmt = conn +// .prepareStatement("insert into tb_user (id,name,role_id,c_time,test1,test2) values (?,?,?,?,?,?)"); +// pstmt.setBigDecimal(1, new BigDecimal("5")); +// pstmt.setString(2, "test"); +// pstmt.setBigDecimal(3, new BigDecimal("1")); +// pstmt.setDate(4, new Date(new java.util.Date().getTime())); +// byte[] a = { (byte) 1, (byte) 2 }; +// pstmt.setBytes(5, a); +// pstmt.setBytes(6, a); +// pstmt.execute(); +// +// pstmt.close(); + + conn.close(); + dataSource.close(); + } + + private String clob2Str(Clob clob) { + String content = ""; + try { + Reader is = clob.getCharacterStream(); + BufferedReader buff = new BufferedReader(is); + String line = buff.readLine(); + StringBuffer sb = new StringBuffer(); + while (line != null) { + sb.append(line); + line = buff.readLine(); + } + content = sb.toString(); + } catch (Exception e) { + } + return content; + } +} diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/TestConstant.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/TestConstant.java new file mode 100644 index 00000000..813f0e85 --- /dev/null +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/TestConstant.java @@ -0,0 +1,37 @@ +package com.alibaba.otter.canal.client.adapter.rdb.test; + +import java.sql.SQLException; + +import com.alibaba.druid.pool.DruidDataSource; + +public class TestConstant { + + public final static String jdbcUrl = "jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true"; + public final static String jdbcUser = "root"; + public final static String jdbcPassword = "121212"; + + public static DruidDataSource dataSource; + + static { + dataSource = new DruidDataSource(); + dataSource.setDriverClassName("com.mysql.jdbc.Driver"); + dataSource.setUrl(jdbcUrl); + dataSource.setUsername(jdbcUser); + dataSource.setPassword(jdbcPassword); + dataSource.setInitialSize(1); + dataSource.setMinIdle(1); + dataSource.setMaxActive(1); + dataSource.setMaxWait(60000); + dataSource.setTimeBetweenEvictionRunsMillis(60000); + dataSource.setMinEvictableIdleTimeMillis(300000); + dataSource.setPoolPreparedStatements(false); + dataSource.setMaxPoolPreparedStatementPerConnectionSize(20); + dataSource.setValidationQuery("select 1"); + try { + dataSource.init(); + } catch (SQLException e) { + e.printStackTrace(); + } + } + +} diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java new file mode 100644 index 00000000..3e7917de --- /dev/null +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java @@ -0,0 +1,28 @@ +package com.alibaba.otter.canal.client.adapter.rdb.test.sync; + +import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; +import com.alibaba.otter.canal.client.adapter.rdb.test.TestConstant; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; +import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; + +import java.util.HashMap; +import java.util.Map; + +public class Common { + public static RdbAdapter init() { + DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource); + + OuterAdapterConfig outerAdapterConfig = new OuterAdapterConfig(); + outerAdapterConfig.setName("rdb"); + Map properties = new HashMap<>(); + properties.put("jdbc.driveClassName", "oracle.jdbc.OracleDriver"); + properties.put("jdbc.url", "jdbc:oracle:thin:@127.0.0.1:49161:XE"); + properties.put("jdbc.username", "mytest"); + properties.put("jdbc.password", "m121212"); + outerAdapterConfig.setProperties(properties); + + RdbAdapter adapter = new RdbAdapter(); + adapter.init(outerAdapterConfig); + return adapter; + } +} diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java new file mode 100644 index 00000000..3324e41c --- /dev/null +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java @@ -0,0 +1,47 @@ +package com.alibaba.otter.canal.client.adapter.rdb.test.sync; + +import java.io.BufferedReader; +import java.io.Reader; +import java.sql.Clob; +import java.util.*; + +import org.junit.Before; +import org.junit.Test; + +import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; +import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.support.Dml; + +public class OracleSyncTest { + + private RdbAdapter rdbAdapter; + + @Before + public void init() { + AdapterConfigs.put("rdb", "mytest_user.yml"); + rdbAdapter = Common.init(); + } + + @Test + public void test01() { + Dml dml = new Dml(); + dml.setDestination("example"); + dml.setTs(new Date().getTime()); + dml.setType("INSERT"); + dml.setDatabase("mytest"); + dml.setTable("user"); + List> dataList = new ArrayList<>(); + Map data = new LinkedHashMap<>(); + dataList.add(data); + data.put("id", 1L); + data.put("name", "Eric"); + data.put("role_id", 1L); + data.put("c_time", new Date()); + data.put("test1", "sdfasdfawe中国asfwef"); + dml.setData(dataList); + + rdbAdapter.sync(dml); + } + + +} diff --git a/client-adapter/rdb/src/test/resources/log4j2-test.xml b/client-adapter/rdb/src/test/resources/log4j2-test.xml new file mode 100644 index 00000000..9c3d2ee6 --- /dev/null +++ b/client-adapter/rdb/src/test/resources/log4j2-test.xml @@ -0,0 +1,13 @@ + + + + + + + + + + + + + \ No newline at end of file diff --git a/client-adapter/rdb/src/test/resources/logback-test.xml b/client-adapter/rdb/src/test/resources/logback-test.xml new file mode 100644 index 00000000..91f0de80 --- /dev/null +++ b/client-adapter/rdb/src/test/resources/logback-test.xml @@ -0,0 +1,13 @@ + + + + + %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n + + + + + + + + \ No newline at end of file diff --git a/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml b/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml new file mode 100644 index 00000000..1f111b79 --- /dev/null +++ b/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml @@ -0,0 +1,14 @@ +dataSourceKey: defaultDS +destination: example +dbMapping: + database: mytest + table: user + targetTable: mytest.tb_user + commitBatch: 3000 + mapAll: false + targetColums: + id: + name: + role_id: + c_time: + test1: \ No newline at end of file From f7ea8df89e618c1d50844ad7c439759077df6314 Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 8 Nov 2018 11:19:50 +0800 Subject: [PATCH 3/4] =?UTF-8?q?rdb=E6=96=B0=E5=A2=9E=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E5=AE=8C=E6=88=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../adapter/rdb/config/MappingConfig.java | 37 +- .../adapter/rdb/service/RdbSyncService.java | 545 +++++++++++------- .../src/main/resources/rdb/mytest_user.yml | 2 +- .../src/test/resources/rdb/mytest_user.yml | 14 +- .../otter/canal/kafka/CanalKafkaProducer.java | 4 +- .../canal/rocketmq/CanalRocketMQProducer.java | 9 +- 6 files changed, 367 insertions(+), 244 deletions(-) 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 d0709db6..f7113d7f 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 @@ -56,16 +56,19 @@ public class MappingConfig { public static class DbMapping { - private String database; // 数据库名或schema名 - private String table; // 表面名 - private boolean mapAll = false; // 映射所有字段 - private String targetTable; // 目标表名 - private Map targetColums; // 目标表字段映射 - private String etlCondition; // etl条件sql + private String database; // 数据库名或schema名 + private String table; // 表面名 + private boolean mapAll = false; // 映射所有字段 + private String targetTable; // 目标表名 + private Map targetColumns; // 目标表字段映射 - private Set families = new LinkedHashSet<>(); // column family列表 - private int readBatch = 5000; - private int commitBatch = 5000; // etl等批量提交大小 + private String etlCondition; // etl条件sql + + private Set families = new LinkedHashSet<>(); // column family列表 + private int readBatch = 5000; + private int commitBatch = 5000; // etl等批量提交大小 + + private volatile Map allColumns; // mapAll为true,自动设置改字段 public String getDatabase() { return database; @@ -99,12 +102,12 @@ public class MappingConfig { this.targetTable = targetTable; } - public Map getTargetColums() { - return targetColums; + public Map getTargetColumns() { + return targetColumns; } - public void setTargetColums(Map targetColums) { - this.targetColums = targetColums; + public void setTargetColumns(Map targetColumns) { + this.targetColumns = targetColumns; } public String getEtlCondition() { @@ -138,5 +141,13 @@ public class MappingConfig { public void setCommitBatch(int commitBatch) { this.commitBatch = commitBatch; } + + 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/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index 9012b6d5..a6b46fd9 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 @@ -5,7 +5,7 @@ import java.io.StringReader; import java.math.BigDecimal; import java.nio.charset.StandardCharsets; import java.sql.*; -import java.util.Date; +import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -21,6 +21,7 @@ import org.slf4j.LoggerFactory; import com.alibaba.fastjson.JSON; 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.support.Dml; /** @@ -31,9 +32,12 @@ import com.alibaba.otter.canal.client.adapter.support.Dml; */ public class RdbSyncService { - private static Logger logger = LoggerFactory.getLogger(RdbSyncService.class); + private static final Logger logger = LoggerFactory + .getLogger(RdbSyncService.class); - private DataSource dataSource; + private static final Map> COLUMNS_TYPE_CACHE = new ConcurrentHashMap<>(); + + private DataSource dataSource; public RdbSyncService(DataSource dataSource){ this.dataSource = dataSource; @@ -42,6 +46,52 @@ public class RdbSyncService { public void sync(MappingConfig config, Dml dml) { try { if (config != null) { + { + 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<>(); + 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); + } + } + } + } + String type = dml.getType(); if (type != null && type.equalsIgnoreCase("INSERT")) { insert(config, dml); @@ -59,6 +109,72 @@ public class RdbSyncService { } } + private void update(MappingConfig config, Dml dml) throws SQLException { + List> data = dml.getData(); + if (data == null || data.isEmpty()) { + return; + } + + List> old = dml.getOld(); + if (old == null || old.isEmpty()) { + return; + } + + DbMapping dbMapping = config.getDbMapping(); + + int idx = 1; + boolean complete = false; + + Connection conn = dataSource.getConnection(); + boolean oriAutoCommit = conn.getAutoCommit(); + conn.setAutoCommit(false); + + try { + Map columnsMap; + if (dbMapping.isMapAll()) { + columnsMap = dbMapping.getAllColumns(); + } else { + columnsMap = dbMapping.getTargetColumns(); + } + + int i = 0; + for (Map o : old) { + Map d = data.get(i); + StringBuilder updateSql = new StringBuilder(); + updateSql.append("UPDATE ").append(dbMapping.getTargetTable()).append(" SET "); + List values = new ArrayList<>(); + boolean flag = false; + for (String srcColumnName : o.keySet()) { + List targetColumnNames = new ArrayList<>(); + columnsMap.forEach((targetColumn, srcColumn) -> { + if (srcColumnName.toLowerCase().equals(srcColumn)) { + targetColumnNames.add(targetColumn); + } + }); + if (!targetColumnNames.isEmpty()) { + if (!flag) { + flag = true; + } + for (String targetColumnName : targetColumnNames) { + updateSql.append(targetColumnName).append("=?, "); + values.add(data.get(i)); + } + } + } + if (flag) { + int len = updateSql.length(); + updateSql.delete(len - 2, len).append(" WHERE "); + } + i++; + } + } catch (Exception e) { + conn.rollback(); + } finally { + conn.setAutoCommit(oriAutoCommit); + conn.close(); + } + } + /** * 插入操作 * @@ -74,79 +190,68 @@ public class RdbSyncService { DbMapping dbMapping = config.getDbMapping(); int idx = 1; - boolean complete = false; + boolean completed = false; Connection conn = dataSource.getConnection(); - + boolean oriAutoCommit = conn.getAutoCommit(); conn.setAutoCommit(false); + try { + Map columnsMap; + if (dbMapping.isMapAll()) { + columnsMap = dbMapping.getAllColumns(); + } else { + columnsMap = dbMapping.getTargetColumns(); + } - StringBuilder insertSql = new StringBuilder(); - insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" ("); - if (!dbMapping.isMapAll()) { - dbMapping.getTargetColums().forEach((targetColumnName, srcColumnName) -> { - insertSql.append(targetColumnName).append(","); - }); + StringBuilder insertSql = new StringBuilder(); + insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" ("); + + columnsMap.forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(",")); int len = insertSql.length(); insertSql.delete(len - 1, len).append(") VALUES ("); - int mapLen = dbMapping.getTargetColums().size(); + int mapLen = columnsMap.size(); for (int i = 0; i < mapLen; i++) { insertSql.append("?,"); } len = insertSql.length(); insertSql.delete(len - 1, len).append(")"); - } - PreparedStatement pstmt = conn.prepareStatement(insertSql.toString()); + PreparedStatement pstmt = conn.prepareStatement(insertSql.toString()); - for (Map r : data) { - pstmt.clearParameters(); - convertData2DbRow(conn, config, r, pstmt); + for (Map r : data) { + pstmt.clearParameters(); + convertData2DbRow(conn, config, r, pstmt); - pstmt.execute(); + pstmt.execute(); + if (logger.isTraceEnabled()) { + logger.trace("Insert into target db, sql: {}", insertSql); + } - if (idx % config.getDbMapping().getCommitBatch() == 0) { - conn.commit(); - complete = true; + if (idx % config.getDbMapping().getCommitBatch() == 0) { + conn.commit(); + completed = true; + } + idx++; } - idx++; + if (!completed) { + conn.commit(); + } + } catch (Exception e) { + conn.rollback(); + } finally { + conn.setAutoCommit(oriAutoCommit); + conn.close(); } - if (!complete) { - conn.commit(); - } - conn.close(); } private static void sqlRS(Connection conn, String sql, Consumer consumer) { - Statement stmt = null; - ResultSet rs = null; - try { - stmt = conn.createStatement(); - rs = stmt.executeQuery(sql); + try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(sql)) { consumer.accept(rs); } catch (SQLException e) { logger.error(e.getMessage(), e); - } finally { - if (rs != null) { - try { - rs.close(); - rs = null; - } catch (SQLException e) { - // ignore - } - } - if (stmt != null) { - try { - stmt.close(); - stmt = null; - } catch (SQLException e) { - // ignore - } - } } } - private static Map> COLUMNS_TYPE_CACHE = new ConcurrentHashMap<>(); - private Map getTargetColumnType(Connection conn, MappingConfig config) { DbMapping dbMapping = config.getDbMapping(); String cacheKey = config.getDestination() + "." + dbMapping.getDatabase() + "." + dbMapping.getTable(); @@ -187,182 +292,188 @@ public class RdbSyncService { private void convertData2DbRow(Connection conn, MappingConfig config, Map data, PreparedStatement pstmt) throws SQLException { DbMapping dbMapping = config.getDbMapping(); + Map columnsMap; if (dbMapping.isMapAll()) { - + columnsMap = dbMapping.getAllColumns(); } else { - Map ctype = getTargetColumnType(conn, config); + columnsMap = dbMapping.getTargetColumns(); + } + Map ctype = getTargetColumnType(conn, config); - int i = 1; - for (Map.Entry entry : dbMapping.getTargetColums().entrySet()) { - String targetClassName = entry.getKey(); - String srcColumnName = entry.getValue(); - if (srcColumnName == null) { - srcColumnName = targetClassName; + int i = 1; + for (Map.Entry entry : columnsMap.entrySet()) { + String targetClassName = entry.getKey(); + String srcColumnName = entry.getValue(); + if (srcColumnName == null) { + srcColumnName = targetClassName; + } + + Integer type = ctype.get(targetClassName.toLowerCase()); + + Object value = data.get(srcColumnName); + if (value != null) { + if (type == null) { + throw new RuntimeException("No column: " + targetClassName + " found in target db"); } - Integer type = ctype.get(targetClassName.toLowerCase()); - - Object value = data.get(srcColumnName); - if (value != null) { - if (type == null) { - throw new RuntimeException("No column: " + targetClassName + " found in target db"); - } - switch (type) { - case Types.BIT: - case Types.BOOLEAN: - if (value instanceof Boolean) { - pstmt.setBoolean(i, (Boolean) value); - } else if (value instanceof String) { - boolean v = !value.equals("0"); - pstmt.setBoolean(i, v); - } else if (value instanceof Number) { - boolean v = ((Number) value).intValue() != 0; - pstmt.setBoolean(i, v); - } else { - pstmt.setNull(i, type); - } - break; - case Types.CHAR: - case Types.NCHAR: - case Types.VARCHAR: - case Types.LONGVARCHAR: - pstmt.setString(i, value.toString()); - break; - case Types.TINYINT: - if (value instanceof Number) { - pstmt.setByte(i, ((Number) value).byteValue()); - } else if (value instanceof String) { - pstmt.setByte(i, Byte.parseByte((String) value)); - } else { - pstmt.setNull(i, type); - } - break; - case Types.SMALLINT: - if (value instanceof Number) { - pstmt.setShort(i, ((Number) value).shortValue()); - } else if (value instanceof String) { - pstmt.setShort(i, Short.parseShort((String) value)); - } else { - pstmt.setNull(i, type); - } - break; - case Types.INTEGER: - if (value instanceof Number) { - pstmt.setInt(i, ((Number) value).intValue()); - } else if (value instanceof String) { - pstmt.setInt(i, Integer.parseInt((String) value)); - } else { - pstmt.setNull(i, type); - } - break; - case Types.BIGINT: - if (value instanceof Number) { - pstmt.setLong(i, ((Number) value).longValue()); - } else if (value instanceof String) { - pstmt.setLong(i, Long.parseLong((String) value)); - } else { - pstmt.setNull(i, type); - } - break; - case Types.DECIMAL: - case Types.NUMERIC: - pstmt.setBigDecimal(i, new BigDecimal(value.toString())); - break; - case Types.REAL: - if (value instanceof Number) { - pstmt.setFloat(i, ((Number) value).floatValue()); - } else if (value instanceof String) { - pstmt.setFloat(i, Float.parseFloat((String) value)); - } else { - pstmt.setNull(i, type); - } - break; - case Types.FLOAT: - case Types.DOUBLE: - if (value instanceof Number) { - pstmt.setDouble(i, ((Number) value).doubleValue()); - } else if (value instanceof String) { - pstmt.setDouble(i, Double.parseDouble((String) value)); - } else { - pstmt.setNull(i, type); - } - break; - case Types.BINARY: - case Types.VARBINARY: - case Types.LONGVARBINARY: - case Types.BLOB: - - if (value instanceof byte[]) { - pstmt.setBytes(i, (byte[]) value); - } else if (value instanceof String) { - pstmt.setBytes(i, ((String) value).getBytes(StandardCharsets.ISO_8859_1)); - } else { - pstmt.setNull(i, type); - } - break; - case Types.CLOB: - if (value instanceof byte[]) { - pstmt.setBytes(i, (byte[]) value); - } else if (value instanceof String) { - Reader clobReader = new StringReader((String) value); - pstmt.setCharacterStream(i, clobReader); - } else { - pstmt.setNull(i, type); - } - break; - case Types.DATE: - if (value instanceof Date) { - pstmt.setDate(i, new java.sql.Date(((Date) value).getTime())); - } else if (value instanceof String) { - String v = (String) value; - if (!v.startsWith("0000-00-00")) { - v = v.trim().replace(" ", "T"); - DateTime dt = new DateTime(v); - pstmt.setDate(i, new java.sql.Date(dt.toDate().getTime())); - } else { - pstmt.setNull(i, type); - } - } else { - pstmt.setNull(i, type); - } - break; - case Types.TIME: - if (value instanceof Date) { - pstmt.setTime(i, new java.sql.Time(((Date) value).getTime())); - } else if (value instanceof String) { - String v = (String) value; - v = "T" + v; - DateTime dt = new DateTime(v); - pstmt.setTime(i, new Time(dt.toDate().getTime())); - } else { - pstmt.setNull(i, type); - } - break; - case Types.TIMESTAMP: - if (value instanceof Date) { - pstmt.setTimestamp(i, new java.sql.Timestamp(((Date) value).getTime())); - } else if (value instanceof String) { - String v = (String) value; - if (!v.startsWith("0000-00-00")) { - v = v.trim().replace(" ", "T"); - DateTime dt = new DateTime(v); - pstmt.setTimestamp(i, new Timestamp(dt.toDate().getTime())); - } else { - pstmt.setNull(i, type); - } - } else { - pstmt.setNull(i, type); - } - break; - default: - pstmt.setObject(i, value, type); - } + setPStmt(type, pstmt, value, i); + } else { + pstmt.setNull(i, type); + } + i++; + } + } + private void setPStmt(int type, PreparedStatement pstmt, Object value, int i) throws SQLException { + switch (type) { + case Types.BIT: + case Types.BOOLEAN: + if (value instanceof Boolean) { + pstmt.setBoolean(i, (Boolean) value); + } else if (value instanceof String) { + boolean v = !value.equals("0"); + pstmt.setBoolean(i, v); + } else if (value instanceof Number) { + boolean v = ((Number) value).intValue() != 0; + pstmt.setBoolean(i, v); } else { pstmt.setNull(i, type); } - i++; - } + break; + case Types.CHAR: + case Types.NCHAR: + case Types.VARCHAR: + case Types.LONGVARCHAR: + pstmt.setString(i, value.toString()); + break; + case Types.TINYINT: + if (value instanceof Number) { + pstmt.setByte(i, ((Number) value).byteValue()); + } else if (value instanceof String) { + pstmt.setByte(i, Byte.parseByte((String) value)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.SMALLINT: + if (value instanceof Number) { + pstmt.setShort(i, ((Number) value).shortValue()); + } else if (value instanceof String) { + pstmt.setShort(i, Short.parseShort((String) value)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.INTEGER: + if (value instanceof Number) { + pstmt.setInt(i, ((Number) value).intValue()); + } else if (value instanceof String) { + pstmt.setInt(i, Integer.parseInt((String) value)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.BIGINT: + if (value instanceof Number) { + pstmt.setLong(i, ((Number) value).longValue()); + } else if (value instanceof String) { + pstmt.setLong(i, Long.parseLong((String) value)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.DECIMAL: + case Types.NUMERIC: + pstmt.setBigDecimal(i, new BigDecimal(value.toString())); + break; + case Types.REAL: + if (value instanceof Number) { + pstmt.setFloat(i, ((Number) value).floatValue()); + } else if (value instanceof String) { + pstmt.setFloat(i, Float.parseFloat((String) value)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.FLOAT: + case Types.DOUBLE: + if (value instanceof Number) { + pstmt.setDouble(i, ((Number) value).doubleValue()); + } else if (value instanceof String) { + pstmt.setDouble(i, Double.parseDouble((String) value)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.BINARY: + case Types.VARBINARY: + case Types.LONGVARBINARY: + case Types.BLOB: + + if (value instanceof byte[]) { + pstmt.setBytes(i, (byte[]) value); + } else if (value instanceof String) { + pstmt.setBytes(i, ((String) value).getBytes(StandardCharsets.ISO_8859_1)); + } else { + pstmt.setNull(i, type); + } + break; + case Types.CLOB: + if (value instanceof byte[]) { + pstmt.setBytes(i, (byte[]) value); + } else if (value instanceof String) { + Reader clobReader = new StringReader((String) value); + pstmt.setCharacterStream(i, clobReader); + } else { + pstmt.setNull(i, type); + } + break; + case Types.DATE: + if (value instanceof java.util.Date) { + pstmt.setDate(i, new java.sql.Date(((java.util.Date) value).getTime())); + } else if (value instanceof String) { + String v = (String) value; + if (!v.startsWith("0000-00-00")) { + v = v.trim().replace(" ", "T"); + DateTime dt = new DateTime(v); + pstmt.setDate(i, new Date(dt.toDate().getTime())); + } else { + pstmt.setNull(i, type); + } + } else { + pstmt.setNull(i, type); + } + break; + case Types.TIME: + if (value instanceof java.util.Date) { + pstmt.setTime(i, new java.sql.Time(((java.util.Date) value).getTime())); + } else if (value instanceof String) { + String v = (String) value; + v = "T" + v; + DateTime dt = new DateTime(v); + pstmt.setTime(i, new Time(dt.toDate().getTime())); + } else { + pstmt.setNull(i, type); + } + break; + case Types.TIMESTAMP: + if (value instanceof java.util.Date) { + pstmt.setTimestamp(i, new java.sql.Timestamp(((java.util.Date) value).getTime())); + } else if (value instanceof String) { + String v = (String) value; + if (!v.startsWith("0000-00-00")) { + v = v.trim().replace(" ", "T"); + DateTime dt = new DateTime(v); + pstmt.setTimestamp(i, new Timestamp(dt.toDate().getTime())); + } else { + pstmt.setNull(i, type); + } + } else { + pstmt.setNull(i, type); + } + break; + default: + pstmt.setObject(i, value, type); } } } diff --git a/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml b/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml index ff6058b0..d7bb4ae8 100644 --- a/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml +++ b/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml @@ -6,7 +6,7 @@ dbMapping: targetTable: mytest.tb_user commitBatch: 3000 mapAll: false - columns: + targetColumns: id: name: role_id: \ No newline at end of file diff --git a/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml b/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml index 1f111b79..09dda295 100644 --- a/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml +++ b/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml @@ -5,10 +5,10 @@ dbMapping: table: user targetTable: mytest.tb_user commitBatch: 3000 - mapAll: false - targetColums: - id: - name: - role_id: - c_time: - test1: \ No newline at end of file + mapAll: true +# targetColumns: +# id: +# name: +# role_id: +# c_time: +# test1: \ No newline at end of file diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index a4c9eaa8..78a67c13 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -104,7 +104,7 @@ public class CanalKafkaProducer implements CanalMQProducer { ProducerRecord record = new ProducerRecord(canalDestination.getTopic(), canalDestination.getPartition(), null, - JSON.toJSONString(flatMessage)); + JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); producer2.send(record).get(); } catch (Exception e) { logger.error(e.getMessage(), e); @@ -126,7 +126,7 @@ public class CanalKafkaProducer implements CanalMQProducer { ProducerRecord record = new ProducerRecord(canalDestination.getTopic(), i, null, - JSON.toJSONString(flatMessagePart)); + JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue)); producer2.send(record).get(); } catch (Exception e) { logger.error(e.getMessage(), e); diff --git a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java index d8bd45e4..3501930b 100644 --- a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java @@ -1,6 +1,7 @@ package com.alibaba.otter.canal.rocketmq; import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.common.CanalMessageSerializer; import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.protocol.FlatMessage; @@ -75,11 +76,11 @@ public class CanalRocketMQProducer implements CanalMQProducer { if (destination.getPartition() != null) { try { logger.info("send flat message: {} to topic: {} fixed partition: {}", - JSON.toJSONString(flatMessage), + JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue), destination.getTopic(), destination.getPartition()); Message message = new Message(destination.getTopic(), - JSON.toJSONString(flatMessage).getBytes()); + JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue).getBytes()); this.defaultMQProducer.send(message, new MessageQueueSelector() { @Override @@ -102,12 +103,12 @@ public class CanalRocketMQProducer implements CanalMQProducer { FlatMessage flatMessagePart = partitionFlatMessage[i]; if (flatMessagePart != null) { logger.debug("flatMessagePart: {}, partition: {}", - JSON.toJSONString(flatMessagePart), + JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue), i); final int index = i; try { Message message = new Message(destination.getTopic(), - JSON.toJSONString(flatMessagePart).getBytes()); + JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue).getBytes()); this.defaultMQProducer.send(message, new MessageQueueSelector() { @Override From 2ac84b4320bd06d8ee97ded033d741fd62920d5a Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 8 Nov 2018 19:34:15 +0800 Subject: [PATCH 4/4] =?UTF-8?q?=E5=85=B3=E7=B3=BB=E5=9E=8B=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=90=8C=E6=AD=A5adapter,=20=E6=94=AF=E6=8C=81=20mysq?= =?UTF-8?q?l=20oracle=20postgresql=20sqlserver?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/client/adapter/support/Util.java | 90 ++++ .../adapter/es/service/ESSyncService.java | 3 +- .../hbase/service/HbaseEtlService.java | 44 +- .../hbase/service/HbaseSyncService.java | 3 +- client-adapter/launcher/pom.xml | 13 + .../launcher/src/main/assembly/dev.xml | 7 + .../launcher/src/main/assembly/release.xml | 7 + .../loader/CanalAdapterKafkaWorker.java | 3 +- .../src/main/resources/application.yml | 15 + .../adapter/logger/LoggerAdapterExample.java | 6 +- .../canal/client/adapter/rdb/RdbAdapter.java | 161 ++++++- .../adapter/rdb/adapters/MysqlAdapter.java | 8 + .../adapter/rdb/adapters/OracleAdapter.java | 8 + .../rdb/adapters/PostgresqlAdapter.java | 8 + .../rdb/adapters/SqlserverAdapter.java | 8 + .../adapter/rdb/config/MappingConfig.java | 9 + .../rdb/config/MappingConfigLoader.java | 8 +- .../adapter/rdb/service/RdbEtlService.java | 290 +++++++++++++ .../adapter/rdb/service/RdbSyncService.java | 392 ++++++++++-------- ...ba.otter.canal.client.adapter.OuterAdapter | 5 +- .../resources/oracle}/mytest_user.yml | 3 +- .../main/resources/postgresql/mytest_user.yml | 15 + .../adapter/rdb/test/ConfigLoadTest.java | 4 +- .../client/adapter/rdb/test/sync/Common.java | 5 +- .../adapter/rdb/test/sync/OracleSyncTest.java | 27 +- .../resources/oracle}/mytest_user.yml | 15 +- 26 files changed, 906 insertions(+), 251 deletions(-) create mode 100644 client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Util.java create mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/MysqlAdapter.java create mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/OracleAdapter.java create mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/PostgresqlAdapter.java create mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/SqlserverAdapter.java create mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java rename client-adapter/rdb/src/{test/resources/rdb => main/resources/oracle}/mytest_user.yml (90%) create mode 100644 client-adapter/rdb/src/main/resources/postgresql/mytest_user.yml rename client-adapter/rdb/src/{main/resources/rdb => test/resources/oracle}/mytest_user.yml (51%) diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Util.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Util.java new file mode 100644 index 00000000..ac2c32c3 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Util.java @@ -0,0 +1,90 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.function.Consumer; +import java.util.function.Function; + +import javax.sql.DataSource; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class Util { + + private static final Logger logger = LoggerFactory.getLogger(Util.class); + + /** + * 通过DS执行sql + */ + public static Object sqlRS(DataSource ds, String sql, Function fun) throws SQLException { + Connection conn = null; + Statement stmt = null; + ResultSet rs = null; + try { + conn = ds.getConnection(); + stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); + stmt.setFetchSize(Integer.MIN_VALUE); + rs = stmt.executeQuery(sql); + return fun.apply(rs); + } finally { + if (rs != null) { + try { + rs.close(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } + if (stmt != null) { + try { + stmt.close(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } + if (conn != null) { + try { + conn.close(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } + } + } + + /** + * sql执行获取resultSet + * + * @param conn sql connection + * @param sql sql + * @param consumer 回调方法 + */ + public static void sqlRS(Connection conn, String sql, Consumer consumer) { + Statement stmt = null; + ResultSet rs = null; + try { + stmt = conn.createStatement(); + rs = stmt.executeQuery(sql); + consumer.accept(rs); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } finally { + if (rs != null) { + try { + rs.close(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } + if (stmt != null) { + try { + stmt.close(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } + } + } +} diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java index 272ef1a9..04d208f7 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java @@ -10,6 +10,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader; @@ -73,7 +74,7 @@ public class ESSyncService { dml.getDestination()); } if (logger.isDebugEnabled()) { - logger.debug("DML: {}", JSON.toJSONString(dml)); + logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); } } } diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java index 912baf08..172592e2 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java @@ -14,6 +14,7 @@ import java.util.function.Function; import javax.sql.DataSource; +import com.alibaba.otter.canal.client.adapter.support.Util; import org.apache.hadoop.hbase.util.Bytes; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -34,43 +35,6 @@ public class HbaseEtlService { private static Logger logger = LoggerFactory.getLogger(HbaseEtlService.class); - public static Object sqlRS(DataSource ds, String sql, Function fun) throws SQLException { - Connection conn = null; - Statement stmt = null; - ResultSet rs = null; - try { - conn = ds.getConnection(); - stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); - stmt.setFetchSize(Integer.MIN_VALUE); - rs = stmt.executeQuery(sql); - return fun.apply(rs); - } finally { - if (rs != null) { - try { - rs.close(); - } catch (SQLException e) { - logger.error(e.getMessage(), e); - } - } - if (stmt != null) { - try { - stmt.close(); - } catch (SQLException e) { - logger.error(e.getMessage(), e); - } - } - if (conn != null) { - try { - conn.close(); - } catch (SQLException e) { - logger.error(e.getMessage(), e); - } - } - rs = null; - stmt = null; - conn = null; - } - } /** * 建表 @@ -138,7 +102,7 @@ public class HbaseEtlService { if (params != null && params.size() == 1 && hbaseMapping.getEtlCondition() == null) { AtomicBoolean stExists = new AtomicBoolean(false); // 验证是否有SYS_TIME字段 - sqlRS(ds, sql, rs -> { + Util.sqlRS(ds, sql, rs -> { try { ResultSetMetaData rsmd = rs.getMetaData(); int cnt = rsmd.getColumnCount(); @@ -169,7 +133,7 @@ public class HbaseEtlService { // 获取总数 String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT "; - long cnt = (Long) sqlRS(ds, countSql, rs -> { + long cnt = (Long) Util.sqlRS(ds, countSql, rs -> { Long count = null; try { if (rs.next()) { @@ -244,7 +208,7 @@ public class HbaseEtlService { private static boolean executeSqlImport(DataSource ds, String sql, MappingConfig.HbaseMapping hbaseMapping, HbaseTemplate hbaseTemplate, AtomicLong successCount, List errMsg) { try { - sqlRS(ds, sql, rs -> { + Util.sqlRS(ds, sql, rs -> { int i = 1; try { diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java index 2a242093..3ec58369 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java @@ -7,6 +7,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.hbase.support.*; import com.alibaba.otter.canal.client.adapter.support.Dml; @@ -39,7 +40,7 @@ public class HbaseSyncService { delete(config, dml); } if (logger.isDebugEnabled()) { - logger.debug("DML: {}", JSON.toJSONString(dml)); + logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); } } } catch (Exception e) { diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index b6ed1c12..d48e6e71 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -107,6 +107,19 @@ jar-with-dependencies provided + + com.alibaba.otter + client-adapter.rdb + ${project.version} + + + * + * + + + jar-with-dependencies + provided + diff --git a/client-adapter/launcher/src/main/assembly/dev.xml b/client-adapter/launcher/src/main/assembly/dev.xml index 1be787b5..394fe33f 100644 --- a/client-adapter/launcher/src/main/assembly/dev.xml +++ b/client-adapter/launcher/src/main/assembly/dev.xml @@ -43,6 +43,13 @@ **/* + + ../rdb/src/main/resources/ + /config + + META-INF/** + + target logs diff --git a/client-adapter/launcher/src/main/assembly/release.xml b/client-adapter/launcher/src/main/assembly/release.xml index 57a815c8..163f2015 100644 --- a/client-adapter/launcher/src/main/assembly/release.xml +++ b/client-adapter/launcher/src/main/assembly/release.xml @@ -44,6 +44,13 @@ **/* + + ../rdb/src/main/resources/ + /config + + META-INF/** + + target logs diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java index 2e861642..045fee37 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java @@ -3,6 +3,7 @@ package com.alibaba.otter.canal.adapter.launcher.loader; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.apache.kafka.common.errors.WakeupException; @@ -36,7 +37,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { groupId, canalClientConfig.getBatchSize(), flatMessage); - // connector.setSessionTimeout(1L, TimeUnit.MINUTES); + connector.setSessionTimeout(30L, TimeUnit.SECONDS); } @Override diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 894d123e..2de61937 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -4,6 +4,7 @@ logging: level: com.alibaba.otter.canal.client.adapter.hbase: DEBUG com.alibaba.otter.canal.client.adapter.es: DEBUG + com.alibaba.otter.canal.client.adapter.rdb: DEBUG spring: jackson: date-format: yyyy-MM-dd HH:mm:ss @@ -36,6 +37,18 @@ spring: # - groupId: g2 # outAdapters: # - name: logger +# - name: oracle +# properties: +# jdbc.driverClassName: oracle.jdbc.OracleDriver +# jdbc.url: jdbc:oracle:thin:@127.0.0.1:49161:XE +# jdbc.username: mytest +# jdbc.password: m121212 +# - name: postgresql +# properties: +# jdbc.driverClassName: org.postgresql.Driver +# jdbc.url: jdbc:postgresql://localhost:5432/postgres +# jdbc.username: postgres +# jdbc.password: 121212 # mqTopics: # - mqMode: rocketmq # topic: example @@ -53,3 +66,5 @@ spring: # adapterConfigs: # - hbase/mytest_person2.yml # - es/mytest_user.yml +# - oracle/mytest_user.yml +# - postgresql/mytest_user.yml diff --git a/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java index 8f95b212..e9c693b7 100644 --- a/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java +++ b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java @@ -3,9 +3,11 @@ package com.alibaba.otter.canal.client.adapter.logger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.client.adapter.OuterAdapter; -import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; +import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; import com.alibaba.otter.canal.client.adapter.support.SPI; /** @@ -27,7 +29,7 @@ public class LoggerAdapterExample implements OuterAdapter { @Override public void sync(Dml dml) { - logger.info("DML: {}", dml.toString()); + logger.info("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); } @Override 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 d8fb9bcb..5dd47a05 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,8 +1,11 @@ package com.alibaba.otter.canal.client.adapter.rdb; +import java.sql.Connection; +import java.sql.ResultSetMetaData; import java.sql.SQLException; -import java.util.HashMap; -import java.util.Map; +import java.util.*; + +import javax.sql.DataSource; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; @@ -12,29 +15,28 @@ 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.MappingConfigLoader; +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.Dml; -import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; -import com.alibaba.otter.canal.client.adapter.support.SPI; +import com.alibaba.otter.canal.client.adapter.support.*; -@SPI("rdb") -public class RdbAdapter implements OuterAdapter { +public abstract class RdbAdapter implements OuterAdapter { - private static Logger logger = LoggerFactory.getLogger(RdbAdapter.class); + private static Logger logger = LoggerFactory.getLogger(RdbAdapter.class); - private static volatile Map rdbMapping = null; // 文件名对应配置 - private static volatile Map mappingConfigCache = null; // 库名-表名对应配置 + private volatile Map rdbMapping = null; // 文件名对应配置 + private volatile Map mappingConfigCache = null; // 库名-表名对应配置 - private DruidDataSource dataSource; + private DruidDataSource dataSource; - private RdbSyncService rdbSyncService; + private RdbSyncService rdbSyncService; @Override public void init(OuterAdapterConfig configuration) { if (mappingConfigCache == null) { synchronized (MappingConfig.class) { if (mappingConfigCache == null) { - rdbMapping = MappingConfigLoader.load(); + SPI spi = this.getClass().getAnnotation(SPI.class); + rdbMapping = MappingConfigLoader.load(spi.value()); mappingConfigCache = new HashMap<>(); for (MappingConfig mappingConfig : rdbMapping.values()) { mappingConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." @@ -65,6 +67,55 @@ 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); } @@ -74,9 +125,93 @@ public class RdbAdapter implements OuterAdapter { String database = dml.getDatabase(); String table = dml.getTable(); MappingConfig config = mappingConfigCache.get(destination + "." + database + "." + table); + rdbSyncService.sync(config, dml); } + @Override + public EtlResult etl(String task, List params) { + EtlResult etlResult = new EtlResult(); + MappingConfig config = rdbMapping.get(task); + if (config != null) { + DataSource srcDataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); + if (srcDataSource != null) { + return RdbEtlService.importData(srcDataSource, dataSource, config, params); + } else { + etlResult.setSucceeded(false); + etlResult.setErrorMessage("DataSource not found"); + return etlResult; + } + } else { + StringBuilder resultMsg = new StringBuilder(); + boolean resSucc = true; + // ds不为空说明传入的是destination + for (MappingConfig configTmp : rdbMapping.values()) { + // 取所有的destination为task的配置 + if (configTmp.getDestination().equals(task)) { + DataSource srcDataSource = DatasourceConfig.DATA_SOURCES.get(configTmp.getDataSourceKey()); + if (srcDataSource == null) { + continue; + } + EtlResult etlRes = RdbEtlService.importData(srcDataSource, dataSource, configTmp, params); + if (!etlRes.getSucceeded()) { + resSucc = false; + resultMsg.append(etlRes.getErrorMessage()).append("\n"); + } else { + resultMsg.append(etlRes.getResultMessage()).append("\n"); + } + } + } + if (resultMsg.length() > 0) { + etlResult.setSucceeded(resSucc); + if (resSucc) { + etlResult.setResultMessage(resultMsg.toString()); + } else { + etlResult.setErrorMessage(resultMsg.toString()); + } + return etlResult; + } + } + etlResult.setSucceeded(false); + etlResult.setErrorMessage("Task not found"); + return etlResult; + } + + @Override + public Map count(String task) { + MappingConfig config = rdbMapping.get(task); + MappingConfig.DbMapping dbMapping = config.getDbMapping(); + String sql = "SELECT COUNT(1) AS cnt FROM " + dbMapping.getTargetTable(); + Connection conn = null; + Map res = new LinkedHashMap<>(); + try { + conn = dataSource.getConnection(); + Util.sqlRS(conn, sql, rs -> { + try { + if (rs.next()) { + Long rowCount = rs.getLong("cnt"); + res.put("count", rowCount); + } + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + }); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } finally { + if (conn != null) { + try { + conn.close(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } + } + res.put("targetTable", dbMapping.getTargetTable()); + + return res; + } + @Override public void destroy() { if (dataSource != null) { diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/MysqlAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/MysqlAdapter.java new file mode 100644 index 00000000..ea73b1c0 --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/MysqlAdapter.java @@ -0,0 +1,8 @@ +package com.alibaba.otter.canal.client.adapter.rdb.adapters; + +import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; +import com.alibaba.otter.canal.client.adapter.support.SPI; + +@SPI("mysql") +public class MysqlAdapter extends RdbAdapter { +} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/OracleAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/OracleAdapter.java new file mode 100644 index 00000000..fe6111a2 --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/OracleAdapter.java @@ -0,0 +1,8 @@ +package com.alibaba.otter.canal.client.adapter.rdb.adapters; + +import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; +import com.alibaba.otter.canal.client.adapter.support.SPI; + +@SPI("oracle") +public class OracleAdapter extends RdbAdapter { +} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/PostgresqlAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/PostgresqlAdapter.java new file mode 100644 index 00000000..2cf4a444 --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/PostgresqlAdapter.java @@ -0,0 +1,8 @@ +package com.alibaba.otter.canal.client.adapter.rdb.adapters; + +import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; +import com.alibaba.otter.canal.client.adapter.support.SPI; + +@SPI("postgresql") +public class PostgresqlAdapter extends RdbAdapter { +} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/SqlserverAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/SqlserverAdapter.java new file mode 100644 index 00000000..2fe7e0ed --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/SqlserverAdapter.java @@ -0,0 +1,8 @@ +package com.alibaba.otter.canal.client.adapter.rdb.adapters; + +import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; +import com.alibaba.otter.canal.client.adapter.support.SPI; + +@SPI("sqlserver") +public class SqlserverAdapter extends RdbAdapter { +} 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 f7113d7f..da7e6ef8 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 @@ -58,6 +58,7 @@ public class MappingConfig { private String database; // 数据库名或schema名 private String table; // 表面名 + private Map targetPk; // 目标表主键字段 private boolean mapAll = false; // 映射所有字段 private String targetTable; // 目标表名 private Map targetColumns; // 目标表字段映射 @@ -86,6 +87,14 @@ public class MappingConfig { this.table = table; } + public Map getTargetPk() { + return targetPk; + } + + public void setTargetPk(Map targetPk) { + this.targetPk = targetPk; + } + public boolean isMapAll() { return mapAll; } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java index 8d3613a8..8658d45e 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java @@ -25,19 +25,17 @@ public class MappingConfigLoader { private static Logger logger = LoggerFactory.getLogger(MappingConfigLoader.class); - private static final String BASE_PATH = "rdb"; - /** * 加载HBase表映射配置 * * @return 配置名/配置文件名--对象 */ - public static Map load() { + public static Map load(String name) { logger.info("## Start loading rdb mapping config ... "); Map result = new LinkedHashMap<>(); - Collection configs = AdapterConfigs.get("rdb"); + Collection configs = AdapterConfigs.get(name); if (configs == null) { return result; } @@ -53,7 +51,7 @@ public class MappingConfigLoader { String configContent = null; if (c.endsWith(".yml")) { - configContent = readConfigContent(BASE_PATH + "/" + c); + configContent = readConfigContent(name + "/" + c); } MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class); 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 new file mode 100644 index 00000000..321c9f21 --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java @@ -0,0 +1,290 @@ +package com.alibaba.otter.canal.client.adapter.rdb.service; + +import java.sql.*; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +import javax.sql.DataSource; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +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.EtlResult; +import com.alibaba.otter.canal.client.adapter.support.Util; +import com.google.common.base.Joiner; + +/** + * RDB ETL 操作业务类 + * + * @author rewerma @ 2018-11-7 + * @version 1.0.0 + */ +public class RdbEtlService { + + private static final Logger logger = LoggerFactory.getLogger(RdbEtlService.class); + + /** + * 导入数据 + */ + public static EtlResult importData(DataSource srcDS, DataSource targetDS, MappingConfig config, + List params) { + EtlResult etlResult = new EtlResult(); + AtomicLong successCount = new AtomicLong(); + List errMsg = new ArrayList<>(); + String hbaseTable = ""; + try { + if (config == null) { + logger.error("Config is null!"); + etlResult.setSucceeded(false); + etlResult.setErrorMessage("Config is null!"); + return etlResult; + } + DbMapping dbMapping = config.getDbMapping(); + + long start = System.currentTimeMillis(); + + // 拼接sql + StringBuilder sql = new StringBuilder( + "SELECT * FROM " + dbMapping.getDatabase() + "." + dbMapping.getTable()); + + // 拼接条件 + appendCondition(params, dbMapping, srcDS, sql); + + // 获取总数 + String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT "; + long cnt = (Long) Util.sqlRS(srcDS, countSql, rs -> { + Long count = null; + try { + if (rs.next()) { + count = ((Number) rs.getObject(1)).longValue(); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + return count == null ? 0 : count; + }); + + // 当大于1万条记录时开启多线程 + if (cnt >= 10000) { + int threadCount = 3; + long perThreadCnt = cnt / threadCount; + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + List> futures = new ArrayList<>(threadCount); + for (int i = 0; i < threadCount; i++) { + long offset = i * perThreadCnt; + Long size = null; + if (i != threadCount - 1) { + size = perThreadCnt; + } + String sqlFinal; + if (size != null) { + sqlFinal = sql + " LIMIT " + offset + "," + size; + } else { + sqlFinal = sql + " LIMIT " + offset + "," + cnt; + } + Future future = executor.submit( + () -> executeSqlImport(params, srcDS, targetDS, sqlFinal, dbMapping, successCount, errMsg)); + futures.add(future); + } + + for (Future future : futures) { + future.get(); + } + + executor.shutdown(); + } else { + executeSqlImport(params, srcDS, targetDS, sql.toString(), dbMapping, successCount, errMsg); + } + + logger.info( + dbMapping.getTable() + " etl completed in: " + (System.currentTimeMillis() - start) / 1000 + "s!"); + + etlResult.setResultMessage("导入目标表 " + dbMapping.getTargetTable() + " 数据：" + successCount.get() + " 条"); + } catch (Exception e) { + logger.error(e.getMessage(), e); + errMsg.add(hbaseTable + " etl failed! ==>" + e.getMessage()); + } + + if (errMsg.isEmpty()) { + etlResult.setSucceeded(true); + } else { + etlResult.setErrorMessage(Joiner.on("\n").join(errMsg)); + } + return etlResult; + } + + private static void appendCondition(List params, DbMapping dbMapping, DataSource ds, + StringBuilder sql) throws SQLException { + if (params != null && params.size() == 1 && dbMapping.getEtlCondition() == null) { + AtomicBoolean stExists = new AtomicBoolean(false); + // 验证是否有SYS_TIME字段 + Util.sqlRS(ds, sql.toString(), rs -> { + try { + ResultSetMetaData rsmd = rs.getMetaData(); + int cnt = rsmd.getColumnCount(); + for (int i = 1; i <= cnt; i++) { + String columnName = rsmd.getColumnName(i); + if ("SYS_TIME".equalsIgnoreCase(columnName)) { + stExists.set(true); + break; + } + } + } catch (Exception e) { + // ignore + } + return null; + }); + if (stExists.get()) { + sql.append(" WHERE SYS_TIME >= '").append(params.get(0)).append("' "); + } + } else if (dbMapping.getEtlCondition() != null && params != null) { + String etlCondition = dbMapping.getEtlCondition(); + int size = params.size(); + for (int i = 0; i < size; i++) { + etlCondition = etlCondition.replace("{" + i + "}", params.get(i)); + } + + sql.append(" ").append(etlCondition); + } + } + + /** + * 执行导入 + */ + private static boolean executeSqlImport(List params, DataSource srcDS, DataSource targetDS, String sql, + DbMapping dbMapping, AtomicLong successCount, List errMsg) { + try { + Util.sqlRS(srcDS, sql, rs -> { + int idx = 1; + + try { + boolean completed = false; + + Map columnType = new LinkedHashMap<>(); + ResultSetMetaData rsd = rs.getMetaData(); + int columnCount = rsd.getColumnCount(); + for (int i = 1; i <= columnCount; i++) { + columnType.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i)); + } + + Map columnsMap; + if (dbMapping.isMapAll()) { + columnsMap = dbMapping.getAllColumns(); + } else { + columnsMap = dbMapping.getTargetColumns(); + } + + StringBuilder insertSql = new StringBuilder(); + insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" ("); + columnsMap + .forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(",")); + + int len = insertSql.length(); + insertSql.delete(len - 1, len).append(") VALUES ("); + int mapLen = columnsMap.size(); + for (int i = 0; i < mapLen; i++) { + insertSql.append("?,"); + } + len = insertSql.length(); + insertSql.delete(len - 1, len).append(")"); + try (Connection connTarget = targetDS.getConnection(); + PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) { + connTarget.setAutoCommit(false); + + while (rs.next()) { + pstmt.clearParameters(); + + // 删除数据 + Map values = new LinkedHashMap<>(); + StringBuilder deleteSql = new StringBuilder( + "DELETE FROM " + dbMapping.getTargetTable() + " WHERE "); + appendCondition(dbMapping, deleteSql, values, rs); + try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) { + int k = 1; + for (Object val : values.values()) { + pstmt2.setObject(k++, val); + } + pstmt2.execute(); + } + + int i = 1; + for (Map.Entry entry : columnsMap.entrySet()) { + String targetClolumnName = entry.getKey(); + String srcColumnName = entry.getValue(); + if (srcColumnName == null) { + srcColumnName = targetClolumnName; + } + + Integer type = columnType.get(targetClolumnName.toLowerCase()); + + Object value = rs.getObject(srcColumnName); + if (value != null) { + RdbSyncService.setPStmt(type, pstmt, value, i); + } else { + pstmt.setNull(i, type); + } + + i++; + } + + pstmt.execute(); + if (logger.isTraceEnabled()) { + logger.trace("Insert into target table, sql: {}", insertSql); + } + + if (idx % dbMapping.getCommitBatch() == 0) { + connTarget.commit(); + completed = true; + } + idx++; + successCount.incrementAndGet(); + if (logger.isDebugEnabled()) { + logger.debug("successful import count:" + successCount.get()); + } + } + if (!completed) { + connTarget.commit(); + } + } + + } catch (Exception e) { + logger.error(dbMapping.getTable() + " etl failed! ==>" + e.getMessage(), e); + errMsg.add(dbMapping.getTable() + " etl failed! ==>" + e.getMessage()); + } + return idx; + }); + return true; + } catch (Exception e) { + logger.error(e.getMessage(), e); + return false; + } + } + + /** + * 拼接目标表主键where条件 + */ + private static void appendCondition(DbMapping dbMapping, StringBuilder sql, Map values, + ResultSet rs) throws SQLException { + // 拼接主键 + for (Map.Entry entry : dbMapping.getTargetPk().entrySet()) { + String targetColumnName = entry.getKey(); + String srcColumnName = entry.getValue(); + if (srcColumnName == null) { + srcColumnName = targetColumnName; + } + sql.append(targetColumnName).append("=? AND "); + values.put(targetColumnName, rs.getObject(srcColumnName)); + } + int len = sql.length(); + sql.delete(len - 4, len); + } +} 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 a6b46fd9..ab46bf4b 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 @@ -14,11 +14,13 @@ 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; 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; @@ -32,12 +34,11 @@ import com.alibaba.otter.canal.client.adapter.support.Dml; */ public class RdbSyncService { - private static final Logger logger = LoggerFactory - .getLogger(RdbSyncService.class); + private static final Logger logger = LoggerFactory.getLogger(RdbSyncService.class); - private static final Map> COLUMNS_TYPE_CACHE = new ConcurrentHashMap<>(); + private final Map> COLUMNS_TYPE_CACHE = new ConcurrentHashMap<>(); - private DataSource dataSource; + private DataSource dataSource; public RdbSyncService(DataSource dataSource){ this.dataSource = dataSource; @@ -46,62 +47,16 @@ public class RdbSyncService { public void sync(MappingConfig config, Dml dml) { try { if (config != null) { - { - 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<>(); - 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); - } - } - } - } - String type = dml.getType(); if (type != null && type.equalsIgnoreCase("INSERT")) { insert(config, dml); } else if (type != null && type.equalsIgnoreCase("UPDATE")) { - // update(config, dml); + update(config, dml); } else if (type != null && type.equalsIgnoreCase("DELETE")) { - // delete(config, dml); + delete(config, dml); } if (logger.isDebugEnabled()) { - logger.debug("DML: {}", JSON.toJSONString(dml)); + logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); } } } catch (Exception e) { @@ -109,72 +64,6 @@ public class RdbSyncService { } } - private void update(MappingConfig config, Dml dml) throws SQLException { - List> data = dml.getData(); - if (data == null || data.isEmpty()) { - return; - } - - List> old = dml.getOld(); - if (old == null || old.isEmpty()) { - return; - } - - DbMapping dbMapping = config.getDbMapping(); - - int idx = 1; - boolean complete = false; - - Connection conn = dataSource.getConnection(); - boolean oriAutoCommit = conn.getAutoCommit(); - conn.setAutoCommit(false); - - try { - Map columnsMap; - if (dbMapping.isMapAll()) { - columnsMap = dbMapping.getAllColumns(); - } else { - columnsMap = dbMapping.getTargetColumns(); - } - - int i = 0; - for (Map o : old) { - Map d = data.get(i); - StringBuilder updateSql = new StringBuilder(); - updateSql.append("UPDATE ").append(dbMapping.getTargetTable()).append(" SET "); - List values = new ArrayList<>(); - boolean flag = false; - for (String srcColumnName : o.keySet()) { - List targetColumnNames = new ArrayList<>(); - columnsMap.forEach((targetColumn, srcColumn) -> { - if (srcColumnName.toLowerCase().equals(srcColumn)) { - targetColumnNames.add(targetColumn); - } - }); - if (!targetColumnNames.isEmpty()) { - if (!flag) { - flag = true; - } - for (String targetColumnName : targetColumnNames) { - updateSql.append(targetColumnName).append("=?, "); - values.add(data.get(i)); - } - } - } - if (flag) { - int len = updateSql.length(); - updateSql.delete(len - 2, len).append(" WHERE "); - } - i++; - } - } catch (Exception e) { - conn.rollback(); - } finally { - conn.setAutoCommit(oriAutoCommit); - conn.close(); - } - } - /** * 插入操作 * @@ -193,7 +82,6 @@ public class RdbSyncService { boolean completed = false; Connection conn = dataSource.getConnection(); - boolean oriAutoCommit = conn.getAutoCommit(); conn.setAutoCommit(false); try { Map columnsMap; @@ -216,15 +104,38 @@ public class RdbSyncService { len = insertSql.length(); insertSql.delete(len - 1, len).append(")"); + Map ctype = getTargetColumnType(conn, config); + PreparedStatement pstmt = conn.prepareStatement(insertSql.toString()); - for (Map r : data) { + for (Map d : data) { pstmt.clearParameters(); - convertData2DbRow(conn, config, r, pstmt); + int i = 1; + for (Map.Entry entry : columnsMap.entrySet()) { + String targetClolumnName = entry.getKey(); + String srcColumnName = entry.getValue(); + if (srcColumnName == null) { + srcColumnName = targetClolumnName; + } + + Integer type = ctype.get(targetClolumnName.toLowerCase()); + + Object value = d.get(srcColumnName); + if (value != null) { + if (type == null) { + throw new RuntimeException("No column: " + targetClolumnName + " found in target db"); + } + + setPStmt(type, pstmt, value, i); + } else { + pstmt.setNull(i, type); + } + i++; + } pstmt.execute(); if (logger.isTraceEnabled()) { - logger.trace("Insert into target db, sql: {}", insertSql); + logger.trace("Insert into target table, sql: {}", insertSql); } if (idx % config.getDbMapping().getCommitBatch() == 0) { @@ -239,19 +150,154 @@ public class RdbSyncService { } catch (Exception e) { conn.rollback(); } finally { - conn.setAutoCommit(oriAutoCommit); conn.close(); } } - private static void sqlRS(Connection conn, String sql, Consumer consumer) { - try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(sql)) { - consumer.accept(rs); - } catch (SQLException e) { - logger.error(e.getMessage(), e); + /** + * 更新操作 + * + * @param config 配置项 + * @param dml DML数据 + */ + private void update(MappingConfig config, Dml dml) throws SQLException { + List> data = dml.getData(); + if (data == null || data.isEmpty()) { + return; + } + + List> old = dml.getOld(); + if (old == null || old.isEmpty()) { + return; + } + + DbMapping dbMapping = config.getDbMapping(); + + int idx = 1; + boolean completed = false; + + Connection conn = dataSource.getConnection(); + conn.setAutoCommit(false); + + try { + Map columnsMap; + if (dbMapping.isMapAll()) { + columnsMap = dbMapping.getAllColumns(); + } else { + columnsMap = dbMapping.getTargetColumns(); + } + + Map ctype = getTargetColumnType(conn, config); + + for (Map o : old) { + Map d = data.get(idx - 1); + StringBuilder updateSql = new StringBuilder(); + updateSql.append("UPDATE ").append(dbMapping.getTargetTable()).append(" SET "); + Map values = new LinkedHashMap<>(); + for (String srcColumnName : o.keySet()) { + List targetColumnNames = new ArrayList<>(); + columnsMap.forEach((targetColumn, srcColumn) -> { + if (srcColumnName.toLowerCase().equals(srcColumn)) { + targetColumnNames.add(targetColumn); + } + }); + if (!targetColumnNames.isEmpty()) { + + for (String targetColumnName : targetColumnNames) { + updateSql.append(targetColumnName).append("=?, "); + values.put(targetColumnName, d.get(srcColumnName)); + } + } + } + int len = updateSql.length(); + updateSql.delete(len - 2, len).append(" WHERE "); + + // 拼接主键 + appendCondition(dbMapping, updateSql, values, d); + + sqlExe(conn, updateSql.toString(), ctype, values); + + if (logger.isTraceEnabled()) { + logger.trace("Update target table, sql: {}", updateSql); + } + if (idx % config.getDbMapping().getCommitBatch() == 0) { + conn.commit(); + completed = true; + } + idx++; + } + if (!completed) { + conn.commit(); + } + } catch (Exception e) { + conn.rollback(); + } finally { + conn.close(); } } + /** + * 删除操作 + * + * @param config + * @param dml + * @throws SQLException + */ + private void delete(MappingConfig config, Dml dml) throws SQLException { + List> data = dml.getData(); + if (data == null || data.isEmpty()) { + return; + } + + DbMapping dbMapping = config.getDbMapping(); + + int idx = 1; + boolean completed = false; + + Connection conn = dataSource.getConnection(); + conn.setAutoCommit(false); + + try { + Map ctype = getTargetColumnType(conn, config); + + for (Map d : data) { + StringBuilder sql = new StringBuilder(); + sql.append("DELETE FROM ").append(dbMapping.getTargetTable()).append(" WHERE "); + + Map values = new LinkedHashMap<>(); + // 拼接主键 + appendCondition(dbMapping, sql, values, d); + + sqlExe(conn, sql.toString(), ctype, values); + + if (logger.isTraceEnabled()) { + logger.trace("Delete from target table, sql: {}", sql); + } + if (idx % config.getDbMapping().getCommitBatch() == 0) { + conn.commit(); + completed = true; + } + idx++; + } + if (!completed) { + conn.commit(); + } + } catch (Exception e) { + conn.rollback(); + } finally { + conn.close(); + } + } + + + + /** + * 获取目标字段类型 + * + * @param conn sql connection + * @param config 映射配置 + * @return 字段sqlType + */ private Map getTargetColumnType(Connection conn, MappingConfig config) { DbMapping dbMapping = config.getDbMapping(); String cacheKey = config.getDestination() + "." + dbMapping.getDatabase() + "." + dbMapping.getTable(); @@ -263,7 +309,7 @@ public class RdbSyncService { columnType = new LinkedHashMap<>(); final Map columnTypeTmp = columnType; String sql = "SELECT * FROM " + dbMapping.getTargetTable() + " WHERE 1=2"; - sqlRS(conn, sql, rs -> { + Util.sqlRS(conn, sql, rs -> { try { ResultSetMetaData rsd = rs.getMetaData(); int columnCount = rsd.getColumnCount(); @@ -282,49 +328,14 @@ public class RdbSyncService { } /** - * 新增类型转换 - * - * @param config - * @param data - * @param pstmt - * @throws SQLException + * 设置 preparedStatement + * + * @param type sqlType + * @param pstmt 需要设置的preparedStatement + * @param value 值 + * @param i 索引号 */ - private void convertData2DbRow(Connection conn, MappingConfig config, Map data, - PreparedStatement pstmt) throws SQLException { - DbMapping dbMapping = config.getDbMapping(); - Map columnsMap; - if (dbMapping.isMapAll()) { - columnsMap = dbMapping.getAllColumns(); - } else { - columnsMap = dbMapping.getTargetColumns(); - } - Map ctype = getTargetColumnType(conn, config); - - int i = 1; - for (Map.Entry entry : columnsMap.entrySet()) { - String targetClassName = entry.getKey(); - String srcColumnName = entry.getValue(); - if (srcColumnName == null) { - srcColumnName = targetClassName; - } - - Integer type = ctype.get(targetClassName.toLowerCase()); - - Object value = data.get(srcColumnName); - if (value != null) { - if (type == null) { - throw new RuntimeException("No column: " + targetClassName + " found in target db"); - } - - setPStmt(type, pstmt, value, i); - } else { - pstmt.setNull(i, type); - } - i++; - } - } - - private void setPStmt(int type, PreparedStatement pstmt, Object value, int i) throws SQLException { + public static void setPStmt(int type, PreparedStatement pstmt, Object value, int i) throws SQLException { switch (type) { case Types.BIT: case Types.BOOLEAN: @@ -476,4 +487,41 @@ public class RdbSyncService { pstmt.setObject(i, value, type); } } + + /** + * 拼接主键 where条件 + */ + private static void appendCondition(DbMapping dbMapping, StringBuilder sql, Map values, + Map d) { + // 拼接主键 + for (Map.Entry entry : dbMapping.getTargetPk().entrySet()) { + String targetColumnName = entry.getKey(); + String srcColumnName = entry.getValue(); + if (srcColumnName == null) { + srcColumnName = targetColumnName; + } + sql.append(targetColumnName).append("=? AND "); + values.put(targetColumnName, d.get(srcColumnName)); + } + int len = sql.length(); + sql.delete(len - 4, len); + } + + /** + * 执行sql + */ + private static void sqlExe(Connection conn, String sql, Map ctype, Map values) { + try (PreparedStatement pstmt = conn.prepareStatement(sql)) { + int i = 1; + for (Map.Entry entry : values.entrySet()) { + String targetColumnName = entry.getKey(); + Object value = entry.getValue(); + Integer type = ctype.get(targetColumnName.toLowerCase()); + setPStmt(type, pstmt, value, i++); + } + pstmt.execute(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } } diff --git a/client-adapter/rdb/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter b/client-adapter/rdb/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter index 0a840275..c005f7fd 100644 --- a/client-adapter/rdb/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter +++ b/client-adapter/rdb/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter @@ -1 +1,4 @@ -rdb=com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter \ No newline at end of file +mysql=com.alibaba.otter.canal.client.adapter.rdb.adapters.MysqlAdapter +postgresql=com.alibaba.otter.canal.client.adapter.rdb.adapters.PostgresqlAdapter +oracle=com.alibaba.otter.canal.client.adapter.rdb.adapters.OracleAdapter +sqlserver=com.alibaba.otter.canal.client.adapter.rdb.adapters.SqlserverAdapter \ No newline at end of file diff --git a/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml b/client-adapter/rdb/src/main/resources/oracle/mytest_user.yml similarity index 90% rename from client-adapter/rdb/src/test/resources/rdb/mytest_user.yml rename to client-adapter/rdb/src/main/resources/oracle/mytest_user.yml index 09dda295..a7bede03 100644 --- a/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml +++ b/client-adapter/rdb/src/main/resources/oracle/mytest_user.yml @@ -4,7 +4,8 @@ dbMapping: database: mytest table: user targetTable: mytest.tb_user - commitBatch: 3000 + targetPk: + id: id mapAll: true # targetColumns: # id: diff --git a/client-adapter/rdb/src/main/resources/postgresql/mytest_user.yml b/client-adapter/rdb/src/main/resources/postgresql/mytest_user.yml new file mode 100644 index 00000000..1b4eaed7 --- /dev/null +++ b/client-adapter/rdb/src/main/resources/postgresql/mytest_user.yml @@ -0,0 +1,15 @@ +dataSourceKey: defaultDS +destination: example +dbMapping: + database: mytest + table: user + targetTable: public.tb_user + targetPk: + id: id + mapAll: true +# targetColumns: +# id: +# name: +# role_id: +# c_time: +# test1: \ No newline at end of file diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java index ace59b55..dea6cee7 100644 --- a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java @@ -15,14 +15,14 @@ public class ConfigLoadTest { @Before public void before() { - AdapterConfigs.put("rdb", "mytest_user.yml"); + AdapterConfigs.put("oracle", "mytest_user.yml"); // 加载数据源连接池 DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource); } @Test public void testLoad() { - Map configMap = MappingConfigLoader.load(); + Map configMap = MappingConfigLoader.load("oracle"); Assert.assertFalse(configMap.isEmpty()); } diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java index 3e7917de..0682332a 100644 --- a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java @@ -1,6 +1,7 @@ package com.alibaba.otter.canal.client.adapter.rdb.test.sync; import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; +import com.alibaba.otter.canal.client.adapter.rdb.adapters.OracleAdapter; import com.alibaba.otter.canal.client.adapter.rdb.test.TestConstant; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; @@ -13,7 +14,7 @@ public class Common { DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource); OuterAdapterConfig outerAdapterConfig = new OuterAdapterConfig(); - outerAdapterConfig.setName("rdb"); + outerAdapterConfig.setName("oracle"); Map properties = new HashMap<>(); properties.put("jdbc.driveClassName", "oracle.jdbc.OracleDriver"); properties.put("jdbc.url", "jdbc:oracle:thin:@127.0.0.1:49161:XE"); @@ -21,7 +22,7 @@ public class Common { properties.put("jdbc.password", "m121212"); outerAdapterConfig.setProperties(properties); - RdbAdapter adapter = new RdbAdapter(); + RdbAdapter adapter = new OracleAdapter(); adapter.init(outerAdapterConfig); return adapter; } diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java index 3324e41c..514ffaa5 100644 --- a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java @@ -1,8 +1,5 @@ package com.alibaba.otter.canal.client.adapter.rdb.test.sync; -import java.io.BufferedReader; -import java.io.Reader; -import java.sql.Clob; import java.util.*; import org.junit.Before; @@ -18,7 +15,7 @@ public class OracleSyncTest { @Before public void init() { - AdapterConfigs.put("rdb", "mytest_user.yml"); + AdapterConfigs.put("oracle", "mytest_user.yml"); rdbAdapter = Common.init(); } @@ -43,5 +40,27 @@ public class OracleSyncTest { rdbAdapter.sync(dml); } + @Test + public void test02() { + Dml dml = new Dml(); + dml.setDestination("example"); + dml.setTs(new Date().getTime()); + dml.setType("UPDATE"); + dml.setDatabase("mytest"); + dml.setTable("user"); + List> dataList = new ArrayList<>(); + Map data = new LinkedHashMap<>(); + dataList.add(data); + data.put("id", 1L); + data.put("name", "Eric2"); + dml.setData(dataList); + List> oldList = new ArrayList<>(); + Map old = new LinkedHashMap<>(); + oldList.add(old); + old.put("name", "Eric"); + dml.setOld(oldList); + + rdbAdapter.sync(dml); + } } diff --git a/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml b/client-adapter/rdb/src/test/resources/oracle/mytest_user.yml similarity index 51% rename from client-adapter/rdb/src/main/resources/rdb/mytest_user.yml rename to client-adapter/rdb/src/test/resources/oracle/mytest_user.yml index d7bb4ae8..a7bede03 100644 --- a/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml +++ b/client-adapter/rdb/src/test/resources/oracle/mytest_user.yml @@ -4,9 +4,12 @@ dbMapping: database: mytest table: user targetTable: mytest.tb_user - commitBatch: 3000 - mapAll: false - targetColumns: - id: - name: - role_id: \ No newline at end of file + targetPk: + id: id + mapAll: true +# targetColumns: +# id: +# name: +# role_id: +# c_time: +# test1: \ No newline at end of file