From d7ab6e2dc109325c7f584c3d66d075f2640ce42e Mon Sep 17 00:00:00 2001 From: machey Date: Sat, 10 Nov 2018 23:55:09 +0800 Subject: [PATCH] =?UTF-8?q?rdb=E5=90=8C=E6=AD=A5=E9=80=82=E9=85=8D?= =?UTF-8?q?=E5=A4=9A=E6=95=B0=E6=8D=AE=E6=BA=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../adapter/support/CanalClientConfig.java | 17 ++++++- .../adapter/support/ExtensionLoader.java | 16 +++++++ .../adapter/support/OuterAdapterConfig.java | 10 ++++ .../launcher/loader/CanalAdapterLoader.java | 8 +++- .../src/main/resources/application.yml | 29 +++++------ client-adapter/rdb/pom.xml | 48 +++++++++---------- .../canal/client/adapter/rdb/RdbAdapter.java | 35 +++++++------- .../adapter/rdb/adapters/MysqlAdapter.java | 8 ---- .../adapter/rdb/adapters/OracleAdapter.java | 8 ---- .../rdb/adapters/PostgresqlAdapter.java | 8 ---- .../rdb/adapters/SqlserverAdapter.java | 8 ---- .../adapter/rdb/config/MappingConfig.java | 16 +++++-- ...ba.otter.canal.client.adapter.OuterAdapter | 5 +- .../main/resources/postgresql/mytest_user.yml | 15 ------ .../resources/{oracle => rdb}/mytest_user.yml | 1 + .../client/adapter/rdb/test/sync/Common.java | 6 +-- .../adapter/rdb/test/sync/OracleSyncTest.java | 2 +- 17 files changed, 125 insertions(+), 115 deletions(-) delete mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/MysqlAdapter.java delete mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/OracleAdapter.java delete mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/PostgresqlAdapter.java delete mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/SqlserverAdapter.java delete mode 100644 client-adapter/rdb/src/main/resources/postgresql/mytest_user.yml rename client-adapter/rdb/src/main/resources/{oracle => rdb}/mytest_user.yml (90%) diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java index 3d16a067..97a7a426 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java @@ -1,7 +1,9 @@ package com.alibaba.otter.canal.client.adapter.support; import java.util.ArrayList; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; /** * 配置信息类 @@ -129,7 +131,9 @@ public class CanalClientConfig { public static class Group { - private List outAdapters; // 适配器列表 + private List outAdapters; // 适配器列表 + + private Map outAdaptersMap = new LinkedHashMap<>(); public List getOutAdapters() { return outAdapters; @@ -137,6 +141,17 @@ public class CanalClientConfig { public void setOutAdapters(List outAdapters) { this.outAdapters = outAdapters; + if (outAdapters != null) { + outAdapters.forEach(outAdapter -> outAdaptersMap.put(outAdapter.getKey(), outAdapter)); + } + } + + public Map getOutAdaptersMap() { + return outAdaptersMap; + } + + public void setOutAdaptersMap(Map outAdaptersMap) { + this.outAdaptersMap = outAdaptersMap; } } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java index cf69fde5..ce7de834 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java @@ -148,6 +148,22 @@ public class ExtensionLoader { } } + @SuppressWarnings("unchecked") + public T newInstance(String name) { + Class clazz = getExtensionClasses().get(name); + if (clazz == null) { + throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type + + ") could not be instantiated: class could not be found"); + } + try { + return (T) clazz.newInstance(); + } catch (Throwable t) { + throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type + + ") could not be instantiated: " + t.getMessage(), + t); + } + } + private Map> getExtensionClasses() { Map> classes = cachedClasses.get(); if (classes == null) { diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/OuterAdapterConfig.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/OuterAdapterConfig.java index 4e73791e..a767b203 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/OuterAdapterConfig.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/OuterAdapterConfig.java @@ -12,6 +12,8 @@ public class OuterAdapterConfig { private String name; // 适配器名称, 如: logger, hbase, es + private String key; // 适配器唯一键 + private String hosts; // 适配器内部的地址, 比如对应es该参数可以填写es的server地址 private String zkHosts; // 适配器内部的ZK地址, 比如对应HBase该参数可以填写HBase对应的ZK地址 @@ -26,6 +28,14 @@ public class OuterAdapterConfig { this.name = name; } + public String getKey() { + return key; + } + + public void setKey(String key) { + this.key = key; + } + public String getHosts() { return hosts; } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index df641c45..78eab682 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -123,7 +123,12 @@ public class CanalAdapterLoader { private void loadConnector(OuterAdapterConfig config, List canalOutConnectors) { try { - OuterAdapter adapter = loader.getExtension(config.getName()); + OuterAdapter adapter; + if ("rdb".equalsIgnoreCase(config.getName())) { + adapter = loader.newInstance(config.getName()); + } else { + adapter = loader.getExtension(config.getName()); + } ClassLoader cl = Thread.currentThread().getContextClassLoader(); // 替换ClassLoader Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader()); @@ -133,6 +138,7 @@ public class CanalAdapterLoader { logger.info("Load canal adapter: {} succeed", config.getName()); } catch (Exception e) { logger.error("Load canal adapter: {} failed", config.getName(), e); + System.exit(0); } } diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 2de61937..9595e3fb 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -21,6 +21,20 @@ spring: # groups: # - outAdapters: # - name: logger +# - name: rdb +# key: oracle1 +# properties: +# jdbc.driverClassName: oracle.jdbc.OracleDriver +# jdbc.url: jdbc:oracle:thin:@127.0.0.1:1521:orcl +# 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 # - name: hbase # properties: # hbase.zookeeper.quorum: 127.0.0.1 @@ -37,18 +51,6 @@ spring: # - groupId: g2 # outAdapters: # - name: logger -# - name: oracle -# properties: -# jdbc.driverClassName: oracle.jdbc.OracleDriver -# jdbc.url: jdbc:oracle:thin:@127.0.0.1:49161:XE -# jdbc.username: mytest -# jdbc.password: m121212 -# - name: postgresql -# properties: -# jdbc.driverClassName: org.postgresql.Driver -# jdbc.url: jdbc:postgresql://localhost:5432/postgres -# jdbc.username: postgres -# jdbc.password: 121212 # mqTopics: # - mqMode: rocketmq # topic: example @@ -66,5 +68,4 @@ spring: # adapterConfigs: # - hbase/mytest_person2.yml # - es/mytest_user.yml -# - oracle/mytest_user.yml -# - postgresql/mytest_user.yml +# - rdb/mytest_user.yml diff --git a/client-adapter/rdb/pom.xml b/client-adapter/rdb/pom.xml index 951579a9..994d28aa 100644 --- a/client-adapter/rdb/pom.xml +++ b/client-adapter/rdb/pom.xml @@ -27,30 +27,30 @@ provided - - mysql - mysql-connector-java - 5.1.40 - test - - - org.postgresql - postgresql - 42.1.4 - test - - - com.oracle - ojdbc6 - 11.2.0.4.0-atlassian-hosted - test - - - com.microsoft.sqlserver - mssql-jdbc - 7.0.0.jre8 - test - + + + + + + + + + + + + + + + + + + + + + + + + junit junit 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 5dd47a05..c07dee6b 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 @@ -19,12 +19,13 @@ import com.alibaba.otter.canal.client.adapter.rdb.service.RdbEtlService; import com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService; import com.alibaba.otter.canal.client.adapter.support.*; -public abstract class RdbAdapter implements OuterAdapter { +@SPI("rdb") +public class RdbAdapter implements OuterAdapter { - private static Logger logger = LoggerFactory.getLogger(RdbAdapter.class); + private static Logger logger = LoggerFactory.getLogger(RdbAdapter.class); - private volatile Map rdbMapping = null; // 文件名对应配置 - private volatile Map mappingConfigCache = null; // 库名-表名对应配置 + private volatile Map rdbMapping = new HashMap<>(); // 文件名对应配置 + private Map mappingConfigCache; // 库名-表名对应配置 private DruidDataSource dataSource; @@ -32,20 +33,20 @@ public abstract class RdbAdapter implements OuterAdapter { @Override public void init(OuterAdapterConfig configuration) { - if (mappingConfigCache == null) { - synchronized (MappingConfig.class) { - if (mappingConfigCache == null) { - SPI spi = this.getClass().getAnnotation(SPI.class); - rdbMapping = MappingConfigLoader.load(spi.value()); - mappingConfigCache = new HashMap<>(); - for (MappingConfig mappingConfig : rdbMapping.values()) { - mappingConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." - + mappingConfig.getDbMapping().getDatabase() + "." - + mappingConfig.getDbMapping().getTable(), - mappingConfig); - } - } + SPI spi = this.getClass().getAnnotation(SPI.class); + Map rdbMappingTmp = MappingConfigLoader.load(spi.value()); + // 过滤其他key的配置 + rdbMappingTmp.forEach((key, mappingConfig) -> { + if (mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey())) { + rdbMapping.put(key, mappingConfig); } + }); + mappingConfigCache = new HashMap<>(); + for (MappingConfig mappingConfig : rdbMapping.values()) { + mappingConfigCache + .put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." + + mappingConfig.getDbMapping().getDatabase() + "." + mappingConfig.getDbMapping().getTable(), + mappingConfig); } Map properties = configuration.getProperties(); diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/MysqlAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/MysqlAdapter.java deleted file mode 100644 index ea73b1c0..00000000 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/MysqlAdapter.java +++ /dev/null @@ -1,8 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.rdb.adapters; - -import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; -import com.alibaba.otter.canal.client.adapter.support.SPI; - -@SPI("mysql") -public class MysqlAdapter extends RdbAdapter { -} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/OracleAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/OracleAdapter.java deleted file mode 100644 index fe6111a2..00000000 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/OracleAdapter.java +++ /dev/null @@ -1,8 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.rdb.adapters; - -import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; -import com.alibaba.otter.canal.client.adapter.support.SPI; - -@SPI("oracle") -public class OracleAdapter extends RdbAdapter { -} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/PostgresqlAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/PostgresqlAdapter.java deleted file mode 100644 index 2cf4a444..00000000 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/PostgresqlAdapter.java +++ /dev/null @@ -1,8 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.rdb.adapters; - -import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; -import com.alibaba.otter.canal.client.adapter.support.SPI; - -@SPI("postgresql") -public class PostgresqlAdapter extends RdbAdapter { -} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/SqlserverAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/SqlserverAdapter.java deleted file mode 100644 index 2fe7e0ed..00000000 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/SqlserverAdapter.java +++ /dev/null @@ -1,8 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.rdb.adapters; - -import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; -import com.alibaba.otter.canal.client.adapter.support.SPI; - -@SPI("sqlserver") -public class SqlserverAdapter extends RdbAdapter { -} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java index da7e6ef8..95623ea4 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 @@ -12,11 +12,13 @@ import java.util.Set; */ public class MappingConfig { - private String dataSourceKey; // 数据源key + private String dataSourceKey; // 数据源key - private String destination; // canal实例或MQ的topic + private String destination; // canal实例或MQ的topic - private DbMapping dbMapping; // db映射配置 + private String outerAdapterKey; // 对应适配器的key + + private DbMapping dbMapping; // db映射配置 public String getDataSourceKey() { return dataSourceKey; @@ -26,6 +28,14 @@ public class MappingConfig { this.dataSourceKey = dataSourceKey; } + public String getOuterAdapterKey() { + return outerAdapterKey; + } + + public void setOuterAdapterKey(String outerAdapterKey) { + this.outerAdapterKey = outerAdapterKey; + } + public DbMapping getDbMapping() { return dbMapping; } diff --git a/client-adapter/rdb/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter b/client-adapter/rdb/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter index c005f7fd..0a840275 100644 --- a/client-adapter/rdb/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter +++ b/client-adapter/rdb/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter @@ -1,4 +1 @@ -mysql=com.alibaba.otter.canal.client.adapter.rdb.adapters.MysqlAdapter -postgresql=com.alibaba.otter.canal.client.adapter.rdb.adapters.PostgresqlAdapter -oracle=com.alibaba.otter.canal.client.adapter.rdb.adapters.OracleAdapter -sqlserver=com.alibaba.otter.canal.client.adapter.rdb.adapters.SqlserverAdapter \ No newline at end of file +rdb=com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter \ No newline at end of file diff --git a/client-adapter/rdb/src/main/resources/postgresql/mytest_user.yml b/client-adapter/rdb/src/main/resources/postgresql/mytest_user.yml deleted file mode 100644 index 1b4eaed7..00000000 --- a/client-adapter/rdb/src/main/resources/postgresql/mytest_user.yml +++ /dev/null @@ -1,15 +0,0 @@ -dataSourceKey: defaultDS -destination: example -dbMapping: - database: mytest - table: user - targetTable: public.tb_user - targetPk: - id: id - mapAll: true -# targetColumns: -# id: -# name: -# role_id: -# c_time: -# test1: \ No newline at end of file diff --git a/client-adapter/rdb/src/main/resources/oracle/mytest_user.yml b/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml similarity index 90% rename from client-adapter/rdb/src/main/resources/oracle/mytest_user.yml rename to client-adapter/rdb/src/main/resources/rdb/mytest_user.yml index a7bede03..56d8e518 100644 --- a/client-adapter/rdb/src/main/resources/oracle/mytest_user.yml +++ b/client-adapter/rdb/src/main/resources/rdb/mytest_user.yml @@ -1,5 +1,6 @@ dataSourceKey: defaultDS destination: example +outerAdapterKey: oracle1 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 0682332a..4ae04643 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,7 +1,6 @@ package com.alibaba.otter.canal.client.adapter.rdb.test.sync; import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; -import com.alibaba.otter.canal.client.adapter.rdb.adapters.OracleAdapter; import com.alibaba.otter.canal.client.adapter.rdb.test.TestConstant; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; @@ -14,7 +13,8 @@ public class Common { DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource); OuterAdapterConfig outerAdapterConfig = new OuterAdapterConfig(); - outerAdapterConfig.setName("oracle"); + outerAdapterConfig.setName("rdb"); + outerAdapterConfig.setKey("oralce1"); Map properties = new HashMap<>(); properties.put("jdbc.driveClassName", "oracle.jdbc.OracleDriver"); properties.put("jdbc.url", "jdbc:oracle:thin:@127.0.0.1:49161:XE"); @@ -22,7 +22,7 @@ public class Common { properties.put("jdbc.password", "m121212"); outerAdapterConfig.setProperties(properties); - RdbAdapter adapter = new OracleAdapter(); + RdbAdapter adapter = new RdbAdapter(); adapter.init(outerAdapterConfig); return adapter; } diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java index 514ffaa5..5b8a45e3 100644 --- a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java @@ -15,7 +15,7 @@ public class OracleSyncTest { @Before public void init() { - AdapterConfigs.put("oracle", "mytest_user.yml"); + AdapterConfigs.put("rdb", "mytest_user.yml"); rdbAdapter = Common.init(); }