From 997646b2db5860e0750568ed1094f61025fb6f04 Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 15 Nov 2018 10:39:34 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8C=87=E5=AE=9A=E6=98=AF=E5=90=A6=E6=8C=89?= =?UTF-8?q?=E4=B8=BB=E9=94=AEhash=E5=B9=B6=E8=A1=8C=E5=90=8C=E6=AD=A5,=20?= =?UTF-8?q?=E5=90=A6=E5=88=99=E6=8C=89=E8=A1=A8=E5=90=8Dhash=E8=A1=A8?= =?UTF-8?q?=E9=97=B4=E5=B9=B6=E8=A1=8C=E5=90=8C=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client-adapter/README.md | 71 +++++++++++++++++++ .../src/main/resources/application.yml | 4 ++ client-adapter/rdb/README.md | 67 ----------------- .../canal/client/adapter/rdb/RdbAdapter.java | 5 +- .../adapter/rdb/config/MappingConfig.java | 46 +++++++----- .../adapter/rdb/service/BatchExecutor.java | 5 +- .../adapter/rdb/service/RdbSyncService.java | 66 ++++++++++++----- .../src/main/resources/rdb/mytest_user.yml | 1 + .../client/adapter/rdb/test/sync/Common.java | 9 +-- .../src/test/resources/rdb/mytest_user.yml | 2 + 10 files changed, 164 insertions(+), 112 deletions(-) delete mode 100644 client-adapter/rdb/README.md diff --git a/client-adapter/README.md b/client-adapter/README.md index 396624f6..7415b47b 100644 --- a/client-adapter/README.md +++ b/client-adapter/README.md @@ -251,3 +251,74 @@ bin/startup.sh ``` #### 验证 修改mysql mytest.person表的数据, 将会自动同步到HBase的MYTEST.PERSON表下面, 并会打出DML的log + + +## 四、关系型数据库适配器 + +RDB adapter 用于适配mysql到任意关系型数据库(需支持jdbc)的数据同步及导入 + +### 4.1 修改启动器配置: application.yml, 这里以oracle目标库为例 +``` +server: + port: 8081 +logging: + level: + com.alibaba.otter.canal.client.adapter.rdb: DEBUG +...... + +canal.conf: + canalServerHost: 127.0.0.1:11111 + srcDataSources: + defaultDS: + url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true + username: root + password: 121212 + canalInstances: + - instance: example + groups: + - outAdapters: + - name: rdb # 指定为rdb类型同步 + key: oracle1 # 指定adapter的唯一key, 与表映射配置中outerAdapterKey对应 + properties: + jdbc.driverClassName: oracle.jdbc.OracleDriver # jdbc驱动名, jdbc的jar包需要自行放致lib目录下 + jdbc.url: jdbc:oracle:thin:@localhost:49161:XE # jdbc url + jdbc.username: mytest # jdbc username + jdbc.password: m121212 # jdbc password + threads: 5 # 并行执行的线程数, 默认为1 + commitSize: 3000 # 批次提交的最大行数 +``` +其中 outAdapter 的配置: name统一为rdb, key为对应的数据源的唯一标识需和下面的表映射文件中的outerAdapterKey对应, properties为目标库jdb的相关参数 +adapter将会自动加载 conf/rdb 下的所有.yml结尾的表映射配置文件 + +### 4.2 适配器表映射文件 +修改 conf/rdb/mytest_user.yml文件: +``` +dataSourceKey: defaultDS # 源数据源的key, 对应上面配置的srcDataSources中的值 +destination: example # cannal的instance或者MQ的topic +outerAdapterKey: oracle1 # adapter key, 对应上面配置outAdapters中的key +concurrent: true # 是否按主键hase并行同步, 并行同步的表必须保证主键不会更改及主键不能为其他同步表的外键!! +dbMapping: + database: mytest # 源数据源的database/shcema + table: user # 源数据源表名 + targetTable: mytest.tb_user # 目标数据源的库名.表名 + targetPk: # 主键映射 + id: id # 如果是复合主键可以换行映射多个 +# mapAll: true # 是否整表映射, 要求源表和目标表字段名一模一样 (如果targetColumns也配置了映射,则以targetColumns配置为准) + targetColumns: # 字段映射, 格式: 目标表字段: 源表字段, 如果字段名一样源表字段名可不填 + id: + name: + role_id: + c_time: + test1: +``` +导入的类型以目标表的元类型为准, 将自动转换 + +### 4.3 启动RDB数据同步 +#### 将目标库的jdbc jar包放入lib文件夹, 这里放入ojdbc6.jar + +#### 启动canal-adapter启动器 +``` +bin/startup.sh +``` +#### 验证 +修改mysql mytest.user表的数据, 将会自动同步到Oracle的MYTEST.TB_USER表下面, 并会打出DML的log diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 24cff2b6..e5d4f507 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -35,6 +35,8 @@ canal.conf: # jdbc.url: jdbc:oracle:thin:@localhost:49161:XE # jdbc.username: mytest # jdbc.password: m121212 +# threads: 5 +# commitSize: 3000 # - name: rdb # key: postgres1 # properties: @@ -42,6 +44,8 @@ canal.conf: # 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/README.md b/client-adapter/rdb/README.md deleted file mode 100644 index b579b671..00000000 --- a/client-adapter/rdb/README.md +++ /dev/null @@ -1,67 +0,0 @@ -## RDB适配器 -RDB adapter 用于适配mysql到任意关系型数据库(需支持jdbc)的数据同步及导入 -### 1.1 修改启动器配置: application.yml, 这里以oracle目标库为例 -``` -server: - port: 8081 -logging: - level: - com.alibaba.otter.canal.client.adapter.hbase: DEBUG -spring: - jackson: - date-format: yyyy-MM-dd HH:mm:ss - time-zone: GMT+8 - default-property-inclusion: non_null - -canal.conf: - canalServerHost: 127.0.0.1:11111 - srcDataSources: - defaultDS: - url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true - username: root - password: 121212 - canalInstances: - - instance: example - groups: - - outAdapters: - - name: rdb - key: oracle1 - properties: - jdbc.driverClassName: oracle.jdbc.OracleDriver - jdbc.url: jdbc:oracle:thin:@localhost:49161:XE - jdbc.username: mytest - jdbc.password: m121212 -``` -其中 outAdapter 的配置: name统一为rdb, key为对应的数据源的唯一标识需和下面的表映射文件中的outerAdapterKey对应!! properties为目标库jdb的相关参数 -adapter将会自动加载 conf/rdb 下的所有.yml结尾的表映射配置文件 -### 1.2 适配器表映射文件 -修改 conf/rdb/mytest_user.yml文件: -``` -dataSourceKey: defaultDS # 源数据源的key, 对应上面配置的srcDataSources中的值 -destination: example # cannal的instance或者MQ的topic -outerAdapterKey: oracle1 # adapter key, 对应上面配置outAdapters中的key -dbMapping: - database: mytest # 源数据源的database/shcema - table: user # 源数据源表名 - targetTable: mytest.tb_user # 目标数据源的库名.表名 - targetPk: # 主键映射 - id: id # 如果是复合主键可以换行映射多个 - mapAll: true # 是否整表映射, 要求源表和目标表字段名一模一样 (如果targetColumns也配置了映射,则以targetColumns配置为准) -# targetColumns: # 字段映射, 格式: 目标表字段: 源表字段, 如果字段名一样源表字段名可不填 -# id: -# name: -# role_id: -# c_time: -# test1: -``` -导入的类型以目标表的元类型为准, 将自动转换 - -### 1.3 启动RDB数据同步 -#### 将目标库的jdbc jar包放入lib文件夹, 这里放入ojdbc6.jar - -#### 启动canal-adapter启动器 -``` -bin/startup.sh -``` -#### 验证 -修改mysql mytest.user表的数据, 将会自动同步到Oracle的MYTEST.TB_USER表下面, 并会打出DML的log 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 4431e407..68f740a7 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,7 @@ 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.*; import javax.sql.DataSource; diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java index 40389f9a..1d32147a 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 @@ -18,6 +18,8 @@ public class MappingConfig { private String outerAdapterKey; // 对应适配器的key + private Boolean concurrent; // 是否并行同步 + private DbMapping dbMapping; // db映射配置 public String getDataSourceKey() { @@ -36,6 +38,14 @@ public class MappingConfig { this.outerAdapterKey = outerAdapterKey; } + public Boolean getConcurrent() { + return concurrent == null ? false : concurrent; + } + + public void setConcurrent(Boolean concurrent) { + this.concurrent = concurrent; + } + public DbMapping getDbMapping() { return dbMapping; } @@ -66,20 +76,20 @@ public class MappingConfig { 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 String database; // 数据库名或schema名 + private String table; // 表面名 + private Map targetPk; // 目标表主键字段 + private boolean mapAll = false; // 映射所有字段 + 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 Set families = new LinkedHashSet<>(); // column family列表 + private int readBatch = 5000; + private int commitBatch = 5000; // etl等批量提交大小 -// private volatile Map allColumns; // mapAll为true,自动设置改字段 + // private volatile Map allColumns; // mapAll为true,自动设置改字段 public String getDatabase() { return database; @@ -161,12 +171,12 @@ public class MappingConfig { this.commitBatch = commitBatch; } -// public Map getAllColumns() { -// return allColumns; -// } -// -// public void setAllColumns(Map allColumns) { -// this.allColumns = allColumns; -// } + // public Map getAllColumns() { + // return allColumns; + // } + // + // public void setAllColumns(Map allColumns) { + // this.allColumns = allColumns; + // } } } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java index e079c335..ef0ec222 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java @@ -78,12 +78,13 @@ public class BatchExecutor { executor.submit(() -> { try { commitLock.lock(); + conn.commit(); //直接提交一次 if (!condition.await(5, TimeUnit.SECONDS)) { // 超时提交 commit(); } - } catch (InterruptedException e) { - // ignore + } catch (Exception e) { + logger.error(e.getMessage(), e); } finally { commitLock.unlock(); } 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 3755179b..0eff385b 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 @@ -48,7 +48,7 @@ public class RdbSyncService { public RdbSyncService(Integer commitSize, Integer threads, DataSource dataSource){ try { - if (threads != null && threads > 1 && threads < 10) { + if (threads != null && threads > 1 && threads <= 10) { this.threads = threads; } batchExecutors = new BatchExecutor[this.threads]; @@ -121,13 +121,24 @@ public class RdbSyncService { Map ctype = getTargetColumnType(batchExecutors[0].getConn(), config); String sql = insertSql.toString(); + Integer tbHash = null; + // 如果不是并行同步的表 + if (!config.getConcurrent()) { + // 按表名hash到一个线程 + tbHash = Math.abs(Math.abs(dbMapping.getTargetTable().hashCode()) % threads); + } for (Map d : data) { - int pkHash = pkHash(dbMapping, d, threads); - ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[pkHash]; + int hash; + if (tbHash != null) { + hash = tbHash; + } else { + hash = pkHash(dbMapping, d, threads); + } + ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[hash]; checkQueue(tpe); tpe.submit(() -> { try { - BatchExecutor batchExecutor = batchExecutors[pkHash]; + BatchExecutor batchExecutor = batchExecutors[hash]; List> values = new ArrayList<>(); for (Map.Entry entry : columnsMap.entrySet()) { @@ -176,15 +187,26 @@ public class RdbSyncService { Map ctype = getTargetColumnType(batchExecutors[0].getConn(), config); + Integer tbHash = null; + if (!config.getConcurrent()) { + // 按表名hash到一个线程 + tbHash = Math.abs(Math.abs(dbMapping.getTargetTable().hashCode()) % threads); + } for (Map o : old) { Map d = data.get(idx - 1); - int pkHash = pkHash(dbMapping, d, o, threads); - ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[pkHash]; + int hash; + if (tbHash != null) { + hash = tbHash; + } else { + hash = pkHash(dbMapping, d, o, threads); + } + + ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[hash]; checkQueue(tpe); tpe.submit(() -> { try { - BatchExecutor batchExecutor = batchExecutors[pkHash]; + BatchExecutor batchExecutor = batchExecutors[hash]; StringBuilder updateSql = new StringBuilder(); updateSql.append("UPDATE ").append(dbMapping.getTargetTable()).append(" SET "); @@ -211,10 +233,11 @@ public class RdbSyncService { // 拼接主键 appendCondition(dbMapping, updateSql, ctype, values, d, o); - if (logger.isTraceEnabled()) { - logger.trace("Update target table, sql: {}", updateSql); - } batchExecutor.execute(updateSql.toString(), values); + + if (logger.isDebugEnabled()) { + logger.debug("Execute sql: {}", updateSql); + } } catch (Exception e) { logger.error(e.getMessage(), e); } @@ -240,15 +263,24 @@ public class RdbSyncService { DbMapping dbMapping = config.getDbMapping(); Map ctype = getTargetColumnType(batchExecutors[0].getConn(), config); - + Integer tbHash = null; + if (!config.getConcurrent()) { + // 按表名hash到一个线程 + tbHash = Math.abs(Math.abs(dbMapping.getTargetTable().hashCode()) % threads); + } for (Map d : data) { - int pkHash = pkHash(dbMapping, d, threads); + int hash; + if (tbHash != null) { + hash = tbHash; + } else { + hash = pkHash(dbMapping, d, threads); + } - ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[pkHash]; + ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[hash]; checkQueue(tpe); tpe.submit(() -> { try { - BatchExecutor batchExecutor = batchExecutors[pkHash]; + BatchExecutor batchExecutor = batchExecutors[hash]; StringBuilder sql = new StringBuilder(); sql.append("DELETE FROM ").append(dbMapping.getTargetTable()).append(" WHERE "); @@ -257,10 +289,10 @@ public class RdbSyncService { // 拼接主键 appendCondition(dbMapping, sql, ctype, values, d); - if (logger.isTraceEnabled()) { - logger.trace("Delete from target table, sql: {}", sql); - } batchExecutor.execute(sql.toString(), values); + if (logger.isDebugEnabled()) { + logger.debug("Execute sql: {}", sql); + } } catch (Exception e) { logger.error(e.getMessage(), e); } 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 56d8e518..9b40dd75 100644 --- a/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml +++ b/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml @@ -1,6 +1,7 @@ dataSourceKey: defaultDS destination: example outerAdapterKey: oracle1 +concurrent: true dbMapping: database: mytest table: user diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java index 5af9fdcb..9d2d117d 100644 --- a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java @@ -1,20 +1,21 @@ package com.alibaba.otter.canal.client.adapter.rdb.test.sync; +import java.util.HashMap; +import java.util.Map; + import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; import com.alibaba.otter.canal.client.adapter.rdb.test.TestConstant; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; -import java.util.HashMap; -import java.util.Map; - public class Common { + public static RdbAdapter init() { DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource); OuterAdapterConfig outerAdapterConfig = new OuterAdapterConfig(); outerAdapterConfig.setName("rdb"); - //outerAdapterConfig.setKey("oralce1"); + outerAdapterConfig.setKey("oracle1"); Map properties = new HashMap<>(); properties.put("jdbc.driveClassName", "oracle.jdbc.OracleDriver"); properties.put("jdbc.url", "jdbc:oracle:thin:@127.0.0.1:49161:XE"); diff --git a/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml b/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml index a7bede03..9b40dd75 100644 --- a/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml +++ b/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml @@ -1,5 +1,7 @@ dataSourceKey: defaultDS destination: example +outerAdapterKey: oracle1 +concurrent: true dbMapping: database: mytest table: user