From 24d898d87a0f9b7636627842ba5d7ea0d30fc49c Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 10 Dec 2018 09:25:40 +0800 Subject: [PATCH 1/4] =?UTF-8?q?rdb=20=E5=85=A8schema=E5=90=8C=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../loader/AbstractCanalAdapterWorker.java | 4 +- .../src/main/resources/application.yml | 22 +++-- client-adapter/rdb/pom.xml | 7 +- .../canal/client/adapter/rdb/RdbAdapter.java | 41 ++++++--- .../adapter/rdb/config/MappingConfig.java | 61 +++++++------ .../adapter/rdb/service/RdbEtlService.java | 6 +- .../rdb/service/RdbMirrorDbSyncService.java | 66 ++++++++++++++ .../adapter/rdb/service/RdbSyncService.java | 12 +-- .../client/adapter/rdb/support/SyncUtil.java | 10 +++ .../src/main/resources/rdb/mytest_user.yml | 16 ++-- .../canal/client/adapter/rdb/test/DBTest.java | 2 +- .../adapter/rdb/test/SqlParserTest.java | 86 +++++++++++++++++++ 12 files changed, 258 insertions(+), 75 deletions(-) create mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java create mode 100644 client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/SqlParserTest.java diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java index db87bf01..fdf2a91e 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java @@ -70,6 +70,7 @@ public abstract class AbstractCanalAdapterWorker { }); return true; } catch (Exception e) { + logger.error(e.getMessage(), e); return false; } })); @@ -108,6 +109,7 @@ public abstract class AbstractCanalAdapterWorker { }); return true; } catch (Exception e) { + logger.error(e.getMessage(), e); return false; } })); @@ -178,7 +180,7 @@ public abstract class AbstractCanalAdapterWorker { /** * 分批同步 - * + * * @param dmls * @param adapter */ diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 098bc1f3..4a5cac3b 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -15,7 +15,7 @@ spring: canal.conf: canalServerHost: 127.0.0.1:11111 # zookeeperHosts: slave1:2181 -# mqServers: slave1:6667 #or rocketmq +# mqServers: 127.0.0.1:9092 #or rocketmq # flatMessage: true batchSize: 500 syncBatchSize: 1000 @@ -34,7 +34,7 @@ canal.conf: groups: - groupId: g1 outerAdapters: - - name: logger +# - name: logger # - name: rdb # key: oracle1 # properties: @@ -42,15 +42,13 @@ canal.conf: # jdbc.url: jdbc:oracle:thin:@localhost:49161:XE # jdbc.username: mytest # jdbc.password: m121212 -# - name: rdb -# key: postgres1 -# properties: -# jdbc.driverClassName: org.postgresql.Driver -# jdbc.url: jdbc:postgresql://localhost:5432/postgres -# jdbc.username: postgres -# jdbc.password: 121212 -# threads: 1 -# commitSize: 3000 + - name: rdb + key: mysql1 + properties: + jdbc.driverClassName: com.mysql.jdbc.Driver + jdbc.url: jdbc:mysql://192.168.100.36/mytest?useUnicode=true + jdbc.username: root + jdbc.password: Ambari-123 # - name: hbase # properties: # hbase.zookeeper.quorum: 127.0.0.1 @@ -59,4 +57,4 @@ canal.conf: # - name: es # hosts: 127.0.0.1:9300 # properties: -# cluster.name: elasticsearch \ No newline at end of file +# cluster.name: elasticsearch diff --git a/client-adapter/rdb/pom.xml b/client-adapter/rdb/pom.xml index 4ac00ffd..881b9399 100644 --- a/client-adapter/rdb/pom.xml +++ b/client-adapter/rdb/pom.xml @@ -24,6 +24,11 @@ 1.19 provided + + com.alibaba.fastsql + fastsql + 2.0.0_preview_644 + mysql @@ -100,4 +105,4 @@ - \ 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 index ae38b124..cc2fd32e 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 @@ -2,10 +2,10 @@ package com.alibaba.otter.canal.client.adapter.rdb; import java.sql.Connection; import java.sql.SQLException; -import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -21,22 +21,26 @@ import com.alibaba.otter.canal.client.adapter.rdb.config.ConfigLoader; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.rdb.monitor.RdbConfigMonitor; import com.alibaba.otter.canal.client.adapter.rdb.service.RdbEtlService; +import com.alibaba.otter.canal.client.adapter.rdb.service.RdbMirrorDbSyncService; import com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService; +import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil; import com.alibaba.otter.canal.client.adapter.support.*; @SPI("rdb") public class RdbAdapter implements OuterAdapter { - private static Logger logger = LoggerFactory.getLogger(RdbAdapter.class); + private static Logger logger = LoggerFactory.getLogger(RdbAdapter.class); - private Map rdbMapping = new HashMap<>(); // 文件名对应配置 - private Map> mappingConfigCache = new HashMap<>(); // 库名-表名对应配置 + private Map rdbMapping = new ConcurrentHashMap<>(); // 文件名对应配置 + private Map> mappingConfigCache = new ConcurrentHashMap<>(); // 库名-表名对应配置 + private Map mirrorDbConfigCache = new ConcurrentHashMap<>(); // 镜像库配置 private DruidDataSource dataSource; private RdbSyncService rdbSyncService; + private RdbMirrorDbSyncService rdbMirrorDbSyncService; - private ExecutorService executor = Executors.newFixedThreadPool(1); + private ExecutorService executor = Executors.newFixedThreadPool(1); private RdbConfigMonitor rdbConfigMonitor; @@ -62,12 +66,18 @@ public class RdbAdapter implements OuterAdapter { for (Map.Entry entry : rdbMapping.entrySet()) { String configName = entry.getKey(); MappingConfig mappingConfig = entry.getValue(); - Map configMap = mappingConfigCache - .computeIfAbsent(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." - + mappingConfig.getDbMapping().getDatabase() + "." - + mappingConfig.getDbMapping().getTable(), - k1 -> new HashMap<>()); - configMap.put(configName, mappingConfig); + if (!mappingConfig.getDbMapping().isMirrorDb()) { + Map configMap = mappingConfigCache.computeIfAbsent( + StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." + mappingConfig.getDbMapping() + .getDatabase() + "." + mappingConfig.getDbMapping().getTable(), + k1 -> new ConcurrentHashMap<>()); + configMap.put(configName, mappingConfig); + } else { + // mirrorDB + mirrorDbConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." + + mappingConfig.getDbMapping().getDatabase(), + mappingConfig); + } } Map properties = configuration.getProperties(); @@ -96,6 +106,10 @@ public class RdbAdapter implements OuterAdapter { dataSource, threads != null ? Integer.valueOf(threads) : null); + rdbMirrorDbSyncService = new RdbMirrorDbSyncService(mirrorDbConfigCache, + dataSource, + threads != null ? Integer.valueOf(threads) : null); + rdbConfigMonitor = new RdbConfigMonitor(); rdbConfigMonitor.init(configuration.getKey(), this); } @@ -103,6 +117,7 @@ public class RdbAdapter implements OuterAdapter { @Override public void sync(List dmls) { rdbSyncService.sync(dmls); + rdbMirrorDbSyncService.sync(dmls); } @Override @@ -157,7 +172,7 @@ public class RdbAdapter implements OuterAdapter { 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(); + String sql = "SELECT COUNT(1) AS cnt FROM " + SyncUtil.dbTable(dbMapping); Connection conn = null; Map res = new LinkedHashMap<>(); try { @@ -183,7 +198,7 @@ public class RdbAdapter implements OuterAdapter { } } } - res.put("targetTable", dbMapping.getTargetTable()); + res.put("targetTable", SyncUtil.dbTable(dbMapping)); return res; } 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 1d32147a..11594cde 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 @@ -1,8 +1,6 @@ package com.alibaba.otter.canal.client.adapter.rdb.config; -import java.util.LinkedHashSet; import java.util.Map; -import java.util.Set; /** * RDB表映射配置 @@ -66,30 +64,37 @@ public class MappingConfig { if (dbMapping.database == null || dbMapping.database.isEmpty()) { throw new NullPointerException("dbMapping.database"); } - if (dbMapping.table == null || dbMapping.table.isEmpty()) { + if (!dbMapping.isMirrorDb() && (dbMapping.table == null || dbMapping.table.isEmpty())) { throw new NullPointerException("dbMapping.table"); } - if (dbMapping.targetTable == null || dbMapping.targetTable.isEmpty()) { + if (!dbMapping.isMirrorDb() && (dbMapping.targetTable == null || dbMapping.targetTable.isEmpty())) { throw new NullPointerException("dbMapping.targetTable"); } } public static class DbMapping { - private String database; // 数据库名或schema名 - private String table; // 表面名 - private Map targetPk; // 目标表主键字段 - private boolean mapAll = false; // 映射所有字段 - private String targetTable; // 目标表名 - private Map targetColumns; // 目标表字段映射 + private Boolean mirrorDb = false; // 是否镜像库 + private String database; // 数据库名或schema名 + private String table; // 表名 + private Map targetPk; // 目标表主键字段 + private Boolean mapAll = false; // 映射所有字段 + private String targetDb; // 目标库名 + private String targetTable; // 目标表名 + private Map targetColumns; // 目标表字段映射 - private String etlCondition; // etl条件sql + private String etlCondition; // etl条件sql - private Set families = new LinkedHashSet<>(); // column family列表 private int readBatch = 5000; - private int commitBatch = 5000; // etl等批量提交大小 + private int commitBatch = 5000; // etl等批量提交大小 - // private volatile Map allColumns; // mapAll为true,自动设置改字段 + public boolean isMirrorDb() { + return mirrorDb == null ? false : mirrorDb; + } + + public void setMirrorDb(Boolean mirrorDb) { + this.mirrorDb = mirrorDb; + } public String getDatabase() { return database; @@ -116,13 +121,21 @@ public class MappingConfig { } public boolean isMapAll() { - return mapAll; + return mapAll == null ? false : mapAll; } - public void setMapAll(boolean mapAll) { + public void setMapAll(Boolean mapAll) { this.mapAll = mapAll; } + public String getTargetDb() { + return targetDb; + } + + public void setTargetDb(String targetDb) { + this.targetDb = targetDb; + } + public String getTargetTable() { return targetTable; } @@ -147,14 +160,6 @@ public class MappingConfig { this.etlCondition = etlCondition; } - public Set getFamilies() { - return families; - } - - public void setFamilies(Set families) { - this.families = families; - } - public int getReadBatch() { return readBatch; } @@ -170,13 +175,5 @@ 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/RdbEtlService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java index a42e5f3c..22f1bc03 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java @@ -109,7 +109,7 @@ public class RdbEtlService { logger.info( dbMapping.getTable() + " etl completed in: " + (System.currentTimeMillis() - start) / 1000 + "s!"); - etlResult.setResultMessage("导入目标表 " + dbMapping.getTargetTable() + " 数据：" + successCount.get() + " 条"); + etlResult.setResultMessage("导入目标表 " + SyncUtil.dbTable(dbMapping) + " 数据：" + successCount.get() + " 条"); } catch (Exception e) { logger.error(e.getMessage(), e); errMsg.add(hbaseTable + " etl failed! ==>" + e.getMessage()); @@ -187,7 +187,7 @@ public class RdbEtlService { // } StringBuilder insertSql = new StringBuilder(); - insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" ("); + insertSql.append("INSERT INTO ").append(SyncUtil.dbTable(dbMapping)).append(" ("); columnsMap .forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(",")); @@ -209,7 +209,7 @@ public class RdbEtlService { // 删除数据 Map values = new LinkedHashMap<>(); StringBuilder deleteSql = new StringBuilder( - "DELETE FROM " + dbMapping.getTargetTable() + " WHERE "); + "DELETE FROM " + SyncUtil.dbTable(dbMapping) + " WHERE "); appendCondition(dbMapping, deleteSql, values, rs); try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) { int k = 1; diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java new file mode 100644 index 00000000..7b3f9bdc --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java @@ -0,0 +1,66 @@ +package com.alibaba.otter.canal.client.adapter.rdb.service; + +import java.io.StringWriter; +import java.sql.Connection; +import java.sql.Statement; +import java.util.List; +import java.util.Map; + +import javax.sql.DataSource; + +import com.alibaba.fastsql.sql.ast.SQLName; +import com.alibaba.fastsql.sql.ast.SQLStatement; +import com.alibaba.fastsql.sql.ast.statement.SQLExprTableSource; +import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlStatementParser; +import com.alibaba.fastsql.sql.dialect.mysql.visitor.MySqlOutputVisitor; +import com.alibaba.fastsql.sql.dialect.mysql.visitor.MySqlSchemaStatVisitor; +import com.alibaba.fastsql.sql.parser.SQLStatementParser; +import org.apache.commons.lang.StringUtils; +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.support.Dml; + +public class RdbMirrorDbSyncService { + + private static final Logger logger = LoggerFactory.getLogger(RdbMirrorDbSyncService.class); + + private Map mirrorDbConfigCache; // 镜像库配置 + private DataSource dataSource; + + public RdbMirrorDbSyncService(Map mirrorDbConfigCache, DataSource dataSource, + Integer threads){ + this.mirrorDbConfigCache = mirrorDbConfigCache; + this.dataSource = dataSource; + } + + public void sync(List dmls) { + for (Dml dml : dmls) { + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String database = dml.getDatabase(); + MappingConfig configMap = mirrorDbConfigCache.get(destination + "." + database); + if (configMap == null) { + continue; + } + if (dml.getSql() != null) { + // DDL + executeDdl(database, dml.getSql()); + } else { + // DML + // TODO + } + } + } + + private void executeDdl(String database, String sql) { + try (Connection conn = dataSource.getConnection(); Statement statement = conn.createStatement()) { + statement.execute(sql); + if (logger.isTraceEnabled()) { + logger.trace("Execute DDL sql: {} for database: {}", sql, database); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } +} 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 ef80bb2f..38469ac4 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 @@ -91,7 +91,9 @@ public class RdbSyncService { dmlsPartition[hash].add(syncItem); }); } else { - int hash = Math.abs(Math.abs(config.getDbMapping().getTargetTable().hashCode()) % threads); + int hash = 0; + // Math.abs(Math.abs(config.getDbMapping().getTargetTable().hashCode()) % + // threads); List singleDmls = SingleDml.dml2SingleDmls(dml); singleDmls.forEach(singleDml -> { SyncItem syncItem = new SyncItem(config, singleDml); @@ -164,7 +166,7 @@ public class RdbSyncService { Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data); StringBuilder insertSql = new StringBuilder(); - insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" ("); + insertSql.append("INSERT INTO ").append(SyncUtil.dbTable(dbMapping)).append(" ("); columnsMap.forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(",")); int len = insertSql.length(); @@ -228,7 +230,7 @@ public class RdbSyncService { Map ctype = getTargetColumnType(batchExecutor.getConn(), config); StringBuilder updateSql = new StringBuilder(); - updateSql.append("UPDATE ").append(dbMapping.getTargetTable()).append(" SET "); + updateSql.append("UPDATE ").append(SyncUtil.dbTable(dbMapping)).append(" SET "); List> values = new ArrayList<>(); for (String srcColumnName : old.keySet()) { List targetColumnNames = new ArrayList<>(); @@ -280,7 +282,7 @@ public class RdbSyncService { Map ctype = getTargetColumnType(batchExecutor.getConn(), config); StringBuilder sql = new StringBuilder(); - sql.append("DELETE FROM ").append(dbMapping.getTargetTable()).append(" WHERE "); + sql.append("DELETE FROM ").append(SyncUtil.dbTable(dbMapping)).append(" WHERE "); List> values = new ArrayList<>(); // 拼接主键 @@ -313,7 +315,7 @@ public class RdbSyncService { if (columnType == null) { columnType = new LinkedHashMap<>(); final Map columnTypeTmp = columnType; - String sql = "SELECT * FROM " + dbMapping.getTargetTable() + " WHERE 1=2"; + String sql = "SELECT * FROM " + SyncUtil.dbTable(dbMapping) + " WHERE 1=2"; Util.sqlRS(conn, sql, rs -> { try { ResultSetMetaData rsd = rs.getMetaData(); diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java index 6be7ef3f..33cf5344 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java @@ -9,6 +9,7 @@ import java.util.Collection; import java.util.LinkedHashMap; import java.util.Map; +import org.apache.commons.lang.StringUtils; import org.joda.time.DateTime; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; @@ -253,4 +254,13 @@ public class SyncUtil { pstmt.setObject(i, value, type); } } + + public static String dbTable(MappingConfig.DbMapping dbMapping) { + String result = ""; + if (StringUtils.isNotEmpty(dbMapping.getTargetDb())) { + result += dbMapping.getTargetDb() + "."; + } + result += dbMapping.getTargetTable(); + return result; + } } 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 9b40dd75..68fb859e 100644 --- a/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml +++ b/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml @@ -1,17 +1,19 @@ dataSourceKey: defaultDS destination: example -outerAdapterKey: oracle1 +outerAdapterKey: mysql1 concurrent: true dbMapping: + mirrorDb: true database: mytest - table: user - targetTable: mytest.tb_user - targetPk: - id: id - mapAll: true +# table: user + targetDb: mytest2 +# targetTable: mytest.tb_user +# targetPk: +# id: id +# mapAll: true # targetColumns: # id: # name: # role_id: # c_time: -# test1: \ No newline at end of file +# test1: 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 index 5057765f..c983d7d8 100644 --- 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 @@ -42,7 +42,7 @@ public class DBTest { .prepareStatement("insert into user (id,name,role_id,c_time,test1,test2) values (?,?,?,?,?,?)"); java.util.Date now = new java.util.Date(); - for (int i = 1; i <= 100000; i++) { + for (int i = 1; i <= 10000; i++) { pstmt.clearParameters(); pstmt.setLong(1, (long) i); pstmt.setString(2, "test_" + i); diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/SqlParserTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/SqlParserTest.java new file mode 100644 index 00000000..59d0bcff --- /dev/null +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/SqlParserTest.java @@ -0,0 +1,86 @@ +wpackage com.alibaba.otter.canal.client.adapter.rdb.test; + +import com.alibaba.fastsql.sql.ast.SQLName; +import com.alibaba.fastsql.sql.ast.SQLStatement; +import com.alibaba.fastsql.sql.ast.statement.SQLCreateTableStatement; +import com.alibaba.fastsql.sql.ast.statement.SQLExprTableSource; +import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlCreateTableParser; +import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlStatementParser; +import com.alibaba.fastsql.sql.dialect.mysql.visitor.MySqlOutputVisitor; +import com.alibaba.fastsql.sql.dialect.mysql.visitor.MySqlSchemaStatVisitor; +import com.alibaba.fastsql.sql.parser.SQLStatementParser; + +import java.io.StringWriter; + +public class SqlParserTest { + + public static class TableNameVisitor extends MySqlOutputVisitor { + + public TableNameVisitor(Appendable appender){ + super(appender); + } + + @Override + public boolean visit(SQLExprTableSource x) { + SQLName table = (SQLName) x.getExpr(); + String tableName = table.getSimpleName(); + + // 改写tableName + print0("new_" + tableName.toUpperCase()); + + return true; + } + + } + + public static void main(String[] args) { + // String sql = "select * from `mytest`.`t` where id=1 and name=ming group by + // uid limit 1,200 order by ctime"; + + String sql = "CREATE TABLE `mytest`.`user` (\n" + " `id` bigint(20) NOT NULL AUTO_INCREMENT,\n" + + " `name` varchar(30) NOT NULL,\n" + " `c_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,\n" + + " `role_id` bigint(20) DEFAULT NULL,\n" + " `test1` text,\n" + " `test2` blob,\n" + + " `key` varchar(30) DEFAULT NULL,\n" + " PRIMARY KEY (`id`)\n" + + ") ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;"; + + // // 新建 MySQL Parser + // SQLStatementParser parser = new MySqlStatementParser(sql); + // + // // 使用Parser解析生成AST，这里SQLStatement就是AST + // SQLStatement sqlStatement = parser.parseStatement(); + // + // MySqlSchemaStatVisitor visitor = new MySqlSchemaStatVisitor(); + // sqlStatement.accept(visitor); + // + // System.out.println("getTables:" + visitor.getTables()); + // System.out.println("getParameters:" + visitor.getParameters()); + // System.out.println("getOrderByColumns:" + visitor.getOrderByColumns()); + // System.out.println("getGroupByColumns:" + visitor.getGroupByColumns()); + // System.out.println("---------------------------------------------------------------------------"); + // + // // 使用select访问者进行select的关键信息打印 + // // SelectPrintVisitor selectPrintVisitor = new SelectPrintVisitor(); + // // sqlStatement.accept(selectPrintVisitor); + // + // System.out.println("---------------------------------------------------------------------------"); + // // 最终sql输出 + // StringWriter out = new StringWriter(); + // TableNameVisitor outputVisitor = new TableNameVisitor(out); + // sqlStatement.accept(outputVisitor); + // System.out.println(out.toString()); + + MySqlCreateTableParser parser1 = new MySqlCreateTableParser(sql); + SQLCreateTableStatement createTableStatement = parser1.parseCreateTable(); +// MySqlSchemaStatVisitor visitor1 = new MySqlSchemaStatVisitor(); +// createTableStatement.accept(visitor1); + // visitor1.getTables().forEach((k, v) -> { + // System.out.println(k.); + // System.out.println(v); + // }); + // 最终sql输出 + StringWriter out = new StringWriter(); + TableNameVisitor outputVisitor = new TableNameVisitor(out); + createTableStatement.accept(outputVisitor); + System.out.println(out.toString()); + } +} From 5c4b1f6e6312978802c5f638c2e1ba32388e49ff Mon Sep 17 00:00:00 2001 From: mcy Date: Wed, 12 Dec 2018 10:11:06 +0800 Subject: [PATCH 2/4] modify --- .../canal/client/adapter/support/Dml.java | 9 + .../client/adapter/support/MessageUtil.java | 17 +- client-adapter/rdb/pom.xml | 5 - .../canal/client/adapter/rdb/RdbAdapter.java | 4 +- .../adapter/rdb/config/MappingConfig.java | 21 ++- .../adapter/rdb/service/RdbEtlService.java | 6 +- .../rdb/service/RdbMirrorDbSyncService.java | 42 +++-- .../adapter/rdb/service/RdbSyncService.java | 8 +- .../client/adapter/rdb/support/SyncUtil.java | 2 +- .../src/main/resources/rdb/mytest_user.yml | 1 - .../adapter/rdb/test/SqlParserTest.java | 172 +++++++++--------- 11 files changed, 158 insertions(+), 129 deletions(-) diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java index 1d5357b4..cc19813f 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java @@ -17,6 +17,7 @@ public class Dml implements Serializable { private String destination; // 对应canal的实例或者MQ的topic private String database; // 数据库或schema private String table; // 表名 + private List pkNames; private String type; // 类型: INSERT UPDATE DELETE // binlog executeTime private Long es; // 执行耗时 @@ -50,6 +51,14 @@ public class Dml implements Serializable { this.table = table; } + public List getPkNames() { + return pkNames; + } + + public void setPkNames(List pkNames) { + this.pkNames = pkNames; + } + public String getType() { return type; } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java index e40c9204..87fc2fb9 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java @@ -1,11 +1,6 @@ package com.alibaba.otter.canal.client.adapter.support; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; +import java.util.*; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.FlatMessage; @@ -55,6 +50,8 @@ public class MessageUtil { if (!rowChange.getIsDdl()) { Set updateSet = new HashSet<>(); + dml.setPkNames(new ArrayList<>()); + int i = 0; for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { if (eventType != CanalEntry.EventType.INSERT && eventType != CanalEntry.EventType.UPDATE && eventType != CanalEntry.EventType.DELETE) { @@ -71,6 +68,11 @@ public class MessageUtil { } for (CanalEntry.Column column : columns) { + if (i == 0) { + if (column.getIsKey()) { + dml.getPkNames().add(column.getName()); + } + } row.put(column.getName(), JdbcTypeUtil.typeConvert(column.getName(), column.getValue(), @@ -101,6 +103,8 @@ public class MessageUtil { old.add(rowOld); } } + + i++; } if (!data.isEmpty()) { dml.setData(data); @@ -134,6 +138,7 @@ public class MessageUtil { dml.setDestination(destination); dml.setDatabase(flatMessage.getDatabase()); dml.setTable(flatMessage.getTable()); + dml.setPkNames(flatMessage.getPkNames()); dml.setType(flatMessage.getType()); dml.setTs(flatMessage.getTs()); dml.setEs(flatMessage.getEs()); diff --git a/client-adapter/rdb/pom.xml b/client-adapter/rdb/pom.xml index 881b9399..211950e9 100644 --- a/client-adapter/rdb/pom.xml +++ b/client-adapter/rdb/pom.xml @@ -24,11 +24,6 @@ 1.19 provided - - com.alibaba.fastsql - fastsql - 2.0.0_preview_644 - mysql 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 cc2fd32e..afadb147 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 @@ -172,7 +172,7 @@ public class RdbAdapter implements OuterAdapter { public Map count(String task) { MappingConfig config = rdbMapping.get(task); MappingConfig.DbMapping dbMapping = config.getDbMapping(); - String sql = "SELECT COUNT(1) AS cnt FROM " + SyncUtil.dbTable(dbMapping); + String sql = "SELECT COUNT(1) AS cnt FROM " + SyncUtil.getDbTableName(dbMapping); Connection conn = null; Map res = new LinkedHashMap<>(); try { @@ -198,7 +198,7 @@ public class RdbAdapter implements OuterAdapter { } } } - res.put("targetTable", SyncUtil.dbTable(dbMapping)); + res.put("targetTable", SyncUtil.getDbTableName(dbMapping)); return res; } 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 11594cde..88c0fc84 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 @@ -1,5 +1,6 @@ package com.alibaba.otter.canal.client.adapter.rdb.config; +import java.util.LinkedHashMap; import java.util.Map; /** @@ -74,19 +75,19 @@ public class MappingConfig { public static class DbMapping { - private Boolean mirrorDb = false; // 是否镜像库 - private String database; // 数据库名或schema名 - private String table; // 表名 - private Map targetPk; // 目标表主键字段 - private Boolean mapAll = false; // 映射所有字段 - private String targetDb; // 目标库名 - private String targetTable; // 目标表名 - private Map targetColumns; // 目标表字段映射 + private Boolean mirrorDb = false; // 是否镜像库 + private String database; // 数据库名或schema名 + private String table; // 表名 + private Map targetPk = new LinkedHashMap<>(); // 目标表主键字段 + private Boolean mapAll = false; // 映射所有字段 + private String targetDb; // 目标库名 + private String targetTable; // 目标表名 + private Map targetColumns; // 目标表字段映射 - private String etlCondition; // etl条件sql + private String etlCondition; // etl条件sql private int readBatch = 5000; - private int commitBatch = 5000; // etl等批量提交大小 + private int commitBatch = 5000; // etl等批量提交大小 public boolean isMirrorDb() { return mirrorDb == null ? false : mirrorDb; diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java index 22f1bc03..28db80af 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java @@ -109,7 +109,7 @@ public class RdbEtlService { logger.info( dbMapping.getTable() + " etl completed in: " + (System.currentTimeMillis() - start) / 1000 + "s!"); - etlResult.setResultMessage("导入目标表 " + SyncUtil.dbTable(dbMapping) + " 数据：" + successCount.get() + " 条"); + etlResult.setResultMessage("导入目标表 " + SyncUtil.getDbTableName(dbMapping) + " 数据：" + successCount.get() + " 条"); } catch (Exception e) { logger.error(e.getMessage(), e); errMsg.add(hbaseTable + " etl failed! ==>" + e.getMessage()); @@ -187,7 +187,7 @@ public class RdbEtlService { // } StringBuilder insertSql = new StringBuilder(); - insertSql.append("INSERT INTO ").append(SyncUtil.dbTable(dbMapping)).append(" ("); + insertSql.append("INSERT INTO ").append(SyncUtil.getDbTableName(dbMapping)).append(" ("); columnsMap .forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(",")); @@ -209,7 +209,7 @@ public class RdbEtlService { // 删除数据 Map values = new LinkedHashMap<>(); StringBuilder deleteSql = new StringBuilder( - "DELETE FROM " + SyncUtil.dbTable(dbMapping) + " WHERE "); + "DELETE FROM " + SyncUtil.getDbTableName(dbMapping) + " WHERE "); appendCondition(dbMapping, deleteSql, values, rs); try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) { int k = 1; diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java index 7b3f9bdc..f7957c37 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java @@ -1,20 +1,14 @@ package com.alibaba.otter.canal.client.adapter.rdb.service; -import java.io.StringWriter; import java.sql.Connection; import java.sql.Statement; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import javax.sql.DataSource; -import com.alibaba.fastsql.sql.ast.SQLName; -import com.alibaba.fastsql.sql.ast.SQLStatement; -import com.alibaba.fastsql.sql.ast.statement.SQLExprTableSource; -import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlStatementParser; -import com.alibaba.fastsql.sql.dialect.mysql.visitor.MySqlOutputVisitor; -import com.alibaba.fastsql.sql.dialect.mysql.visitor.MySqlSchemaStatVisitor; -import com.alibaba.fastsql.sql.parser.SQLStatementParser; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -24,10 +18,13 @@ import com.alibaba.otter.canal.client.adapter.support.Dml; public class RdbMirrorDbSyncService { - private static final Logger logger = LoggerFactory.getLogger(RdbMirrorDbSyncService.class); + private static final Logger logger = LoggerFactory + .getLogger(RdbMirrorDbSyncService.class); - private Map mirrorDbConfigCache; // 镜像库配置 - private DataSource dataSource; + private Map mirrorDbConfigCache; // 镜像库配置 + private DataSource dataSource; + + private static Map tableDbConfigCache = new ConcurrentHashMap<>(); public RdbMirrorDbSyncService(Map mirrorDbConfigCache, DataSource dataSource, Integer threads){ @@ -49,6 +46,29 @@ public class RdbMirrorDbSyncService { } else { // DML // TODO + MappingConfig mappingConfig = tableDbConfigCache + .get(destination + "." + database + "." + dml.getTable()); + if (mappingConfig == null) { + // 构造一个配置 + mappingConfig = new MappingConfig(); + mappingConfig.setDataSourceKey(configMap.getDataSourceKey()); + mappingConfig.setDestination(configMap.getDestination()); + mappingConfig.setOuterAdapterKey(configMap.getOuterAdapterKey()); + mappingConfig.setConcurrent(configMap.getConcurrent()); + MappingConfig.DbMapping dbMapping = new MappingConfig.DbMapping(); + mappingConfig.setDbMapping(dbMapping); + dbMapping.setDatabase(dml.getDatabase()); + dbMapping.setTable(dml.getTable()); + dbMapping.setTargetDb(dml.getDatabase()); + dbMapping.setTargetTable(dml.getTable()); + dbMapping.setMapAll(true); + List pkNames = dml.getPkNames(); + Map pkMapping = new LinkedHashMap<>(); + pkNames.forEach(pkName -> pkMapping.put(pkName, pkName)); + dbMapping.setTargetPk(pkMapping); + + tableDbConfigCache.put(destination + "." + database + "." + dml.getTable(), mappingConfig); + } } } } 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 38469ac4..918699f4 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 @@ -166,7 +166,7 @@ public class RdbSyncService { Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data); StringBuilder insertSql = new StringBuilder(); - insertSql.append("INSERT INTO ").append(SyncUtil.dbTable(dbMapping)).append(" ("); + insertSql.append("INSERT INTO ").append(SyncUtil.getDbTableName(dbMapping)).append(" ("); columnsMap.forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(",")); int len = insertSql.length(); @@ -230,7 +230,7 @@ public class RdbSyncService { Map ctype = getTargetColumnType(batchExecutor.getConn(), config); StringBuilder updateSql = new StringBuilder(); - updateSql.append("UPDATE ").append(SyncUtil.dbTable(dbMapping)).append(" SET "); + updateSql.append("UPDATE ").append(SyncUtil.getDbTableName(dbMapping)).append(" SET "); List> values = new ArrayList<>(); for (String srcColumnName : old.keySet()) { List targetColumnNames = new ArrayList<>(); @@ -282,7 +282,7 @@ public class RdbSyncService { Map ctype = getTargetColumnType(batchExecutor.getConn(), config); StringBuilder sql = new StringBuilder(); - sql.append("DELETE FROM ").append(SyncUtil.dbTable(dbMapping)).append(" WHERE "); + sql.append("DELETE FROM ").append(SyncUtil.getDbTableName(dbMapping)).append(" WHERE "); List> values = new ArrayList<>(); // 拼接主键 @@ -315,7 +315,7 @@ public class RdbSyncService { if (columnType == null) { columnType = new LinkedHashMap<>(); final Map columnTypeTmp = columnType; - String sql = "SELECT * FROM " + SyncUtil.dbTable(dbMapping) + " WHERE 1=2"; + String sql = "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping) + " WHERE 1=2"; Util.sqlRS(conn, sql, rs -> { try { ResultSetMetaData rsd = rs.getMetaData(); diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java index 33cf5344..e8362190 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java @@ -255,7 +255,7 @@ public class SyncUtil { } } - public static String dbTable(MappingConfig.DbMapping dbMapping) { + public static String getDbTableName(MappingConfig.DbMapping dbMapping) { String result = ""; if (StringUtils.isNotEmpty(dbMapping.getTargetDb())) { result += dbMapping.getTargetDb() + "."; 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 68fb859e..a8e97f6d 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,6 @@ dbMapping: mirrorDb: true database: mytest # table: user - targetDb: mytest2 # targetTable: mytest.tb_user # targetPk: # id: id diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/SqlParserTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/SqlParserTest.java index 59d0bcff..698f5059 100644 --- a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/SqlParserTest.java +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/SqlParserTest.java @@ -1,86 +1,86 @@ -wpackage com.alibaba.otter.canal.client.adapter.rdb.test; - -import com.alibaba.fastsql.sql.ast.SQLName; -import com.alibaba.fastsql.sql.ast.SQLStatement; -import com.alibaba.fastsql.sql.ast.statement.SQLCreateTableStatement; -import com.alibaba.fastsql.sql.ast.statement.SQLExprTableSource; -import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlCreateTableParser; -import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlStatementParser; -import com.alibaba.fastsql.sql.dialect.mysql.visitor.MySqlOutputVisitor; -import com.alibaba.fastsql.sql.dialect.mysql.visitor.MySqlSchemaStatVisitor; -import com.alibaba.fastsql.sql.parser.SQLStatementParser; - -import java.io.StringWriter; - -public class SqlParserTest { - - public static class TableNameVisitor extends MySqlOutputVisitor { - - public TableNameVisitor(Appendable appender){ - super(appender); - } - - @Override - public boolean visit(SQLExprTableSource x) { - SQLName table = (SQLName) x.getExpr(); - String tableName = table.getSimpleName(); - - // 改写tableName - print0("new_" + tableName.toUpperCase()); - - return true; - } - - } - - public static void main(String[] args) { - // String sql = "select * from `mytest`.`t` where id=1 and name=ming group by - // uid limit 1,200 order by ctime"; - - String sql = "CREATE TABLE `mytest`.`user` (\n" + " `id` bigint(20) NOT NULL AUTO_INCREMENT,\n" - + " `name` varchar(30) NOT NULL,\n" + " `c_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,\n" - + " `role_id` bigint(20) DEFAULT NULL,\n" + " `test1` text,\n" + " `test2` blob,\n" - + " `key` varchar(30) DEFAULT NULL,\n" + " PRIMARY KEY (`id`)\n" - + ") ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;"; - - // // 新建 MySQL Parser - // SQLStatementParser parser = new MySqlStatementParser(sql); - // - // // 使用Parser解析生成AST，这里SQLStatement就是AST - // SQLStatement sqlStatement = parser.parseStatement(); - // - // MySqlSchemaStatVisitor visitor = new MySqlSchemaStatVisitor(); - // sqlStatement.accept(visitor); - // - // System.out.println("getTables:" + visitor.getTables()); - // System.out.println("getParameters:" + visitor.getParameters()); - // System.out.println("getOrderByColumns:" + visitor.getOrderByColumns()); - // System.out.println("getGroupByColumns:" + visitor.getGroupByColumns()); - // System.out.println("---------------------------------------------------------------------------"); - // - // // 使用select访问者进行select的关键信息打印 - // // SelectPrintVisitor selectPrintVisitor = new SelectPrintVisitor(); - // // sqlStatement.accept(selectPrintVisitor); - // - // System.out.println("---------------------------------------------------------------------------"); - // // 最终sql输出 - // StringWriter out = new StringWriter(); - // TableNameVisitor outputVisitor = new TableNameVisitor(out); - // sqlStatement.accept(outputVisitor); - // System.out.println(out.toString()); - - MySqlCreateTableParser parser1 = new MySqlCreateTableParser(sql); - SQLCreateTableStatement createTableStatement = parser1.parseCreateTable(); -// MySqlSchemaStatVisitor visitor1 = new MySqlSchemaStatVisitor(); -// createTableStatement.accept(visitor1); - // visitor1.getTables().forEach((k, v) -> { - // System.out.println(k.); - // System.out.println(v); - // }); - // 最终sql输出 - StringWriter out = new StringWriter(); - TableNameVisitor outputVisitor = new TableNameVisitor(out); - createTableStatement.accept(outputVisitor); - System.out.println(out.toString()); - } -} +//wpackage com.alibaba.otter.canal.client.adapter.rdb.test; +// +//import com.alibaba.fastsql.sql.ast.SQLName; +//import com.alibaba.fastsql.sql.ast.SQLStatement; +//import com.alibaba.fastsql.sql.ast.statement.SQLCreateTableStatement; +//import com.alibaba.fastsql.sql.ast.statement.SQLExprTableSource; +//import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlCreateTableParser; +//import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlStatementParser; +//import com.alibaba.fastsql.sql.dialect.mysql.visitor.MySqlOutputVisitor; +//import com.alibaba.fastsql.sql.dialect.mysql.visitor.MySqlSchemaStatVisitor; +//import com.alibaba.fastsql.sql.parser.SQLStatementParser; +// +//import java.io.StringWriter; +// +//public class SqlParserTest { +// +// public static class TableNameVisitor extends MySqlOutputVisitor { +// +// public TableNameVisitor(Appendable appender){ +// super(appender); +// } +// +// @Override +// public boolean visit(SQLExprTableSource x) { +// SQLName table = (SQLName) x.getExpr(); +// String tableName = table.getSimpleName(); +// +// // 改写tableName +// print0("new_" + tableName.toUpperCase()); +// +// return true; +// } +// +// } +// +// public static void main(String[] args) { +// // String sql = "select * from `mytest`.`t` where id=1 and name=ming group by +// // uid limit 1,200 order by ctime"; +// +// String sql = "CREATE TABLE `mytest`.`user` (\n" + " `id` bigint(20) NOT NULL AUTO_INCREMENT,\n" +// + " `name` varchar(30) NOT NULL,\n" + " `c_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,\n" +// + " `role_id` bigint(20) DEFAULT NULL,\n" + " `test1` text,\n" + " `test2` blob,\n" +// + " `key` varchar(30) DEFAULT NULL,\n" + " PRIMARY KEY (`id`)\n" +// + ") ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;"; +// +// // // 新建 MySQL Parser +// // SQLStatementParser parser = new MySqlStatementParser(sql); +// // +// // // 使用Parser解析生成AST，这里SQLStatement就是AST +// // SQLStatement sqlStatement = parser.parseStatement(); +// // +// // MySqlSchemaStatVisitor visitor = new MySqlSchemaStatVisitor(); +// // sqlStatement.accept(visitor); +// // +// // System.out.println("getTables:" + visitor.getTables()); +// // System.out.println("getParameters:" + visitor.getParameters()); +// // System.out.println("getOrderByColumns:" + visitor.getOrderByColumns()); +// // System.out.println("getGroupByColumns:" + visitor.getGroupByColumns()); +// // System.out.println("---------------------------------------------------------------------------"); +// // +// // // 使用select访问者进行select的关键信息打印 +// // // SelectPrintVisitor selectPrintVisitor = new SelectPrintVisitor(); +// // // sqlStatement.accept(selectPrintVisitor); +// // +// // System.out.println("---------------------------------------------------------------------------"); +// // // 最终sql输出 +// // StringWriter out = new StringWriter(); +// // TableNameVisitor outputVisitor = new TableNameVisitor(out); +// // sqlStatement.accept(outputVisitor); +// // System.out.println(out.toString()); +// +// MySqlCreateTableParser parser1 = new MySqlCreateTableParser(sql); +// SQLCreateTableStatement createTableStatement = parser1.parseCreateTable(); +//// MySqlSchemaStatVisitor visitor1 = new MySqlSchemaStatVisitor(); +//// createTableStatement.accept(visitor1); +// // visitor1.getTables().forEach((k, v) -> { +// // System.out.println(k.); +// // System.out.println(v); +// // }); +// // 最终sql输出 +// StringWriter out = new StringWriter(); +// TableNameVisitor outputVisitor = new TableNameVisitor(out); +// createTableStatement.accept(outputVisitor); +// System.out.println(out.toString()); +// } +//} From fbf249580f595e973f1548193de66c4f43af304d Mon Sep 17 00:00:00 2001 From: mcy Date: Fri, 14 Dec 2018 11:30:41 +0800 Subject: [PATCH 3/4] mirror db --- .../loader/AbstractCanalAdapterWorker.java | 4 +- .../canal/client/adapter/rdb/RdbAdapter.java | 74 +++++-- .../adapter/rdb/config/MappingConfig.java | 10 + .../rdb/service/RdbMirrorDbSyncService.java | 155 ++++++++++----- .../adapter/rdb/service/RdbSyncService.java | 185 ++++++++++++------ .../client/adapter/rdb/support/SyncUtil.java | 4 + 6 files changed, 309 insertions(+), 123 deletions(-) diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java index fdf2a91e..363d1410 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java @@ -200,7 +200,9 @@ public abstract class AbstractCanalAdapterWorker { len = 0; } } - adapter.sync(dmlsBatch); + if (!dmlsBatch.isEmpty()) { + adapter.sync(dmlsBatch); + } } } 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 afadb147..dc1ebd0e 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 @@ -5,9 +5,7 @@ import java.sql.SQLException; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.*; import javax.sql.DataSource; @@ -26,6 +24,12 @@ import com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService; import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil; import com.alibaba.otter.canal.client.adapter.support.*; +/** + * RDB适配器实现类 + * + * @author rewerma 2018-11-7 下午06:45:49 + * @version 1.0.0 + */ @SPI("rdb") public class RdbAdapter implements OuterAdapter { @@ -40,8 +44,6 @@ public class RdbAdapter implements OuterAdapter { private RdbSyncService rdbSyncService; private RdbMirrorDbSyncService rdbMirrorDbSyncService; - private ExecutorService executor = Executors.newFixedThreadPool(1); - private RdbConfigMonitor rdbConfigMonitor; public Map getRdbMapping() { @@ -52,6 +54,11 @@ public class RdbAdapter implements OuterAdapter { return mappingConfigCache; } + /** + * 初始化方法 + * + * @param configuration 外部适配器配置信息 + */ @Override public void init(OuterAdapterConfig configuration) { Map rdbMappingTmp = ConfigLoader.load(); @@ -80,6 +87,7 @@ public class RdbAdapter implements OuterAdapter { } } + // 初始化连接池 Map properties = configuration.getProperties(); dataSource = new DruidDataSource(); dataSource.setDriverClassName(properties.get("jdbc.driverClassName")); @@ -88,7 +96,7 @@ public class RdbAdapter implements OuterAdapter { dataSource.setPassword(properties.get("jdbc.password")); dataSource.setInitialSize(1); dataSource.setMinIdle(1); - dataSource.setMaxActive(20); + dataSource.setMaxActive(10); dataSource.setMaxWait(60000); dataSource.setTimeBetweenEvictionRunsMillis(60000); dataSource.setMinEvictableIdleTimeMillis(300000); @@ -102,24 +110,49 @@ public class RdbAdapter implements OuterAdapter { String threads = properties.get("threads"); // String commitSize = properties.get("commitSize"); - rdbSyncService = new RdbSyncService(mappingConfigCache, - dataSource, - threads != null ? Integer.valueOf(threads) : null); + rdbSyncService = new RdbSyncService(dataSource, threads != null ? Integer.valueOf(threads) : null); rdbMirrorDbSyncService = new RdbMirrorDbSyncService(mirrorDbConfigCache, dataSource, - threads != null ? Integer.valueOf(threads) : null); + threads != null ? Integer.valueOf(threads) : null, + rdbSyncService.getColumnsTypeCache()); rdbConfigMonitor = new RdbConfigMonitor(); rdbConfigMonitor.init(configuration.getKey(), this); } + /** + * 同步方法 + * + * @param dmls 数据包 + */ @Override public void sync(List dmls) { - rdbSyncService.sync(dmls); - rdbMirrorDbSyncService.sync(dmls); + ExecutorService executorService = Executors.newFixedThreadPool(2); + + Future future1 = executorService.submit(() -> { + rdbSyncService.sync(mappingConfigCache, dmls); + return true; + }); + Future future2 = executorService.submit(() -> { + rdbMirrorDbSyncService.sync(dmls); + return true; + }); + try { + future1.get(); + future2.get(); + } catch (ExecutionException | InterruptedException e) { + // ignore + } } + /** + * ETL方法 + * + * @param task 任务名, 对应配置名 + * @param params etl筛选条件 + * @return ETL结果 + */ @Override public EtlResult etl(String task, List params) { EtlResult etlResult = new EtlResult(); @@ -168,6 +201,12 @@ public class RdbAdapter implements OuterAdapter { return etlResult; } + /** + * 获取总数方法 + * + * @param task 任务名, 对应配置名 + * @return 总数 + */ @Override public Map count(String task) { MappingConfig config = rdbMapping.get(task); @@ -203,6 +242,12 @@ public class RdbAdapter implements OuterAdapter { return res; } + /** + * 获取对应canal instance name 或 mq topic + * + * @param task 任务名, 对应配置名 + * @return destination + */ @Override public String getDestination(String task) { MappingConfig config = rdbMapping.get(task); @@ -212,6 +257,9 @@ public class RdbAdapter implements OuterAdapter { return null; } + /** + * 销毁方法 + */ @Override public void destroy() { if (rdbConfigMonitor != null) { @@ -222,8 +270,6 @@ public class RdbAdapter implements OuterAdapter { rdbSyncService.close(); } - executor.shutdown(); - 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 index 88c0fc84..aa6a7d1b 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 @@ -89,6 +89,8 @@ public class MappingConfig { private int readBatch = 5000; private int commitBatch = 5000; // etl等批量提交大小 + private Map allMapColumns; + public boolean isMirrorDb() { return mirrorDb == null ? false : mirrorDb; } @@ -176,5 +178,13 @@ public class MappingConfig { public void setCommitBatch(int commitBatch) { this.commitBatch = commitBatch; } + + public Map getAllMapColumns() { + return allMapColumns; + } + + public void setAllMapColumns(Map allMapColumns) { + this.allMapColumns = allMapColumns; + } } } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java index f7957c37..99ee875f 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java @@ -2,6 +2,7 @@ package com.alibaba.otter.canal.client.adapter.rdb.service; import java.sql.Connection; import java.sql.Statement; +import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -14,70 +15,136 @@ 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.support.SingleDml; import com.alibaba.otter.canal.client.adapter.support.Dml; +/** + * RDB镜像库同步操作业务 + * + * @author rewerma 2018-12-12 下午011:23 + * @version 1.0.0 + */ public class RdbMirrorDbSyncService { - private static final Logger logger = LoggerFactory - .getLogger(RdbMirrorDbSyncService.class); + private static final Logger logger = LoggerFactory.getLogger(RdbMirrorDbSyncService.class); - private Map mirrorDbConfigCache; // 镜像库配置 - private DataSource dataSource; + private Map mirrorDbConfigCache; // 镜像库配置 + private DataSource dataSource; + private RdbSyncService rdbSyncService; // rdbSyncService代理 - private static Map tableDbConfigCache = new ConcurrentHashMap<>(); + private final Map tableDbConfigCache = new ConcurrentHashMap<>(); // 自动生成的库表配置缓存 public RdbMirrorDbSyncService(Map mirrorDbConfigCache, DataSource dataSource, - Integer threads){ + Integer threads, Map> columnsTypeCache){ this.mirrorDbConfigCache = mirrorDbConfigCache; this.dataSource = dataSource; + this.rdbSyncService = new RdbSyncService(dataSource, threads, columnsTypeCache); } + /** + * 批量同步方法 + * + * @param dmls 批量 DML + */ public void sync(List dmls) { - for (Dml dml : dmls) { - String destination = StringUtils.trimToEmpty(dml.getDestination()); - String database = dml.getDatabase(); - MappingConfig configMap = mirrorDbConfigCache.get(destination + "." + database); - if (configMap == null) { - continue; - } - if (dml.getSql() != null) { - // DDL - executeDdl(database, dml.getSql()); - } else { - // DML - // TODO - MappingConfig mappingConfig = tableDbConfigCache - .get(destination + "." + database + "." + dml.getTable()); - if (mappingConfig == null) { - // 构造一个配置 - mappingConfig = new MappingConfig(); - mappingConfig.setDataSourceKey(configMap.getDataSourceKey()); - mappingConfig.setDestination(configMap.getDestination()); - mappingConfig.setOuterAdapterKey(configMap.getOuterAdapterKey()); - mappingConfig.setConcurrent(configMap.getConcurrent()); - MappingConfig.DbMapping dbMapping = new MappingConfig.DbMapping(); - mappingConfig.setDbMapping(dbMapping); - dbMapping.setDatabase(dml.getDatabase()); - dbMapping.setTable(dml.getTable()); - dbMapping.setTargetDb(dml.getDatabase()); - dbMapping.setTargetTable(dml.getTable()); - dbMapping.setMapAll(true); - List pkNames = dml.getPkNames(); - Map pkMapping = new LinkedHashMap<>(); - pkNames.forEach(pkName -> pkMapping.put(pkName, pkName)); - dbMapping.setTargetPk(pkMapping); - - tableDbConfigCache.put(destination + "." + database + "." + dml.getTable(), mappingConfig); + try { + List dmlList = new ArrayList<>(); + for (Dml dml : dmls) { + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String database = dml.getDatabase(); + MappingConfig configMap = mirrorDbConfigCache.get(destination + "." + database); + if (configMap == null) { + continue; + } + if (dml.getSql() != null) { + // DDL + executeDdl(dml); + rdbSyncService.getColumnsTypeCache().remove(destination + "." + database + "." + dml.getTable()); + tableDbConfigCache.remove(destination + "." + database + "." + dml.getTable()); // 删除对应库表配置 + } else { + // DML + initMappingConfig(destination + "." + database + "." + dml.getTable(), configMap, dml); + dmlList.add(dml); } } + if (!dmlList.isEmpty()) { + rdbSyncService.sync(dmlList, dml -> { + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String database = dml.getDatabase(); + String table = dml.getTable(); + MappingConfig config = tableDbConfigCache.get(destination + "." + database + "." + table); + + if (config == null) { + return false; + } + + if (config.getConcurrent()) { + List singleDmls = SingleDml.dml2SingleDmls(dml); + singleDmls.forEach(singleDml -> { + int hash = rdbSyncService.pkHash(config.getDbMapping(), singleDml.getData()); + RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); + rdbSyncService.getDmlsPartition()[hash].add(syncItem); + }); + } else { + int hash = 0; + List singleDmls = SingleDml.dml2SingleDmls(dml); + singleDmls.forEach(singleDml -> { + RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); + rdbSyncService.getDmlsPartition()[hash].add(syncItem); + }); + } + return true; + }); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); } } - private void executeDdl(String database, String sql) { + /** + * 初始化表配置 + * + * @param key 配置key: destination.database.table + * @param baseConfigMap db sync config + * @param dml DML + */ + private void initMappingConfig(String key, MappingConfig baseConfigMap, Dml dml) { + MappingConfig mappingConfig = tableDbConfigCache.get(key); + if (mappingConfig == null) { + // 构造一个配置 + mappingConfig = new MappingConfig(); + mappingConfig.setDataSourceKey(baseConfigMap.getDataSourceKey()); + mappingConfig.setDestination(baseConfigMap.getDestination()); + mappingConfig.setOuterAdapterKey(baseConfigMap.getOuterAdapterKey()); + mappingConfig.setConcurrent(baseConfigMap.getConcurrent()); + MappingConfig.DbMapping dbMapping = new MappingConfig.DbMapping(); + mappingConfig.setDbMapping(dbMapping); + dbMapping.setDatabase(dml.getDatabase()); + dbMapping.setTable(dml.getTable()); + dbMapping.setTargetDb(dml.getDatabase()); + dbMapping.setTargetTable(dml.getTable()); + dbMapping.setMapAll(true); + List pkNames = dml.getPkNames(); + Map pkMapping = new LinkedHashMap<>(); + pkNames.forEach(pkName -> pkMapping.put(pkName, pkName)); + dbMapping.setTargetPk(pkMapping); + + tableDbConfigCache.put(key, mappingConfig); + } + } + + /** + * DDL 操作 + * + * @param ddl DDL + */ + private void executeDdl(Dml ddl) { try (Connection conn = dataSource.getConnection(); Statement statement = conn.createStatement()) { - statement.execute(sql); + statement.execute(ddl.getSql()); + // 移除对应配置 + tableDbConfigCache.remove(ddl.getDatabase() + "." + ddl.getDatabase() + "." + ddl.getTable()); if (logger.isTraceEnabled()) { - logger.trace("Execute DDL sql: {} for database: {}", sql, database); + logger.trace("Execute DDL sql: {} for database: {}", ddl.getSql(), ddl.getDatabase()); } } catch (Exception e) { logger.error(e.getMessage(), e); 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 918699f4..bd4ef77b 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 @@ -11,6 +11,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.function.Function; import javax.sql.DataSource; @@ -36,26 +37,37 @@ import com.alibaba.otter.canal.client.adapter.support.Util; */ public class RdbSyncService { - private static final Logger logger = LoggerFactory.getLogger(RdbSyncService.class); + private static final Logger logger = LoggerFactory.getLogger(RdbSyncService.class); - private final Map> COLUMNS_TYPE_CACHE = new ConcurrentHashMap<>(); + // 源库表字段类型缓存: instance.schema.table -> + private Map> columnsTypeCache; - private Map> mappingConfigCache; // 库名-表名对应配置 + private int threads = 3; - private int threads = 3; + private List[] dmlsPartition; + private BatchExecutor[] batchExecutors; + private ExecutorService[] executorThreads; - private List[] dmlsPartition; - private BatchExecutor[] batchExecutors; - private ExecutorService[] executorThreads; + public List[] getDmlsPartition() { + return dmlsPartition; + } + + public Map> getColumnsTypeCache() { + return columnsTypeCache; + } @SuppressWarnings("unchecked") - public RdbSyncService(Map> mappingConfigCache, DataSource dataSource, - Integer threads){ + public RdbSyncService(DataSource dataSource, Integer threads){ + this(dataSource, threads, new ConcurrentHashMap<>()); + } + + @SuppressWarnings("unchecked") + public RdbSyncService(DataSource dataSource, Integer threads, Map> columnsTypeCache){ + this.columnsTypeCache = columnsTypeCache; try { if (threads != null) { this.threads = threads; } - this.mappingConfigCache = mappingConfigCache; this.dmlsPartition = new List[this.threads]; this.batchExecutors = new BatchExecutor[this.threads]; this.executorThreads = new ExecutorService[this.threads]; @@ -69,66 +81,111 @@ public class RdbSyncService { } } - public void sync(List dmls) { + /** + * 批量同步回调 + * + * @param dmls 批量 DML + * @param function 回调方法 + */ + public void sync(List dmls, Function function) { try { + boolean toExecute = false; for (Dml dml : dmls) { - String destination = StringUtils.trimToEmpty(dml.getDestination()); - String database = dml.getDatabase(); - String table = dml.getTable(); - Map configMap = mappingConfigCache - .get(destination + "." + database + "." + table); - - if (configMap == null) { - continue; + if (!toExecute) { + toExecute = function.apply(dml); + } else { + function.apply(dml); + } + } + if (toExecute) { + List futures = new ArrayList<>(); + for (int i = 0; i < threads; i++) { + int j = i; + futures.add(executorThreads[i].submit(() -> { + dmlsPartition[j] + .forEach(syncItem -> sync(batchExecutors[j], syncItem.config, syncItem.singleDml)); + batchExecutors[j].commit(); + return true; + })); } - for (MappingConfig config : configMap.values()) { - if (config.getConcurrent()) { - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - int hash = pkHash(config.getDbMapping(), singleDml.getData()); - SyncItem syncItem = new SyncItem(config, singleDml); - dmlsPartition[hash].add(syncItem); - }); - } else { - int hash = 0; - // Math.abs(Math.abs(config.getDbMapping().getTargetTable().hashCode()) % - // threads); - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - SyncItem syncItem = new SyncItem(config, singleDml); - dmlsPartition[hash].add(syncItem); - }); + futures.forEach(future -> { + try { + future.get(); + } catch (Exception e) { + logger.error(e.getMessage(), e); } - } - } - List futures = new ArrayList<>(); - for (int i = 0; i < threads; i++) { - int j = i; - futures.add(executorThreads[i].submit(() -> { - dmlsPartition[j].forEach(syncItem -> sync(batchExecutors[j], syncItem.config, syncItem.singleDml)); - batchExecutors[j].commit(); - return true; - })); - } + }); - futures.forEach(future -> { - try { - future.get(); - } catch (Exception e) { - logger.error(e.getMessage(), e); + for (int i = 0; i < threads; i++) { + dmlsPartition[i].clear(); } - }); - - for (int i = 0; i < threads; i++) { - dmlsPartition[i].clear(); } } catch (Exception e) { logger.error(e.getMessage(), e); } } - private void sync(BatchExecutor batchExecutor, MappingConfig config, SingleDml dml) { + /** + * 批量同步 + * + * @param mappingConfig 配置集合 + * @param dmls 批量 DML + */ + public void sync(Map> mappingConfig, List dmls) { + try { + sync(dmls, dml -> { + if (dml.getSql() != null) { + // DDL + columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable()); + return false; + } else { + // DML + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String database = dml.getDatabase(); + String table = dml.getTable(); + Map configMap = mappingConfig + .get(destination + "." + database + "." + table); + + if (configMap == null) { + return false; + } + + boolean executed = false; + for (MappingConfig config : configMap.values()) { + if (config.getConcurrent()) { + List singleDmls = SingleDml.dml2SingleDmls(dml); + singleDmls.forEach(singleDml -> { + int hash = pkHash(config.getDbMapping(), singleDml.getData()); + SyncItem syncItem = new SyncItem(config, singleDml); + dmlsPartition[hash].add(syncItem); + }); + } else { + int hash = 0; + List singleDmls = SingleDml.dml2SingleDmls(dml); + singleDmls.forEach(singleDml -> { + SyncItem syncItem = new SyncItem(config, singleDml); + dmlsPartition[hash].add(syncItem); + }); + } + executed = true; + } + return executed; + } + }); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + /** + * 单条 dml 同步 + * + * @param batchExecutor 批量事务执行器 + * @param config 对应配置对象 + * @param dml DML + */ + public void sync(BatchExecutor batchExecutor, MappingConfig config, SingleDml dml) { try { if (config != null) { String type = dml.getType(); @@ -308,10 +365,10 @@ public class RdbSyncService { 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); + Map columnType = columnsTypeCache.get(cacheKey); if (columnType == null) { synchronized (RdbSyncService.class) { - columnType = COLUMNS_TYPE_CACHE.get(cacheKey); + columnType = columnsTypeCache.get(cacheKey); if (columnType == null) { columnType = new LinkedHashMap<>(); final Map columnTypeTmp = columnType; @@ -323,7 +380,7 @@ public class RdbSyncService { for (int i = 1; i <= columnCount; i++) { columnTypeTmp.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i)); } - COLUMNS_TYPE_CACHE.put(cacheKey, columnTypeTmp); + columnsTypeCache.put(cacheKey, columnTypeTmp); } catch (SQLException e) { logger.error(e.getMessage(), e); } @@ -364,12 +421,12 @@ public class RdbSyncService { sql.delete(len - 4, len); } - private class SyncItem { + public static class SyncItem { private MappingConfig config; private SingleDml singleDml; - private SyncItem(MappingConfig config, SingleDml singleDml){ + public SyncItem(MappingConfig config, SingleDml singleDml){ this.config = config; this.singleDml = singleDml; } @@ -378,11 +435,11 @@ public class RdbSyncService { /** * 取主键hash */ - private int pkHash(DbMapping dbMapping, Map d) { + public int pkHash(DbMapping dbMapping, Map d) { return pkHash(dbMapping, d, null); } - private int pkHash(DbMapping dbMapping, Map d, Map o) { + public int pkHash(DbMapping dbMapping, Map d, Map o) { int hash = 0; // 取主键 for (Map.Entry entry : dbMapping.getTargetPk().entrySet()) { diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java index e8362190..bf21a300 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java @@ -23,6 +23,9 @@ public class SyncUtil { public static Map getColumnsMap(MappingConfig.DbMapping dbMapping, Collection columns) { Map columnsMap; if (dbMapping.isMapAll()) { + if (dbMapping.getAllMapColumns() != null) { + return dbMapping.getAllMapColumns(); + } columnsMap = new LinkedHashMap<>(); for (String srcColumn : columns) { boolean flag = true; @@ -39,6 +42,7 @@ public class SyncUtil { columnsMap.put(srcColumn, srcColumn); } } + dbMapping.setAllMapColumns(columnsMap); } else { columnsMap = dbMapping.getTargetColumns(); } From 3acdffb29708fe4e9a393c7032ad0db840cdafa3 Mon Sep 17 00:00:00 2001 From: mcy Date: Fri, 14 Dec 2018 14:58:15 +0800 Subject: [PATCH 4/4] =?UTF-8?q?mirror=20schema=20=E5=AE=8C=E6=88=90?= =?UTF-8?q?=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../adapter/es/config/ESSyncConfigLoader.java | 6 ++- .../hbase/config/MappingConfigLoader.java | 8 +++- .../src/main/resources/application.yml | 25 ++++++---- .../canal/client/adapter/rdb/RdbAdapter.java | 12 +++-- .../adapter/rdb/config/ConfigLoader.java | 8 +++- .../adapter/rdb/config/MappingConfig.java | 12 ++--- .../adapter/rdb/config/MirrorDbConfig.java | 44 +++++++++++++++++ .../adapter/rdb/monitor/RdbConfigMonitor.java | 28 ++++++++--- .../rdb/service/RdbMirrorDbSyncService.java | 47 ++++++++++--------- .../adapter/rdb/service/RdbSyncService.java | 2 +- .../client/adapter/rdb/support/SyncUtil.java | 2 +- .../src/main/resources/rdb/mytest_user.yml | 23 ++++++--- 12 files changed, 158 insertions(+), 59 deletions(-) create mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MirrorDbConfig.java 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 e83a96e6..fcf94105 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 @@ -3,6 +3,7 @@ package com.alibaba.otter.canal.client.adapter.es.config; import java.util.LinkedHashMap; import java.util.Map; +import com.alibaba.fastjson.JSONObject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.yaml.snakeyaml.Yaml; @@ -19,6 +20,7 @@ public class ESSyncConfigLoader { private static Logger logger = LoggerFactory.getLogger(ESSyncConfigLoader.class); + @SuppressWarnings("unchecked") public static synchronized Map load() { logger.info("## Start loading es mapping config ... "); @@ -26,7 +28,9 @@ public class ESSyncConfigLoader { Map configContentMap = MappingConfigsLoader.loadConfigs("es"); configContentMap.forEach((fileName, content) -> { - ESSyncConfig config = new Yaml().loadAs(content, ESSyncConfig.class); + Map configMap = new Yaml().loadAs(content, Map.class); // yml自带的对象反射不是很稳定 + JSONObject configJson = new JSONObject(configMap); + ESSyncConfig config = configJson.toJavaObject(ESSyncConfig.class); try { config.validate(); } catch (Exception e) { 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 1f8c05c1..3cf16a71 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 @@ -3,6 +3,7 @@ package com.alibaba.otter.canal.client.adapter.hbase.config; import java.util.LinkedHashMap; import java.util.Map; +import com.alibaba.fastjson.JSONObject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.yaml.snakeyaml.Yaml; @@ -21,9 +22,10 @@ public class MappingConfigLoader { /** * 加载HBase表映射配置 - * + * * @return 配置名/配置文件名--对象 */ + @SuppressWarnings("unchecked") public static Map load() { logger.info("## Start loading hbase mapping config ... "); @@ -31,7 +33,9 @@ public class MappingConfigLoader { Map configContentMap = MappingConfigsLoader.loadConfigs("hbase"); configContentMap.forEach((fileName, content) -> { - MappingConfig config = new Yaml().loadAs(content, MappingConfig.class); + Map configMap = new Yaml().loadAs(content, Map.class); // yml自带的对象反射不是很稳定 + JSONObject configJson = new JSONObject(configMap); + MappingConfig config = configJson.toJavaObject(MappingConfig.class); try { config.validate(); } catch (Exception e) { diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 4a5cac3b..6478db45 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -34,7 +34,14 @@ canal.conf: groups: - groupId: g1 outerAdapters: -# - name: logger + - name: logger +# - name: rdb +# key: mysql1 +# properties: +# jdbc.driverClassName: com.mysql.jdbc.Driver +# jdbc.url: jdbc:mysql://192.168.0.36/mytest?useUnicode=true +# jdbc.username: root +# jdbc.password: 121212 # - name: rdb # key: oracle1 # properties: @@ -42,13 +49,15 @@ canal.conf: # jdbc.url: jdbc:oracle:thin:@localhost:49161:XE # jdbc.username: mytest # jdbc.password: m121212 - - name: rdb - key: mysql1 - properties: - jdbc.driverClassName: com.mysql.jdbc.Driver - jdbc.url: jdbc:mysql://192.168.100.36/mytest?useUnicode=true - jdbc.username: root - jdbc.password: Ambari-123 +# - name: rdb +# key: postgres1 +# properties: +# jdbc.driverClassName: org.postgresql.Driver +# jdbc.url: jdbc:postgresql://localhost:5432/postgres +# jdbc.username: postgres +# jdbc.password: 121212 +# threads: 1 +# commitSize: 3000 # - name: hbase # properties: # hbase.zookeeper.quorum: 127.0.0.1 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 dc1ebd0e..3c4f8387 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 @@ -17,6 +17,7 @@ import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.rdb.config.ConfigLoader; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; +import com.alibaba.otter.canal.client.adapter.rdb.config.MirrorDbConfig; import com.alibaba.otter.canal.client.adapter.rdb.monitor.RdbConfigMonitor; import com.alibaba.otter.canal.client.adapter.rdb.service.RdbEtlService; import com.alibaba.otter.canal.client.adapter.rdb.service.RdbMirrorDbSyncService; @@ -37,7 +38,7 @@ public class RdbAdapter implements OuterAdapter { private Map rdbMapping = new ConcurrentHashMap<>(); // 文件名对应配置 private Map> mappingConfigCache = new ConcurrentHashMap<>(); // 库名-表名对应配置 - private Map mirrorDbConfigCache = new ConcurrentHashMap<>(); // 镜像库配置 + private Map mirrorDbConfigCache = new ConcurrentHashMap<>(); // 镜像库配置 private DruidDataSource dataSource; @@ -54,6 +55,10 @@ public class RdbAdapter implements OuterAdapter { return mappingConfigCache; } + public Map getMirrorDbConfigCache() { + return mirrorDbConfigCache; + } + /** * 初始化方法 * @@ -73,7 +78,7 @@ public class RdbAdapter implements OuterAdapter { for (Map.Entry entry : rdbMapping.entrySet()) { String configName = entry.getKey(); MappingConfig mappingConfig = entry.getValue(); - if (!mappingConfig.getDbMapping().isMirrorDb()) { + if (!mappingConfig.getDbMapping().getMirrorDb()) { Map configMap = mappingConfigCache.computeIfAbsent( StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." + mappingConfig.getDbMapping() .getDatabase() + "." + mappingConfig.getDbMapping().getTable(), @@ -81,9 +86,10 @@ public class RdbAdapter implements OuterAdapter { configMap.put(configName, mappingConfig); } else { // mirrorDB + mirrorDbConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." + mappingConfig.getDbMapping().getDatabase(), - mappingConfig); + MirrorDbConfig.create(configName, mappingConfig)); } } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java index c2fe3339..a55644f3 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java @@ -3,6 +3,7 @@ package com.alibaba.otter.canal.client.adapter.rdb.config; import java.util.LinkedHashMap; import java.util.Map; +import com.alibaba.fastjson.JSONObject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.yaml.snakeyaml.Yaml; @@ -21,9 +22,10 @@ public class ConfigLoader { /** * 加载HBase表映射配置 - * + * * @return 配置名/配置文件名--对象 */ + @SuppressWarnings("unchecked") public static Map load() { logger.info("## Start loading rdb mapping config ... "); @@ -31,7 +33,9 @@ public class ConfigLoader { Map configContentMap = MappingConfigsLoader.loadConfigs("rdb"); configContentMap.forEach((fileName, content) -> { - MappingConfig config = new Yaml().loadAs(content, MappingConfig.class); + Map configMap = new Yaml().loadAs(content, Map.class); // yml自带的对象反射不是很稳定 + JSONObject configJson = new JSONObject(configMap); + MappingConfig config = configJson.toJavaObject(MappingConfig.class); try { config.validate(); } catch (Exception e) { 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 aa6a7d1b..ce1b9f15 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 @@ -65,10 +65,10 @@ public class MappingConfig { if (dbMapping.database == null || dbMapping.database.isEmpty()) { throw new NullPointerException("dbMapping.database"); } - if (!dbMapping.isMirrorDb() && (dbMapping.table == null || dbMapping.table.isEmpty())) { + if (!dbMapping.getMirrorDb() && (dbMapping.table == null || dbMapping.table.isEmpty())) { throw new NullPointerException("dbMapping.table"); } - if (!dbMapping.isMirrorDb() && (dbMapping.targetTable == null || dbMapping.targetTable.isEmpty())) { + if (!dbMapping.getMirrorDb() && (dbMapping.targetTable == null || dbMapping.targetTable.isEmpty())) { throw new NullPointerException("dbMapping.targetTable"); } } @@ -91,8 +91,8 @@ public class MappingConfig { private Map allMapColumns; - public boolean isMirrorDb() { - return mirrorDb == null ? false : mirrorDb; + public Boolean getMirrorDb() { + return mirrorDb; } public void setMirrorDb(Boolean mirrorDb) { @@ -123,8 +123,8 @@ public class MappingConfig { this.targetPk = targetPk; } - public boolean isMapAll() { - return mapAll == null ? false : mapAll; + public Boolean getMapAll() { + return mapAll; } public void setMapAll(Boolean mapAll) { diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MirrorDbConfig.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MirrorDbConfig.java new file mode 100644 index 00000000..d84f66e7 --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MirrorDbConfig.java @@ -0,0 +1,44 @@ +package com.alibaba.otter.canal.client.adapter.rdb.config; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +public class MirrorDbConfig { + + private String fileName; + private MappingConfig mappingConfig; + Map tableConfig = new ConcurrentHashMap<>(); + + public static MirrorDbConfig create(String fileName, MappingConfig mappingConfig) { + return new MirrorDbConfig(fileName, mappingConfig); + } + + public MirrorDbConfig(String fileName, MappingConfig mappingConfig){ + this.fileName = fileName; + this.mappingConfig = mappingConfig; + } + + public String getFileName() { + return fileName; + } + + public void setFileName(String fileName) { + this.fileName = fileName; + } + + public MappingConfig getMappingConfig() { + return mappingConfig; + } + + public void setMappingConfig(MappingConfig mappingConfig) { + this.mappingConfig = mappingConfig; + } + + public Map getTableConfig() { + return tableConfig; + } + + public void setTableConfig(Map tableConfig) { + this.tableConfig = tableConfig; + } +} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java index d74cd884..271f1393 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java @@ -4,6 +4,7 @@ import java.io.File; import java.util.HashMap; import java.util.Map; +import com.alibaba.otter.canal.client.adapter.rdb.config.MirrorDbConfig; import org.apache.commons.io.filefilter.FileFilterUtils; import org.apache.commons.io.monitor.FileAlterationListenerAdaptor; import org.apache.commons.io.monitor.FileAlterationMonitor; @@ -20,7 +21,7 @@ import com.alibaba.otter.canal.client.adapter.support.Util; public class RdbConfigMonitor { - private static final Logger logger = LoggerFactory.getLogger(RdbConfigMonitor.class); + private static final Logger logger = LoggerFactory.getLogger(RdbConfigMonitor.class); private static final String adapterName = "rdb"; @@ -83,7 +84,8 @@ public class RdbConfigMonitor { try { if (rdbAdapter.getRdbMapping().containsKey(file.getName())) { // 加载配置文件 - String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName()); + String configContent = MappingConfigsLoader + .loadConfig(adapterName + File.separator + file.getName()); MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class); config.validate(); if ((key == null && config.getOuterAdapterKey() == null) @@ -118,13 +120,19 @@ public class RdbConfigMonitor { } } - private void addConfigToCache(File file, MappingConfig config) { - rdbAdapter.getRdbMapping().put(file.getName(), config); + private void addConfigToCache(File file, MappingConfig mappingConfig) { + rdbAdapter.getRdbMapping().put(file.getName(), mappingConfig); Map configMap = rdbAdapter.getMappingConfigCache() - .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "." - + config.getDbMapping().getDatabase() + "." + config.getDbMapping().getTable(), + .computeIfAbsent(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." + + mappingConfig.getDbMapping().getDatabase() + "." + + mappingConfig.getDbMapping().getTable(), k1 -> new HashMap<>()); - configMap.put(file.getName(), config); + configMap.put(file.getName(), mappingConfig); + + Map mirrorDbConfigCache = rdbAdapter.getMirrorDbConfigCache(); + mirrorDbConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." + + mappingConfig.getDbMapping().getDatabase(), + MirrorDbConfig.create(file.getName(), mappingConfig)); } private void deleteConfigFromCache(File file) { @@ -136,6 +144,12 @@ public class RdbConfigMonitor { } } + rdbAdapter.getMirrorDbConfigCache().forEach((key, mirrorDbConfig) -> { + if (mirrorDbConfig.getFileName().equals(file.getName())) { + rdbAdapter.getMirrorDbConfigCache().remove(key); + } + }); + } } } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java index 99ee875f..4a18ce99 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java @@ -6,7 +6,6 @@ import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; import javax.sql.DataSource; @@ -14,7 +13,10 @@ import org.apache.commons.lang.StringUtils; 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.MirrorDbConfig; import com.alibaba.otter.canal.client.adapter.rdb.support.SingleDml; import com.alibaba.otter.canal.client.adapter.support.Dml; @@ -26,15 +28,13 @@ import com.alibaba.otter.canal.client.adapter.support.Dml; */ public class RdbMirrorDbSyncService { - private static final Logger logger = LoggerFactory.getLogger(RdbMirrorDbSyncService.class); + private static final Logger logger = LoggerFactory.getLogger(RdbMirrorDbSyncService.class); - private Map mirrorDbConfigCache; // 镜像库配置 - private DataSource dataSource; - private RdbSyncService rdbSyncService; // rdbSyncService代理 + private Map mirrorDbConfigCache; // 镜像库配置 + private DataSource dataSource; + private RdbSyncService rdbSyncService; // rdbSyncService代理 - private final Map tableDbConfigCache = new ConcurrentHashMap<>(); // 自动生成的库表配置缓存 - - public RdbMirrorDbSyncService(Map mirrorDbConfigCache, DataSource dataSource, + public RdbMirrorDbSyncService(Map mirrorDbConfigCache, DataSource dataSource, Integer threads, Map> columnsTypeCache){ this.mirrorDbConfigCache = mirrorDbConfigCache; this.dataSource = dataSource; @@ -52,27 +52,32 @@ public class RdbMirrorDbSyncService { for (Dml dml : dmls) { String destination = StringUtils.trimToEmpty(dml.getDestination()); String database = dml.getDatabase(); - MappingConfig configMap = mirrorDbConfigCache.get(destination + "." + database); - if (configMap == null) { + MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache.get(destination + "." + database); + if (mirrorDbConfig == null) { continue; } - if (dml.getSql() != null) { + if (StringUtils.isNotEmpty(dml.getSql())) { // DDL - executeDdl(dml); + if (logger.isDebugEnabled()) { + logger.debug("DDL: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); + } + executeDdl(mirrorDbConfig, dml); rdbSyncService.getColumnsTypeCache().remove(destination + "." + database + "." + dml.getTable()); - tableDbConfigCache.remove(destination + "." + database + "." + dml.getTable()); // 删除对应库表配置 + mirrorDbConfig.getTableConfig().remove(dml.getTable()); // 删除对应库表配置 } else { // DML - initMappingConfig(destination + "." + database + "." + dml.getTable(), configMap, dml); + initMappingConfig(dml.getTable(), mirrorDbConfig.getMappingConfig(), mirrorDbConfig, dml); dmlList.add(dml); } } if (!dmlList.isEmpty()) { rdbSyncService.sync(dmlList, dml -> { + MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache + .get(dml.getDestination() + "." + dml.getDatabase()); String destination = StringUtils.trimToEmpty(dml.getDestination()); String database = dml.getDatabase(); String table = dml.getTable(); - MappingConfig config = tableDbConfigCache.get(destination + "." + database + "." + table); + MappingConfig config = mirrorDbConfig.getTableConfig().get(table); if (config == null) { return false; @@ -108,10 +113,10 @@ public class RdbMirrorDbSyncService { * @param baseConfigMap db sync config * @param dml DML */ - private void initMappingConfig(String key, MappingConfig baseConfigMap, Dml dml) { - MappingConfig mappingConfig = tableDbConfigCache.get(key); + private void initMappingConfig(String key, MappingConfig baseConfigMap, MirrorDbConfig mirrorDbConfig, Dml dml) { + MappingConfig mappingConfig = mirrorDbConfig.getTableConfig().get(key); if (mappingConfig == null) { - // 构造一个配置 + // 构造表配置 mappingConfig = new MappingConfig(); mappingConfig.setDataSourceKey(baseConfigMap.getDataSourceKey()); mappingConfig.setDestination(baseConfigMap.getDestination()); @@ -129,7 +134,7 @@ public class RdbMirrorDbSyncService { pkNames.forEach(pkName -> pkMapping.put(pkName, pkName)); dbMapping.setTargetPk(pkMapping); - tableDbConfigCache.put(key, mappingConfig); + mirrorDbConfig.getTableConfig().put(key, mappingConfig); } } @@ -138,11 +143,11 @@ public class RdbMirrorDbSyncService { * * @param ddl DDL */ - private void executeDdl(Dml ddl) { + private void executeDdl(MirrorDbConfig mirrorDbConfig, Dml ddl) { try (Connection conn = dataSource.getConnection(); Statement statement = conn.createStatement()) { statement.execute(ddl.getSql()); // 移除对应配置 - tableDbConfigCache.remove(ddl.getDatabase() + "." + ddl.getDatabase() + "." + ddl.getTable()); + mirrorDbConfig.getTableConfig().remove(ddl.getTable()); if (logger.isTraceEnabled()) { logger.trace("Execute DDL sql: {} for database: {}", ddl.getSql(), ddl.getDatabase()); } 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 bd4ef77b..93797093 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 @@ -135,7 +135,7 @@ public class RdbSyncService { public void sync(Map> mappingConfig, List dmls) { try { sync(dmls, dml -> { - if (dml.getSql() != null) { + if (StringUtils.isNotEmpty(dml.getSql())) { // DDL columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable()); return false; diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java index bf21a300..6e9b81a6 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java @@ -22,7 +22,7 @@ public class SyncUtil { public static Map getColumnsMap(MappingConfig.DbMapping dbMapping, Collection columns) { Map columnsMap; - if (dbMapping.isMapAll()) { + if (dbMapping.getMapAll()) { if (dbMapping.getAllMapColumns() != null) { return dbMapping.getAllMapColumns(); } 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 a8e97f6d..a838e1c8 100644 --- a/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml +++ b/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml @@ -1,18 +1,27 @@ dataSourceKey: defaultDS destination: example -outerAdapterKey: mysql1 +outerAdapterKey: oracle1 concurrent: true dbMapping: - mirrorDb: true database: mytest -# table: user -# targetTable: mytest.tb_user -# targetPk: -# id: id -# mapAll: true + table: user + targetTable: mytest.tb_user + targetPk: + id: id + mapAll: true # targetColumns: # id: # name: # role_id: # c_time: # test1: + + +# Mirror schema synchronize config +#dataSourceKey: defaultDS +#destination: example +#outerAdapterKey: mysql1 +#concurrent: true +#dbMapping: +# mirrorDb: true +# database: mytest