diff --git a/client-adapter/README.md b/client-adapter/README.md index 0164aac0..396624f6 100644 --- a/client-adapter/README.md +++ b/client-adapter/README.md @@ -146,6 +146,11 @@ spring: canal.conf: canalServerHost: 127.0.0.1:11111 flatMessage: true + srcDataSources: + defaultDS: + url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true + username: root + password: 121212 canalInstances: - instance: example adapterGroups: @@ -155,17 +160,8 @@ canal.conf: hbase.zookeeper.quorum: slave1 hbase.zookeeper.property.clientPort: 2181 zookeeper.znode.parent: /hbase - -adapter.conf: - datasourceConfigs: - defaultDS: - url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true - username: root - password: 121212 - adapterConfigs: - - hbase/mytest_person.yml ``` -其中指定了一个HBase表映射文件: mytest_person.yml +adapter将会自动加载 conf/hbase 下的所有.yml结尾的配置文件 ### 3.2 适配器表映射文件 修改 conf/hbase/mytest_person.yml文件: ``` @@ -251,7 +247,7 @@ create 'MYTEST.PERSON', {NAME=>'CF'} ``` #### 启动canal-adapter启动器 ``` -java -jar canal-adapter-launcher.jar +bin/startup.sh ``` #### 验证 修改mysql mytest.person表的数据, 将会自动同步到HBase的MYTEST.PERSON表下面, 并会打出DML的log diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java deleted file mode 100644 index 0ccfb509..00000000 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java +++ /dev/null @@ -1,41 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.support; - -import java.util.HashMap; -import java.util.LinkedHashSet; -import java.util.Map; -import java.util.Set; - -/** - * 适配器配置集合, 用于配置加载, 线程不安全 - * - * @author rewerma @ 2018-10-20 - * @version 1.0.0 - */ -public class AdapterConfigs { - - /** - * 类型下对应所有配置名, 如: - * hbase - * ┗━ mytest_person.yml - * ┗━ mytest_role.yml - * ┗━ mytest_department.yml - */ - private static Map> configs = new HashMap<>(); - - public static void put(String key, String value) { - Set values = configs.get(key); - if (values == null) { - values = new LinkedHashSet<>(); - } - values.add(value); - configs.put(key, values); - } - - public static Set get(String key) { - return configs.get(key); - } - - public static void clear() { - configs.clear(); - } -} 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 5a09d27f..62ea78fd 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 @@ -13,6 +13,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.regex.Pattern; +import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -40,7 +41,9 @@ public class ExtensionLoader { private static final ConcurrentMap, Object> EXTENSION_INSTANCES = new ConcurrentHashMap<>(); - private static final ConcurrentMap EXTENSION_KEY_INSTANCES = new ConcurrentHashMap<>(); + private static final ConcurrentMap EXTENSION_KEY_INSTANCE = new ConcurrentHashMap<>(); + + private static final ConcurrentMap> EXTENSION_KEY_INSTANCES = new ConcurrentHashMap<>(); private final Class type; @@ -124,10 +127,11 @@ public class ExtensionLoader { if ("true".equals(name)) { return getDefaultExtension(); } - Holder holder = cachedInstances.get(name + "-" + key); + String extKey = name + "-" + StringUtils.trimToEmpty(key); + Holder holder = cachedInstances.get(extKey); if (holder == null) { - cachedInstances.putIfAbsent(name + "-" + key, new Holder<>()); - holder = cachedInstances.get(name + "-" + key); + cachedInstances.putIfAbsent(extKey, new Holder<>()); + holder = cachedInstances.get(extKey); } Object instance = holder.get(); if (instance == null) { @@ -182,10 +186,10 @@ public class ExtensionLoader { + ") could not be instantiated: class could not be found"); } try { - T instance = (T) EXTENSION_KEY_INSTANCES.get(name + "-" + key); + T instance = (T) EXTENSION_KEY_INSTANCE.get(name + "-" + key); if (instance == null) { - EXTENSION_KEY_INSTANCES.putIfAbsent(name + "-" + key, clazz.newInstance()); - instance = (T) EXTENSION_KEY_INSTANCES.get(name + "-" + key); + EXTENSION_KEY_INSTANCE.putIfAbsent(name + "-" + key, clazz.newInstance()); + instance = (T) EXTENSION_KEY_INSTANCE.get(name + "-" + key); } return instance; } catch (Throwable t) { @@ -195,24 +199,6 @@ 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/MappingConfigsLoader.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java new file mode 100644 index 00000000..711a52b6 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java @@ -0,0 +1,46 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; + +public class MappingConfigsLoader { + + public static Map loadConfigs(String name) { + Map configContentMap = new HashMap<>(); + + // 先取本地文件，再取类路径 + File configDir = new File("../conf/" + name); + if (!configDir.exists()) { + URL url = MappingConfigsLoader.class.getClassLoader().getResource(""); + if (url != null) { + configDir = new File(url.getPath() + name + File.separator); + } + } + + File[] files = configDir.listFiles(); + if (files != null) { + for (File file : files) { + String fileName = file.getName(); + if (!fileName.endsWith(".yml")) { + continue; + } + try (InputStream in = new FileInputStream(file)) { + byte[] bytes = new byte[in.available()]; + in.read(bytes); + String configContent = new String(bytes, StandardCharsets.UTF_8); + configContentMap.put(fileName, configContent); + } catch (IOException e) { + throw new RuntimeException("Read " + name + "mapping config: " + fileName + " error. ", e); + } + } + } + + return configContentMap; + } +} diff --git a/client-adapter/elasticsearch/pom.xml b/client-adapter/elasticsearch/pom.xml index bff1bffd..2d6c3def 100644 --- a/client-adapter/elasticsearch/pom.xml +++ b/client-adapter/elasticsearch/pom.xml @@ -71,6 +71,26 @@ + + maven-antrun-plugin + + + package + + run + + + + + + + + + + + + + diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java index 7d10d702..24cc9029 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java @@ -1,9 +1,12 @@ package com.alibaba.otter.canal.client.adapter.es; import java.net.InetAddress; +import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.regex.Matcher; +import java.util.regex.Pattern; import javax.sql.DataSource; @@ -13,10 +16,13 @@ import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.transport.TransportAddress; import org.elasticsearch.transport.client.PreBuiltTransportClient; +import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader; +import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem; +import com.alibaba.otter.canal.client.adapter.es.config.SqlParser; import com.alibaba.otter.canal.client.adapter.es.service.ESEtlService; import com.alibaba.otter.canal.client.adapter.es.service.ESSyncService; import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate; @@ -31,9 +37,12 @@ import com.alibaba.otter.canal.client.adapter.support.*; @SPI("es") public class ESAdapter implements OuterAdapter { - private TransportClient transportClient; + private Map esSyncConfig = new LinkedHashMap<>(); // 文件名对应配置 + private Map> dbTableEsSyncConfig = new LinkedHashMap<>(); // schema-table对应配置 - private ESSyncService esSyncService; + private TransportClient transportClient; + + private ESSyncService esSyncService; public TransportClient getTransportClient() { return transportClient; @@ -43,10 +52,48 @@ public class ESAdapter implements OuterAdapter { return esSyncService; } + public Map getEsSyncConfig() { + return esSyncConfig; + } + + public Map> getDbTableEsSyncConfig() { + return dbTableEsSyncConfig; + } + @Override public void init(OuterAdapterConfig configuration) { try { - ESSyncConfigLoader.load(); + Map esSyncConfigTmp = ESSyncConfigLoader.load(); + // 过滤不匹配的key的配置 + esSyncConfigTmp.forEach((key, config) -> { + if ((config.getOuterAdapterKey() == null && configuration.getKey() == null) + || (config.getOuterAdapterKey() != null + && config.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) { + esSyncConfig.put(key, config); + } + }); + + for (ESSyncConfig config : esSyncConfig.values()) { + SchemaItem schemaItem = SqlParser.parse(config.getEsMapping().getSql()); + config.getEsMapping().setSchemaItem(schemaItem); + + DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); + if (dataSource == null || dataSource.getUrl() == null) { + throw new RuntimeException("No data source found: " + config.getDataSourceKey()); + } + Pattern pattern = Pattern.compile(".*:(.*)://.*/(.*)\\?.*$"); + Matcher matcher = pattern.matcher(dataSource.getUrl()); + if (!matcher.find()) { + throw new RuntimeException("Not found the schema of jdbc-url: " + config.getDataSourceKey()); + } + String schema = matcher.group(2); + + schemaItem.getAliasTableItems().values().forEach(tableItem -> { + List esSyncConfigs = dbTableEsSyncConfig + .computeIfAbsent(schema + "-" + tableItem.getTableName(), k -> new ArrayList<>()); + esSyncConfigs.add(config); + }); + } Map properties = configuration.getProperties(); Settings.Builder settingBuilder = Settings.builder(); @@ -68,13 +115,16 @@ public class ESAdapter implements OuterAdapter { @Override public void sync(Dml dml) { - esSyncService.sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = dbTableEsSyncConfig.get(database + "-" + table); + esSyncService.sync(esSyncConfigs, dml); } @Override public EtlResult etl(String task, List params) { EtlResult etlResult = new EtlResult(); - ESSyncConfig config = ESSyncConfigLoader.getEsSyncConfig().get(task); + ESSyncConfig config = esSyncConfig.get(task); if (config != null) { DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); ESEtlService esEtlService = new ESEtlService(transportClient, config); @@ -89,7 +139,7 @@ public class ESAdapter implements OuterAdapter { StringBuilder resultMsg = new StringBuilder(); boolean resSuccess = true; // ds不为空说明传入的是datasourceKey - for (ESSyncConfig configTmp : ESSyncConfigLoader.getEsSyncConfig().values()) { + for (ESSyncConfig configTmp : esSyncConfig.values()) { // 取所有的destination为task的配置 if (configTmp.getDestination().equals(task)) { ESEtlService esEtlService = new ESEtlService(transportClient, configTmp); @@ -119,7 +169,7 @@ public class ESAdapter implements OuterAdapter { @Override public Map count(String task) { - ESSyncConfig config = ESSyncConfigLoader.getEsSyncConfig().get(task); + ESSyncConfig config = esSyncConfig.get(task); ESMapping mapping = config.getEsMapping(); SearchResponse response = transportClient.prepareSearch(mapping.get_index()) .setTypes(mapping.get_type()) @@ -142,7 +192,7 @@ public class ESAdapter implements OuterAdapter { @Override public String getDestination(String task) { - ESSyncConfig config = ESSyncConfigLoader.getEsSyncConfig().get(task); + ESSyncConfig config = esSyncConfig.get(task); if (config != null) { return config.getDestination(); } diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java index 9d4a9da2..65087437 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java @@ -13,9 +13,11 @@ import java.util.Map; */ public class ESSyncConfig { - private String dataSourceKey; // 数据源key + private String dataSourceKey; // 数据源key - private String destination; // canal destination + private String outerAdapterKey; // adapter key + + private String destination; // canal destination private ESMapping esMapping; @@ -42,6 +44,14 @@ public class ESSyncConfig { this.dataSourceKey = dataSourceKey; } + public String getOuterAdapterKey() { + return outerAdapterKey; + } + + public void setOuterAdapterKey(String outerAdapterKey) { + this.outerAdapterKey = outerAdapterKey; + } + public String getDestination() { return destination; } @@ -67,15 +77,15 @@ public class ESSyncConfig { private String parent; private String sql; // 对象字段, 例: objFields: - // - _labels: array:; - private Map objFields = new LinkedHashMap<>(); + // - _labels: array:; + private Map objFields = new LinkedHashMap<>(); private List skips = new ArrayList<>(); private int commitBatch = 1000; private String etlCondition; - private boolean syncByTimestamp = false; // 是否按时间戳定时同步 - private Long syncInterval; // 同步时间间隔 + private boolean syncByTimestamp = false; // 是否按时间戳定时同步 + private Long syncInterval; // 同步时间间隔 - private SchemaItem schemaItem; // sql解析结果模型 + private SchemaItem schemaItem; // sql解析结果模型 public String get_index() { return _index; 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 81beede2..e83a96e6 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 @@ -1,25 +1,13 @@ package com.alibaba.otter.canal.client.adapter.es.config; -import java.io.File; -import java.io.FileInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.nio.charset.StandardCharsets; -import java.util.ArrayList; -import java.util.Collection; import java.util.LinkedHashMap; -import java.util.List; import java.util.Map; -import java.util.regex.Matcher; -import java.util.regex.Pattern; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.yaml.snakeyaml.Yaml; -import com.alibaba.druid.pool.DruidDataSource; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; -import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; +import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader; /** * ES 配置装载器 @@ -29,103 +17,25 @@ import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; */ public class ESSyncConfigLoader { - private static Logger logger = LoggerFactory - .getLogger(ESSyncConfigLoader.class); + private static Logger logger = LoggerFactory.getLogger(ESSyncConfigLoader.class); - private static final String BASE_PATH = "es"; - - private static volatile Map esSyncConfig = new LinkedHashMap<>(); // 文件名对应配置 - private static volatile Map> dbTableEsSyncConfig = new LinkedHashMap<>(); // schema-table对应配置 - - public static Map getEsSyncConfig() { - return esSyncConfig; - } - - public static Map> getDbTableEsSyncConfig() { - return dbTableEsSyncConfig; - } - - public static synchronized void load() { + public static synchronized Map load() { logger.info("## Start loading es mapping config ... "); - Collection configs = AdapterConfigs.get("es"); - if (configs == null) { - return; - } - for (String c : configs) { - if (c == null) { - continue; - } - c = c.trim(); - if (c.equals("") || c.startsWith("#")) { - continue; - } - ESSyncConfig config; - String configContent = null; - - if (c.endsWith(".yml")) { - configContent = readConfigContent(BASE_PATH + "/" + c); - } - - config = new Yaml().loadAs(configContent, ESSyncConfig.class); + Map esSyncConfig = new LinkedHashMap<>(); + Map configContentMap = MappingConfigsLoader.loadConfigs("es"); + configContentMap.forEach((fileName, content) -> { + ESSyncConfig config = new Yaml().loadAs(content, ESSyncConfig.class); try { config.validate(); - SchemaItem schemaItem = SqlParser.parse(config.getEsMapping().getSql()); - config.getEsMapping().setSchemaItem(schemaItem); - - DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); - if (dataSource == null || dataSource.getUrl() == null) { - throw new RuntimeException("No data source found: " + config.getDataSourceKey()); - } - Pattern pattern = Pattern.compile(".*:(.*)://.*/(.*)\\?.*$"); - Matcher matcher = pattern.matcher(dataSource.getUrl()); - if (!matcher.find()) { - throw new RuntimeException("Not found the schema of jdbc-url: " + config.getDataSourceKey()); - } - String schema = matcher.group(2); - - schemaItem.getAliasTableItems().values().forEach(tableItem -> { - List esSyncConfigs = dbTableEsSyncConfig - .computeIfAbsent(schema + "-" + tableItem.getTableName(), k -> new ArrayList<>()); - esSyncConfigs.add(config); - }); } catch (Exception e) { - throw new RuntimeException("ERROR Config: " + c, e); + throw new RuntimeException("ERROR Config: " + fileName + " " + e.getMessage(), e); } - esSyncConfig.put(c, config); - } + esSyncConfig.put(fileName, config); + }); logger.info("## ES mapping config loaded"); - } - - private static String readConfigContent(String config) { - InputStream in = null; - try { - // 先取本地文件，再取类路径 - File configFile = new File("../conf/" + config); - if (configFile.exists()) { - in = new FileInputStream(configFile); - } else { - in = ESSyncConfigLoader.class.getClassLoader().getResourceAsStream(config); - } - if (in == null) { - throw new RuntimeException("Config file: " + config + " not found."); - } - - byte[] bytes = new byte[in.available()]; - in.read(bytes); - return new String(bytes, StandardCharsets.UTF_8); - } catch (IOException e) { - throw new RuntimeException("Read es mapping config error ", e); - } finally { - try { - if (in != null) { - in.close(); - } - } catch (IOException e) { - // ignore - } - } + return esSyncConfig; } } diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java index 04d208f7..73b1a6bc 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java @@ -39,11 +39,8 @@ public class ESSyncService { this.esTemplate = esTemplate; } - public void sync(Dml dml) { + public void sync(List esSyncConfigs, Dml dml) { long begin = System.currentTimeMillis(); - String database = dml.getDatabase(); - String table = dml.getTable(); - List esSyncConfigs = ESSyncConfigLoader.getDbTableEsSyncConfig().get(database + "-" + table); if (esSyncConfigs != null) { if (logger.isTraceEnabled()) { logger.trace("Destination: {}, database:{}, table:{}, type:{}, effect index count: {}", diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java index 83f0cfb9..7ab197d1 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java @@ -1,6 +1,5 @@ package com.alibaba.otter.canal.client.adapter.es.test; -import java.util.List; import java.util.Map; import org.junit.Assert; @@ -9,22 +8,20 @@ import org.junit.Test; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; public class ConfigLoadTest { @Before public void before() { - AdapterConfigs.put("es", "mytest_user.yml"); + // AdapterConfigs.put("es", "mytest_user.yml"); // 加载数据源连接池 DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource); } @Test public void testLoad() { - ESSyncConfigLoader.load(); - Map configMap = ESSyncConfigLoader.getEsSyncConfig(); + Map configMap = ESSyncConfigLoader.load(); ESSyncConfig config = configMap.get("mytest_user.yml"); Assert.assertNotNull(config); Assert.assertEquals("defaultDS", config.getDataSourceKey()); @@ -34,7 +31,8 @@ public class ConfigLoadTest { Assert.assertEquals("id", esMapping.get_id()); Assert.assertNotNull(esMapping.getSql()); - Map> dbTableEsSyncConfig = ESSyncConfigLoader.getDbTableEsSyncConfig(); - Assert.assertFalse(dbTableEsSyncConfig.isEmpty()); + // Map> dbTableEsSyncConfig = + // ESSyncConfigLoader.getDbTableEsSyncConfig(); + // Assert.assertFalse(dbTableEsSyncConfig.isEmpty()); } } diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java index cb947de9..2d59f565 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java @@ -2,27 +2,25 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; import java.util.*; +import javax.sql.DataSource; + import org.elasticsearch.action.get.GetResponse; -import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; -import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.client.adapter.es.ESAdapter; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; -import javax.sql.DataSource; - public class LabelSyncJoinSub2Test { private ESAdapter esAdapter; @Before public void init() { - AdapterConfigs.put("es", "mytest_user_join_sub2.yml"); + // AdapterConfigs.put("es", "mytest_user_join_sub2.yml"); esAdapter = Common.init(); } @@ -32,9 +30,9 @@ public class LabelSyncJoinSub2Test { @Test public void test01() { DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS"); - Common.sqlExe(ds,"delete from label where id=1 or id=2"); - Common.sqlExe(ds,"insert into label (id,user_id,label) values (1,1,'a')"); - Common.sqlExe(ds,"insert into label (id,user_id,label) values (2,1,'b')"); + Common.sqlExe(ds, "delete from label where id=1 or id=2"); + Common.sqlExe(ds, "insert into label (id,user_id,label) values (1,1,'a')"); + Common.sqlExe(ds, "insert into label (id,user_id,label) values (2,1,'b')"); Dml dml = new Dml(); dml.setDestination("example"); @@ -46,12 +44,15 @@ public class LabelSyncJoinSub2Test { Map data = new LinkedHashMap<>(); dataList.add(data); data.put("id", 2L); - data.put("user_id",1L); + data.put("user_id", 1L); data.put("label", "b"); - dml.setData(dataList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("b;a_", response.getSource().get("_labels")); @@ -63,7 +64,7 @@ public class LabelSyncJoinSub2Test { @Test public void test02() { DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS"); - Common.sqlExe(ds,"update label set label='aa' where id=1"); + Common.sqlExe(ds, "update label set label='aa' where id=1"); Dml dml = new Dml(); dml.setDestination("example"); @@ -75,7 +76,7 @@ public class LabelSyncJoinSub2Test { Map data = new LinkedHashMap<>(); dataList.add(data); data.put("id", 1L); - data.put("user_id",1L); + data.put("user_id", 1L); data.put("label", "aa"); dml.setData(dataList); @@ -85,7 +86,11 @@ public class LabelSyncJoinSub2Test { old.put("label", "v"); dml.setOld(oldList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("b;aa_", response.getSource().get("_labels")); @@ -97,7 +102,7 @@ public class LabelSyncJoinSub2Test { @Test public void test03() { DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS"); - Common.sqlExe(ds,"delete from label where id=1"); + Common.sqlExe(ds, "delete from label where id=1"); Dml dml = new Dml(); dml.setDestination("example"); @@ -109,12 +114,15 @@ public class LabelSyncJoinSub2Test { Map data = new LinkedHashMap<>(); dataList.add(data); data.put("id", 1L); - data.put("user_id",1L); + data.put("user_id", 1L); data.put("label", "a"); - dml.setData(dataList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("b_", response.getSource().get("_labels")); diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java index bcf4d372..eafcfe75 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java @@ -2,27 +2,25 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; import java.util.*; +import javax.sql.DataSource; + import org.elasticsearch.action.get.GetResponse; -import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; -import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.client.adapter.es.ESAdapter; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; -import javax.sql.DataSource; - public class LabelSyncJoinSubTest { private ESAdapter esAdapter; @Before public void init() { - AdapterConfigs.put("es", "mytest_user_join_sub.yml"); + // AdapterConfigs.put("es", "mytest_user_join_sub.yml"); esAdapter = Common.init(); } @@ -32,9 +30,9 @@ public class LabelSyncJoinSubTest { @Test public void test01() { DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS"); - Common.sqlExe(ds,"delete from label where id=1 or id=2"); - Common.sqlExe(ds,"insert into label (id,user_id,label) values (1,1,'a')"); - Common.sqlExe(ds,"insert into label (id,user_id,label) values (2,1,'b')"); + Common.sqlExe(ds, "delete from label where id=1 or id=2"); + Common.sqlExe(ds, "insert into label (id,user_id,label) values (1,1,'a')"); + Common.sqlExe(ds, "insert into label (id,user_id,label) values (2,1,'b')"); Dml dml = new Dml(); dml.setDestination("example"); @@ -46,12 +44,15 @@ public class LabelSyncJoinSubTest { Map data = new LinkedHashMap<>(); dataList.add(data); data.put("id", 2L); - data.put("user_id",1L); + data.put("user_id", 1L); data.put("label", "b"); - dml.setData(dataList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("b;a", response.getSource().get("_labels")); @@ -63,7 +64,7 @@ public class LabelSyncJoinSubTest { @Test public void test02() { DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS"); - Common.sqlExe(ds,"update label set label='aa' where id=1"); + Common.sqlExe(ds, "update label set label='aa' where id=1"); Dml dml = new Dml(); dml.setDestination("example"); @@ -75,7 +76,7 @@ public class LabelSyncJoinSubTest { Map data = new LinkedHashMap<>(); dataList.add(data); data.put("id", 1L); - data.put("user_id",1L); + data.put("user_id", 1L); data.put("label", "aa"); dml.setData(dataList); @@ -85,7 +86,11 @@ public class LabelSyncJoinSubTest { old.put("label", "a"); dml.setOld(oldList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("b;aa", response.getSource().get("_labels")); @@ -97,7 +102,7 @@ public class LabelSyncJoinSubTest { @Test public void test03() { DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS"); - Common.sqlExe(ds,"delete from label where id=1"); + Common.sqlExe(ds, "delete from label where id=1"); Dml dml = new Dml(); dml.setDestination("example"); @@ -109,12 +114,15 @@ public class LabelSyncJoinSubTest { Map data = new LinkedHashMap<>(); dataList.add(data); data.put("id", 1L); - data.put("user_id",1L); + data.put("user_id", 1L); data.put("label", "a"); - dml.setData(dataList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("b", response.getSource().get("_labels")); diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java index a8fdf119..c87669ba 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java @@ -2,27 +2,25 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; import java.util.*; +import javax.sql.DataSource; + import org.elasticsearch.action.get.GetResponse; -import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; -import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.client.adapter.es.ESAdapter; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; -import javax.sql.DataSource; - public class RoleSyncJoinOne2Test { private ESAdapter esAdapter; @Before public void init() { - AdapterConfigs.put("es", "mytest_user_join_one2.yml"); + // AdapterConfigs.put("es", "mytest_user_join_one2.yml"); esAdapter = Common.init(); } @@ -32,8 +30,8 @@ public class RoleSyncJoinOne2Test { @Test public void test01() { DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS"); - Common.sqlExe(ds,"delete from role where id=1"); - Common.sqlExe(ds,"insert into role (id,role_name) values (1,'admin')"); + Common.sqlExe(ds, "delete from role where id=1"); + Common.sqlExe(ds, "insert into role (id,role_name) values (1,'admin')"); Dml dml = new Dml(); dml.setDestination("example"); @@ -46,10 +44,13 @@ public class RoleSyncJoinOne2Test { dataList.add(data); data.put("id", 1L); data.put("role_name", "admin"); - dml.setData(dataList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("admin_", response.getSource().get("_role_name")); @@ -61,7 +62,7 @@ public class RoleSyncJoinOne2Test { @Test public void test02() { DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS"); - Common.sqlExe(ds,"update role set role_name='admin3' where id=1"); + Common.sqlExe(ds, "update role set role_name='admin3' where id=1"); Dml dml = new Dml(); dml.setDestination("example"); @@ -82,7 +83,11 @@ public class RoleSyncJoinOne2Test { old.put("role_name", "admin"); dml.setOld(oldList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("admin3_", response.getSource().get("_role_name")); diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java index 9c5ab300..a7a61d5a 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java @@ -10,7 +10,7 @@ import org.junit.Before; import org.junit.Test; import com.alibaba.otter.canal.client.adapter.es.ESAdapter; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; @@ -20,7 +20,7 @@ public class RoleSyncJoinOneTest { @Before public void init() { - AdapterConfigs.put("es", "mytest_user_join_one.yml"); + // AdapterConfigs.put("es", "mytest_user_join_one.yml"); esAdapter = Common.init(); } @@ -47,7 +47,11 @@ public class RoleSyncJoinOneTest { dml.setData(dataList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("admin", response.getSource().get("_role_name")); @@ -80,7 +84,11 @@ public class RoleSyncJoinOneTest { old.put("role_name", "admin"); dml.setOld(oldList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("admin2", response.getSource().get("_role_name")); @@ -96,28 +104,32 @@ public class RoleSyncJoinOneTest { Common.sqlExe(ds, "insert into role (id,role_name) values (2,'operator')"); Common.sqlExe(ds, "update user set role_id=2 where id=1"); - Dml dml = new Dml(); - dml.setDestination("example"); - dml.setTs(new Date().getTime()); - dml.setType("UPDATE"); - dml.setDatabase("mytest"); - dml.setTable("user"); - List> dataList = new ArrayList<>(); - Map data = new LinkedHashMap<>(); - dataList.add(data); - data.put("id", 1L); - data.put("role_id", 2L); - dml.setData(dataList); - List> oldList = new ArrayList<>(); - Map old = new LinkedHashMap<>(); - oldList.add(old); - old.put("role_id", 1L); - dml.setOld(oldList); - esAdapter.getEsSyncService().sync(dml); + Dml dml = new Dml(); + dml.setDestination("example"); + dml.setTs(new Date().getTime()); + dml.setType("UPDATE"); + dml.setDatabase("mytest"); + dml.setTable("user"); + List> dataList = new ArrayList<>(); + Map data = new LinkedHashMap<>(); + dataList.add(data); + data.put("id", 1L); + data.put("role_id", 2L); + dml.setData(dataList); + List> oldList = new ArrayList<>(); + Map old = new LinkedHashMap<>(); + oldList.add(old); + old.put("role_id", 1L); + dml.setOld(oldList); - GetResponse response = - esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); - Assert.assertEquals("operator", response.getSource().get("_role_name")); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); + + GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + Assert.assertEquals("operator", response.getSource().get("_role_name")); Common.sqlExe(ds, "update user set role_id=1 where id=1"); @@ -138,7 +150,8 @@ public class RoleSyncJoinOneTest { oldList2.add(old2); old2.put("role_id", 2L); dml2.setOld(oldList2); - esAdapter.getEsSyncService().sync(dml2); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml2); GetResponse response2 = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("admin2", response2.getSource().get("_role_name")); @@ -166,7 +179,11 @@ public class RoleSyncJoinOneTest { dml.setData(dataList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertNull(response.getSource().get("_role_name")); diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java index d34d61b2..532cbc89 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java @@ -5,14 +5,12 @@ import java.util.*; import javax.sql.DataSource; import org.elasticsearch.action.get.GetResponse; -import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; -import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.client.adapter.es.ESAdapter; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; @@ -22,7 +20,7 @@ public class UserSyncJoinOneTest { @Before public void init() { - AdapterConfigs.put("es", "mytest_user_join_one.yml"); +// AdapterConfigs.put("es", "mytest_user_join_one.yml"); esAdapter = Common.init(); } @@ -32,8 +30,8 @@ public class UserSyncJoinOneTest { @Test public void test01() { DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS"); - Common.sqlExe(ds,"delete from user where id=1"); - Common.sqlExe(ds,"insert into user (id,name,role_id) values (1,'Eric',1)"); + Common.sqlExe(ds, "delete from user where id=1"); + Common.sqlExe(ds, "insert into user (id,name,role_id) values (1,'Eric',1)"); Dml dml = new Dml(); dml.setDestination("example"); @@ -48,10 +46,13 @@ public class UserSyncJoinOneTest { data.put("name", "Eric"); data.put("role_id", 1L); data.put("c_time", new Date()); - dml.setData(dataList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("Eric_", response.getSource().get("_name")); @@ -63,7 +64,7 @@ public class UserSyncJoinOneTest { @Test public void test02() { DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS"); - Common.sqlExe(ds,"update user set name='Eric2' where id=1"); + Common.sqlExe(ds, "update user set name='Eric2' where id=1"); Dml dml = new Dml(); dml.setDestination("example"); @@ -83,7 +84,11 @@ public class UserSyncJoinOneTest { old.put("name", "Eric"); dml.setOld(oldList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("Eric2_", response.getSource().get("_name")); diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java index 0b98a2b0..bcc3b7f6 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java @@ -3,15 +3,12 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; import java.util.*; import org.elasticsearch.action.get.GetResponse; -import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; -import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.client.adapter.es.ESAdapter; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; -import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; +import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; public class UserSyncSingleTest { @@ -20,7 +17,7 @@ public class UserSyncSingleTest { @Before public void init() { - AdapterConfigs.put("es", "mytest_user_single.yml"); + // AdapterConfigs.put("es", "mytest_user_single.yml"); esAdapter = Common.init(); } @@ -42,10 +39,13 @@ public class UserSyncSingleTest { data.put("name", "Eric"); data.put("role_id", 1L); data.put("c_time", new Date()); - dml.setData(dataList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("Eric", response.getSource().get("_name")); @@ -74,7 +74,11 @@ public class UserSyncSingleTest { old.put("name", "Eric"); dml.setOld(oldList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("Eric2", response.getSource().get("_name")); @@ -98,10 +102,13 @@ public class UserSyncSingleTest { data.put("name", "Eric"); data.put("role_id", 1L); data.put("c_time", new Date()); - dml.setData(dataList); - esAdapter.getEsSyncService().sync(dml); + String database = dml.getDatabase(); + String table = dml.getTable(); + List esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table); + + esAdapter.getEsSyncService().sync(esSyncConfigs, dml); GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertNull(response.getSource()); diff --git a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one.yml b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one.yml deleted file mode 100644 index 8e2f4efa..00000000 --- a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one.yml +++ /dev/null @@ -1,10 +0,0 @@ -dataSourceKey: defaultDS -destination: example -esMapping: - _index: mytest_user - _type: _doc - _id: _id - sql: "select a.id as _id, concat(a.name,'_') as _name, a.role_id as _role_id, - b.role_name as _role_name, a.c_time as _c_time from user a - left join role b on b.id=a.role_id" - commitBatch: 3000 \ No newline at end of file diff --git a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one2.yml b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one2.yml deleted file mode 100644 index 4a44c5e0..00000000 --- a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one2.yml +++ /dev/null @@ -1,10 +0,0 @@ -dataSourceKey: defaultDS -destination: example -esMapping: - _index: mytest_user - _type: _doc - _id: _id - sql: "select a.id as _id, concat(a.name,'_') as _name, a.role_id as _role_id, - concat(b.role_name,'_') as _role_name, a.c_time as _c_time from user a - left join role b on b.id=a.role_id" - commitBatch: 3000 \ No newline at end of file diff --git a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub.yml b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub.yml deleted file mode 100644 index 82c4e247..00000000 --- a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub.yml +++ /dev/null @@ -1,11 +0,0 @@ -dataSourceKey: defaultDS -destination: example -esMapping: - _index: mytest_user - _type: _doc - _id: _id - sql: "select a.id as _id, concat(a.name,'_') as _name, a.role_id as _role_id, - b.labels _labels, a.c_time as _c_time from user a - left join (select user_id, group_concat(label order by id desc separator ';') as labels from label - group by user_id) b on b.user_id=a.id" - commitBatch: 3000 \ No newline at end of file diff --git a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub2.yml b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub2.yml deleted file mode 100644 index c8c22c36..00000000 --- a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub2.yml +++ /dev/null @@ -1,11 +0,0 @@ -dataSourceKey: defaultDS -destination: example -esMapping: - _index: mytest_user - _type: _doc - _id: _id - sql: "select a.id as _id, concat(a.name,'_') as _name, a.role_id as _role_id, - concat(b.labels, '_') as _labels, a.c_time as _c_time from user a - left join (select user_id, group_concat(label order by id desc separator ';') as labels from label - group by user_id) b on b.user_id=a.id" - commitBatch: 3000 \ No newline at end of file diff --git a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_single.yml b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_single.yml_ similarity index 100% rename from client-adapter/elasticsearch/src/test/resources/es/mytest_user_single.yml rename to client-adapter/elasticsearch/src/test/resources/es/mytest_user_single.yml_ diff --git a/client-adapter/hbase/pom.xml b/client-adapter/hbase/pom.xml index 623d919b..10d08681 100644 --- a/client-adapter/hbase/pom.xml +++ b/client-adapter/hbase/pom.xml @@ -62,6 +62,26 @@ + + maven-antrun-plugin + + + package + + run + + + + + + + + + + + + + diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java index 8e4c6d36..a64ca3dc 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java @@ -35,31 +35,32 @@ import com.alibaba.otter.canal.client.adapter.support.*; @SPI("hbase") public class HbaseAdapter implements OuterAdapter { - private static Logger logger = LoggerFactory.getLogger(HbaseAdapter.class); + private static Logger logger = LoggerFactory.getLogger(HbaseAdapter.class); - private static volatile Map hbaseMapping = null; // 文件名对应配置 - private static volatile Map mappingConfigCache = null; // 库名-表名对应配置 + private Map hbaseMapping = new HashMap<>(); // 文件名对应配置 + private Map mappingConfigCache = new HashMap<>(); // 库名-表名对应配置 - private Connection conn; - private HbaseSyncService hbaseSyncService; - private HbaseTemplate hbaseTemplate; + private Connection conn; + private HbaseSyncService hbaseSyncService; + private HbaseTemplate hbaseTemplate; @Override public void init(OuterAdapterConfig configuration) { try { - if (mappingConfigCache == null) { - synchronized (MappingConfig.class) { - if (mappingConfigCache == null) { - hbaseMapping = MappingConfigLoader.load(); - mappingConfigCache = new HashMap<>(); - for (MappingConfig mappingConfig : hbaseMapping.values()) { - mappingConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." - + mappingConfig.getHbaseMapping().getDatabase() + "." - + mappingConfig.getHbaseMapping().getTable(), - mappingConfig); - } - } + Map hbaseMappingTmp = MappingConfigLoader.load(); + // 过滤不匹配的key的配置 + hbaseMappingTmp.forEach((key, mappingConfig) -> { + if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null) + || (mappingConfig.getOuterAdapterKey() != null + && mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) { + hbaseMapping.put(key, mappingConfig); } + }); + for (MappingConfig mappingConfig : hbaseMapping.values()) { + mappingConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." + + mappingConfig.getHbaseMapping().getDatabase() + "." + + mappingConfig.getHbaseMapping().getTable(), + mappingConfig); } Map properties = configuration.getProperties(); diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java index 326971ac..2d86adee 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java @@ -10,11 +10,13 @@ import java.util.*; */ public class MappingConfig { - private String dataSourceKey; // 数据源key + private String dataSourceKey; // 数据源key - private String destination; // canal实例或MQ的topic + private String outerAdapterKey; // adapter key - private HbaseMapping hbaseMapping; // hbase映射配置 + private String destination; // canal实例或MQ的topic + + private HbaseMapping hbaseMapping; // hbase映射配置 public String getDataSourceKey() { return dataSourceKey; @@ -24,6 +26,14 @@ public class MappingConfig { this.dataSourceKey = dataSourceKey; } + public String getOuterAdapterKey() { + return outerAdapterKey; + } + + public void setOuterAdapterKey(String outerAdapterKey) { + this.outerAdapterKey = outerAdapterKey; + } + public String getDestination() { return destination; } 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 2ac3a0d0..1f8c05c1 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 @@ -1,20 +1,13 @@ package com.alibaba.otter.canal.client.adapter.hbase.config; -import java.io.File; -import java.io.FileInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.nio.charset.StandardCharsets; -import java.util.Collection; import java.util.LinkedHashMap; import java.util.Map; -import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.yaml.snakeyaml.Yaml; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader; /** * HBase表映射配置加载器 @@ -24,9 +17,7 @@ import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; */ public class MappingConfigLoader { - private static Logger logger = LoggerFactory.getLogger(MappingConfigLoader.class); - - private static final String BASE_PATH = "hbase"; + private static Logger logger = LoggerFactory.getLogger(MappingConfigLoader.class); /** * 加载HBase表映射配置 @@ -38,116 +29,18 @@ public class MappingConfigLoader { Map result = new LinkedHashMap<>(); - Collection configs = AdapterConfigs.get("hbase"); - if (configs == null) { - return result; - } - for (String c : configs) { - if (c == null) { - continue; - } - c = c.trim(); - if (c.equals("") || c.startsWith("#")) { - continue; - } - - MappingConfig config; - String configContent = null; - - if (c.endsWith(".yml")) { - configContent = readConfigContent(BASE_PATH + "/" + c); - } - - // 简单配置database.table@datasourcekey?rowKey=key1,key2 - if (StringUtils.isEmpty(configContent)) { - String[] mapping = c.split("\\?"); - String params = mapping.length == 2 ? mapping[1] : null; - String rowKey = null; - String srcMeta = mapping[0]; - // - if (params != null) { - for (String entry : params.split("&")) { - if ("rowKey".equals(entry.split("=")[0])) { - rowKey = entry.split("=")[1]; - } - } - } - String dsKey = srcMeta.split("@").length == 2 ? srcMeta.split("@")[1] : null; - String[] dbTable; - if (dsKey == null) { - dbTable = srcMeta.split("\\."); - - } else { - dbTable = srcMeta.split("@")[0].split("\\."); - } - - if (dbTable.length == 2) { - config = new MappingConfig(); - - MappingConfig.HbaseMapping hbaseMapping = new MappingConfig.HbaseMapping(); - hbaseMapping.setHbaseTable(dbTable[0].toUpperCase() + "." + dbTable[1].toUpperCase()); - hbaseMapping.setAutoCreateTable(true); - hbaseMapping.setDatabase(dbTable[0]); - hbaseMapping.setTable(dbTable[1]); - hbaseMapping.setMode(MappingConfig.Mode.PHOENIX); - hbaseMapping.setRowKey(rowKey); - // 有定义rowKey - if (rowKey != null) { - MappingConfig.ColumnItem columnItem = new MappingConfig.ColumnItem(); - columnItem.setRowKey(true); - columnItem.setColumn(rowKey); - hbaseMapping.setRowKeyColumn(columnItem); - } - config.setHbaseMapping(hbaseMapping); - config.setDataSourceKey(dsKey); - - } else { - throw new RuntimeException(String.format("配置项[%s]内容为空, 或格式不符合database.table", c)); - } - - } else { // 配置文件配置 - config = new Yaml().loadAs(configContent, MappingConfig.class); - } - + Map configContentMap = MappingConfigsLoader.loadConfigs("hbase"); + configContentMap.forEach((fileName, content) -> { + MappingConfig config = new Yaml().loadAs(content, MappingConfig.class); try { config.validate(); } catch (Exception e) { - throw new RuntimeException("ERROR Config: " + c + " " + e.getMessage(), e); + throw new RuntimeException("ERROR load Config: " + fileName + " " + e.getMessage(), e); } - result.put(c, config); - } + result.put(fileName, config); + }); logger.info("## Hbase mapping config loaded"); return result; } - - public static String readConfigContent(String config) { - InputStream in = null; - try { - // 先取本地文件，再取类路径 - File configFile = new File("../conf/" + config); - if (configFile.exists()) { - in = new FileInputStream(configFile); - } else { - in = MappingConfigLoader.class.getClassLoader().getResourceAsStream(config); - } - if (in == null) { - throw new RuntimeException("Config file not found."); - } - - byte[] bytes = new byte[in.available()]; - in.read(bytes); - return new String(bytes, StandardCharsets.UTF_8); - } catch (IOException e) { - throw new RuntimeException("Read hbase mapping config error. ", e); - } finally { - try { - if (in != null) { - in.close(); - } - } catch (IOException e) { - // ignore - } - } - } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java index 2579e1f7..9e4737d6 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java @@ -1,9 +1,13 @@ package com.alibaba.otter.canal.adapter.launcher.config; +import java.sql.SQLException; import java.util.LinkedHashSet; import java.util.List; +import java.util.Map; import java.util.Set; +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; @@ -19,7 +23,9 @@ import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; @ConfigurationProperties(prefix = "canal.conf") public class AdapterCanalConfig extends CanalClientConfig { - public final Set DESTINATIONS = new LinkedHashSet<>(); + public final Set DESTINATIONS = new LinkedHashSet<>(); + + private Map srcDataSources; @Override public void setCanalInstances(List canalInstances) { @@ -48,4 +54,37 @@ public class AdapterCanalConfig extends CanalClientConfig { } } } + + public Map getSrcDataSources() { + return srcDataSources; + } + + public void setSrcDataSources(Map srcDataSources) { + this.srcDataSources = srcDataSources; + + if (srcDataSources != null) { + for (Map.Entry entry : srcDataSources.entrySet()) { + DatasourceConfig datasourceConfig = entry.getValue(); + // 加载数据源连接池 + DruidDataSource ds = new DruidDataSource(); + ds.setDriverClassName(datasourceConfig.getDriver()); + ds.setUrl(datasourceConfig.getUrl()); + ds.setUsername(datasourceConfig.getUsername()); + ds.setPassword(datasourceConfig.getPassword()); + ds.setInitialSize(1); + ds.setMinIdle(1); + ds.setMaxActive(datasourceConfig.getMaxActive()); + ds.setMaxWait(60000); + ds.setTimeBetweenEvictionRunsMillis(60000); + ds.setMinEvictableIdleTimeMillis(300000); + ds.setValidationQuery("select 1"); + try { + ds.init(); + } catch (SQLException e) { + throw new RuntimeException(e.getMessage(), e); + } + DatasourceConfig.DATA_SOURCES.put(entry.getKey(), ds); + } + } + } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java deleted file mode 100644 index 0f18c996..00000000 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java +++ /dev/null @@ -1,84 +0,0 @@ -package com.alibaba.otter.canal.adapter.launcher.config; - -import java.sql.SQLException; -import java.util.List; -import java.util.Map; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.stereotype.Component; - -import com.alibaba.druid.pool.DruidDataSource; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; -import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; - -/** - * 适配器数据源及配置文件列表配置类 - * - * @author rewerma @ 2018-10-20 - * @version 1.0.0 - */ -@Component -@ConfigurationProperties(prefix = "adapter.conf") -public class AdapterConfig { - - private static Logger logger = LoggerFactory.getLogger(AdapterConfig.class); - - private Map datasourceConfigs; - - private List adapterConfigs; - - public List getAdapterConfigs() { - return adapterConfigs; - } - - public Map getDatasourceConfigs() { - return datasourceConfigs; - } - - public void setDatasourceConfigs(Map datasourceConfigs) { - this.datasourceConfigs = datasourceConfigs; - - if (datasourceConfigs != null) { - for (Map.Entry entry : datasourceConfigs.entrySet()) { - DatasourceConfig datasourceConfig = entry.getValue(); - // 加载数据源连接池 - DruidDataSource ds = new DruidDataSource(); - ds.setDriverClassName(datasourceConfig.getDriver()); - ds.setUrl(datasourceConfig.getUrl()); - ds.setUsername(datasourceConfig.getUsername()); - ds.setPassword(datasourceConfig.getPassword()); - ds.setInitialSize(1); - ds.setMinIdle(1); - ds.setMaxActive(datasourceConfig.getMaxActive()); - ds.setMaxWait(60000); - ds.setTimeBetweenEvictionRunsMillis(60000); - ds.setMinEvictableIdleTimeMillis(300000); - ds.setValidationQuery("select 1"); - try { - ds.init(); - } catch (SQLException e) { - logger.error("ERROR ## failed to initial datasource: " + datasourceConfig.getUrl(), e); - } - DatasourceConfig.DATA_SOURCES.put(entry.getKey(), ds); - } - } - } - - public void setAdapterConfigs(List adapterConfigs) { - this.adapterConfigs = adapterConfigs; - - if (adapterConfigs != null) { - AdapterConfigs.clear(); - for (String adapterConfig : adapterConfigs) { - int idx = adapterConfig.indexOf("/"); - if (idx > -1) { - String type = adapterConfig.substring(0, idx); - String ymlFile = adapterConfig.substring(idx + 1); - AdapterConfigs.put(type, ymlFile); - } - } - } - } -} 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 e3197e7d..77567a87 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 @@ -33,6 +33,8 @@ public class CanalAdapterLoader { private Map canalMQWorker = new HashMap<>(); + private Map outerAdapters = new HashMap<>(); // 配置文件对应adapter + private ExtensionLoader loader; public CanalAdapterLoader(CanalClientConfig canalClientConfig){ @@ -53,8 +55,8 @@ public class CanalAdapterLoader { } String zkHosts = this.canalClientConfig.getZookeeperHosts(); - // 初始化canal-client的适配器 if (canalClientConfig.getCanalInstances() != null) { + // 初始化canal-client的适配器 for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) { List> canalOuterAdapterGroups = new ArrayList<>(); @@ -83,10 +85,8 @@ public class CanalAdapterLoader { worker.start(); logger.info("Start adapter for canal instance: {} succeed", instance.getInstance()); } - } - - // 初始化canal-client-mq的适配器 - if (canalClientConfig.getMqTopics() != null) { + } else if (canalClientConfig.getMqTopics() != null) { + // 初始化canal-client-mq的适配器 for (CanalClientConfig.MQTopic topic : canalClientConfig.getMqTopics()) { for (CanalClientConfig.MQGroup group : topic.getGroups()) { List> canalOuterAdapterGroups = new ArrayList<>(); @@ -124,11 +124,11 @@ public class CanalAdapterLoader { private void loadConnector(OuterAdapterConfig config, List canalOutConnectors) { try { OuterAdapter adapter; - if ("rdb".equalsIgnoreCase(config.getName())) { - adapter = loader.getExtension(config.getName(), config.getKey()); - } else { - adapter = loader.getExtension(config.getName()); - } + // if ("rdb".equalsIgnoreCase(config.getName())) { + adapter = loader.getExtension(config.getName(), StringUtils.trimToEmpty(config.getKey())); + // } else { + // adapter = loader.getExtension(config.getName()); + // } ClassLoader cl = Thread.currentThread().getContextClassLoader(); // 替换ClassLoader Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader()); diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java index 2e535535..15dafa79 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java @@ -12,7 +12,6 @@ import org.springframework.stereotype.Component; import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.adapter.launcher.config.AdapterCanalConfig; -import com.alibaba.otter.canal.adapter.launcher.config.AdapterConfig; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; /** @@ -33,8 +32,6 @@ public class CanalAdapterService { // 注入bean保证优先注册 @Resource - private AdapterConfig adapterConfig; - @Resource private SpringContext springContext; @Resource private SyncSwitch syncSwitch; diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java index 0a6d00f4..302206be 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java @@ -46,17 +46,17 @@ public class CommonRest { } /** - * ETL curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST - * + * ETL curl http://127.0.0.1:8081/etl/rdb/oracle1/mytest_user.yml -X POST + * * @param type 类型 hbase, es - * @param task 任务名对应配置文件名 mytest_person2.yml + * @param key adapter key + * @param task 任务名对应配置文件名 mytest_user.yml * @param params etl where条件参数, 为空全部导入 - * @return */ - @PostMapping("/etl/{type}/{task}") - public EtlResult etl(@PathVariable String type, @PathVariable String task, + @PostMapping("/etl/{type}/{key}/{task}") + public EtlResult etl(@PathVariable String type, @PathVariable String key, @PathVariable String task, @RequestParam(name = "params", required = false) String params) { - OuterAdapter adapter = loader.getExtension(type); + OuterAdapter adapter = loader.getExtension(type, key); String destination = adapter.getDestination(task); String lockKey = destination == null ? task : destination; @@ -83,12 +83,11 @@ public class CommonRest { } } try { - List paramArr = null; + List paramArray = null; if (params != null) { - String[] parmaArray = params.trim().split(";"); - paramArr = Arrays.asList(parmaArray); + paramArray = Arrays.asList(params.trim().split(";")); } - return adapter.etl(task, paramArr); + return adapter.etl(task, paramArray); } finally { if (destination != null && oriSwitchStatus != null && oriSwitchStatus) { syncSwitch.on(destination); @@ -101,6 +100,33 @@ public class CommonRest { } } + /** + * ETL curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST + * + * @param type 类型 hbase, es + * @param task 任务名对应配置文件名 mytest_person2.yml + * @param params etl where条件参数, 为空全部导入 + */ + @PostMapping("/etl/{type}/{task}") + public EtlResult etl(@PathVariable String type, @PathVariable String task, + @RequestParam(name = "params", required = false) String params) { + return etl(type, null, task, params); + } + + /** + * 统计总数 curl http://127.0.0.1:8081/count/rdb/oracle1/mytest_user.yml + * + * @param type 类型 hbase, es + * @param key adapter key + * @param task 任务名对应配置文件名 mytest_person2.yml + * @return + */ + @GetMapping("/count/{type}/{key}/{task}") + public Map count(@PathVariable String type, @PathVariable String key, @PathVariable String task) { + OuterAdapter adapter = loader.getExtension(type, key); + return adapter.count(task); + } + /** * 统计总数 curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml * @@ -110,8 +136,7 @@ public class CommonRest { */ @GetMapping("/count/{type}/{task}") public Map count(@PathVariable String type, @PathVariable String task) { - OuterAdapter adapter = loader.getExtension(type); - return adapter.count(task); + return count(type, null, task); } /** diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 2c48fa84..61866595 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -16,6 +16,11 @@ canal.conf: # zookeeperHosts: slave1:2181 # bootstrapServers: slave1:6667 #or rocketmq # flatMessage: true +# srcDataSources: +# defaultDS: +# url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true +# username: root +# password: 121212 canalInstances: - instance: example groups: @@ -25,7 +30,7 @@ canal.conf: # key: oracle1 # properties: # jdbc.driverClassName: oracle.jdbc.OracleDriver -# jdbc.url: jdbc:oracle:thin:@127.0.0.1:1521:orcl +# jdbc.url: jdbc:oracle:thin:@localhost:49161:XE # jdbc.username: mytest # jdbc.password: m121212 # - name: rdb @@ -45,27 +50,9 @@ canal.conf: # properties: # cluster.name: elasticsearch # mqTopics: -# - mqMode: kafka +# - mqMode: kafka # or rocketmq # topic: example # groups: # - groupId: g2 # outAdapters: -# - name: logger -# mqTopics: -# - mqMode: rocketmq -# topic: example -# groups: -# - groupId: g2 -# outAdapters: -# - name: logger - -#adapter.conf: -# datasourceConfigs: -# defaultDS: -# url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true -# username: root -# password: 121212 -# adapterConfigs: -# - hbase/mytest_person2.yml -# - es/mytest_user.yml -# - rdb/mytest_user.yml +# - name: logger \ No newline at end of file diff --git a/client-adapter/rdb/README.md b/client-adapter/rdb/README.md new file mode 100644 index 00000000..b579b671 --- /dev/null +++ b/client-adapter/rdb/README.md @@ -0,0 +1,67 @@ +## 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/pom.xml b/client-adapter/rdb/pom.xml index 994d28aa..120aa5e1 100644 --- a/client-adapter/rdb/pom.xml +++ b/client-adapter/rdb/pom.xml @@ -80,6 +80,26 @@ + + maven-antrun-plugin + + + package + + run + + + + + + + + + + + + + \ 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 c07dee6b..85f2f62c 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java @@ -1,9 +1,11 @@ package com.alibaba.otter.canal.client.adapter.rdb; import java.sql.Connection; -import java.sql.ResultSetMetaData; import java.sql.SQLException; -import java.util.*; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import javax.sql.DataSource; @@ -13,8 +15,8 @@ import org.slf4j.LoggerFactory; import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.client.adapter.OuterAdapter; +import com.alibaba.otter.canal.client.adapter.rdb.config.ConfigLoader; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; -import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfigLoader; import com.alibaba.otter.canal.client.adapter.rdb.service.RdbEtlService; import com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService; import com.alibaba.otter.canal.client.adapter.support.*; @@ -22,26 +24,26 @@ 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 volatile Map rdbMapping = new HashMap<>(); // 文件名对应配置 - private Map mappingConfigCache; // 库名-表名对应配置 + private Map rdbMapping = new HashMap<>(); // 文件名对应配置 + private Map mappingConfigCache = new HashMap<>(); // 库名-表名对应配置 - private DruidDataSource dataSource; + private DruidDataSource dataSource; - private RdbSyncService rdbSyncService; + private RdbSyncService rdbSyncService; @Override public void init(OuterAdapterConfig configuration) { - SPI spi = this.getClass().getAnnotation(SPI.class); - Map rdbMappingTmp = MappingConfigLoader.load(spi.value()); - // 过滤其他key的配置 + Map rdbMappingTmp = ConfigLoader.load(); + // 过滤不匹配的key的配置 rdbMappingTmp.forEach((key, mappingConfig) -> { - if (mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey())) { + if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null) + || (mappingConfig.getOuterAdapterKey() != null + && mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) { rdbMapping.put(key, mappingConfig); } }); - mappingConfigCache = new HashMap<>(); for (MappingConfig mappingConfig : rdbMapping.values()) { mappingConfigCache .put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." @@ -68,55 +70,6 @@ public class RdbAdapter implements OuterAdapter { logger.error("ERROR ## failed to initial datasource: " + properties.get("jdbc.url"), e); } - rdbMapping.values().forEach(config -> { - try { - MappingConfig.DbMapping dbMapping = config.getDbMapping(); - // 从源表加载所有字段名 - if (dbMapping.getAllColumns() == null) { - synchronized (RdbSyncService.class) { - if (dbMapping.getAllColumns() == null) { - DataSource srcDS = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); - Connection srcConn = srcDS.getConnection(); - String srcMetaSql = "SELECT * FROM " + dbMapping.getDatabase() + "." + dbMapping.getTable() - + " WHERE 1=2 "; - List srcColumns = new ArrayList<>(); - Util.sqlRS(srcConn, srcMetaSql, rs -> { - try { - ResultSetMetaData rmd = rs.getMetaData(); - int cnt = rmd.getColumnCount(); - for (int i = 1; i <= cnt; i++) { - srcColumns.add(rmd.getColumnName(i).toLowerCase()); - } - } catch (SQLException e) { - logger.error(e.getMessage(), e); - } - }); - Map columnsMap = new LinkedHashMap<>(); - - for (String srcColumn : srcColumns) { - String targetColumn = srcColumn; - if (dbMapping.getTargetColumns() != null) { - for (Map.Entry entry : dbMapping.getTargetColumns().entrySet()) { - String targetColumnName = entry.getKey(); - String srcColumnName = entry.getValue(); - - if (srcColumnName != null - && srcColumnName.toLowerCase().equals(srcColumn.toUpperCase())) { - targetColumn = targetColumnName; - } - } - } - columnsMap.put(targetColumn, srcColumn); - } - dbMapping.setAllColumns(columnsMap); - } - } - } - } catch (SQLException e) { - logger.error(e.getMessage(), e); - } - }); - rdbSyncService = new RdbSyncService(dataSource); } @@ -213,6 +166,15 @@ public class RdbAdapter implements OuterAdapter { return res; } + @Override + public String getDestination(String task) { + MappingConfig config = rdbMapping.get(task); + if (config != null) { + return config.getDestination(); + } + return null; + } + @Override public void destroy() { if (dataSource != null) { 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 new file mode 100644 index 00000000..c2fe3339 --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java @@ -0,0 +1,46 @@ +package com.alibaba.otter.canal.client.adapter.rdb.config; + +import java.util.LinkedHashMap; +import java.util.Map; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.yaml.snakeyaml.Yaml; + +import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader; + +/** + * RDB表映射配置加载器 + * + * @author rewerma 2018-11-07 下午02:41:34 + * @version 1.0.0 + */ +public class ConfigLoader { + + private static Logger logger = LoggerFactory.getLogger(ConfigLoader.class); + + /** + * 加载HBase表映射配置 + * + * @return 配置名/配置文件名--对象 + */ + public static Map load() { + logger.info("## Start loading rdb mapping config ... "); + + Map result = new LinkedHashMap<>(); + + Map configContentMap = MappingConfigsLoader.loadConfigs("rdb"); + configContentMap.forEach((fileName, content) -> { + MappingConfig config = new Yaml().loadAs(content, MappingConfig.class); + try { + config.validate(); + } catch (Exception e) { + throw new RuntimeException("ERROR Config: " + fileName + " " + e.getMessage(), e); + } + result.put(fileName, config); + }); + + logger.info("## Rdb mapping config loaded"); + return result; + } +} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java index 95623ea4..40389f9a 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java @@ -79,7 +79,7 @@ public class MappingConfig { private int readBatch = 5000; private int commitBatch = 5000; // etl等批量提交大小 - private volatile Map allColumns; // mapAll为true,自动设置改字段 +// private volatile Map allColumns; // mapAll为true,自动设置改字段 public String getDatabase() { return database; @@ -161,12 +161,12 @@ public class MappingConfig { this.commitBatch = commitBatch; } - public Map getAllColumns() { - return allColumns; - } - - public void setAllColumns(Map allColumns) { - this.allColumns = allColumns; - } +// public Map getAllColumns() { +// return allColumns; +// } +// +// public void setAllColumns(Map allColumns) { +// this.allColumns = allColumns; +// } } } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java deleted file mode 100644 index aefb467f..00000000 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java +++ /dev/null @@ -1,100 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.rdb.config; - -import java.io.File; -import java.io.FileInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.nio.charset.StandardCharsets; -import java.util.Collection; -import java.util.LinkedHashMap; -import java.util.Map; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.yaml.snakeyaml.Yaml; - -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; - -/** - * RDB表映射配置加载器 - * - * @author rewerma 2018-11-07 下午02:41:34 - * @version 1.0.0 - */ -public class MappingConfigLoader { - - private static Logger logger = LoggerFactory.getLogger(MappingConfigLoader.class); - - /** - * 加载HBase表映射配置 - * - * @return 配置名/配置文件名--对象 - */ - public static Map load(String name) { - logger.info("## Start loading rdb mapping config ... "); - - Map result = new LinkedHashMap<>(); - - Collection configs = AdapterConfigs.get(name); - if (configs == null) { - return result; - } - for (String c : configs) { - if (c == null) { - continue; - } - c = c.trim(); - if (c.equals("") || c.startsWith("#")) { - continue; - } - - String configContent = null; - - if (c.endsWith(".yml")) { - configContent = readConfigContent(name + "/" + c); - } - - MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class); - - try { - config.validate(); - } catch (Exception e) { - throw new RuntimeException("ERROR Config: " + c + " " + e.getMessage(), e); - } - result.put(c, config); - } - - logger.info("## Rdb mapping config loaded"); - return result; - } - - public static String readConfigContent(String config) { - InputStream in = null; - try { - // 先取本地文件，再取类路径 - File configFile = new File("../conf/" + config); - if (configFile.exists()) { - in = new FileInputStream(configFile); - } else { - in = MappingConfigLoader.class.getClassLoader().getResourceAsStream(config); - } - if (in == null) { - throw new RuntimeException("Rdb mapping config file not found."); - } - - byte[] bytes = new byte[in.available()]; - in.read(bytes); - return new String(bytes, StandardCharsets.UTF_8); - } catch (IOException e) { - throw new RuntimeException("Read rdb mapping config error. ", e); - } finally { - try { - if (in != null) { - in.close(); - } - } catch (IOException e) { - // ignore - } - } - } -} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java index 321c9f21..3cf4976a 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java @@ -13,6 +13,7 @@ import java.util.concurrent.atomic.AtomicLong; import javax.sql.DataSource; +import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -91,8 +92,8 @@ public class RdbEtlService { } else { sqlFinal = sql + " LIMIT " + offset + "," + cnt; } - Future future = executor.submit( - () -> executeSqlImport(params, srcDS, targetDS, sqlFinal, dbMapping, successCount, errMsg)); + Future future = executor + .submit(() -> executeSqlImport(srcDS, targetDS, sqlFinal, dbMapping, successCount, errMsg)); futures.add(future); } @@ -102,7 +103,7 @@ public class RdbEtlService { executor.shutdown(); } else { - executeSqlImport(params, srcDS, targetDS, sql.toString(), dbMapping, successCount, errMsg); + executeSqlImport(srcDS, targetDS, sql.toString(), dbMapping, successCount, errMsg); } logger.info( @@ -160,8 +161,8 @@ public class RdbEtlService { /** * 执行导入 */ - private static boolean executeSqlImport(List params, DataSource srcDS, DataSource targetDS, String sql, - DbMapping dbMapping, AtomicLong successCount, List errMsg) { + private static boolean executeSqlImport(DataSource srcDS, DataSource targetDS, String sql, DbMapping dbMapping, + AtomicLong successCount, List errMsg) { try { Util.sqlRS(srcDS, sql, rs -> { int idx = 1; @@ -172,16 +173,18 @@ public class RdbEtlService { Map columnType = new LinkedHashMap<>(); ResultSetMetaData rsd = rs.getMetaData(); int columnCount = rsd.getColumnCount(); + List columns = new ArrayList<>(); for (int i = 1; i <= columnCount; i++) { columnType.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i)); + columns.add(rsd.getColumnName(i)); } - Map columnsMap; - if (dbMapping.isMapAll()) { - columnsMap = dbMapping.getAllColumns(); - } else { - columnsMap = dbMapping.getTargetColumns(); - } + Map columnsMap = SyncUtil.getColumnsMap(dbMapping, columns); + // if (dbMapping.isMapAll()) { + // columnsMap = dbMapping.getAllColumns(); + // } else { + // columnsMap = dbMapping.getTargetColumns(); + // } StringBuilder insertSql = new StringBuilder(); insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" ("); diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index ab46bf4b..def43637 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java @@ -10,11 +10,9 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import java.util.function.Consumer; import javax.sql.DataSource; -import com.alibaba.otter.canal.client.adapter.support.Util; import org.joda.time.DateTime; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -23,8 +21,9 @@ import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig.DbMapping; -import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; +import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil; import com.alibaba.otter.canal.client.adapter.support.Dml; +import com.alibaba.otter.canal.client.adapter.support.Util; /** * RDB同步操作业务 @@ -84,12 +83,7 @@ public class RdbSyncService { Connection conn = dataSource.getConnection(); conn.setAutoCommit(false); try { - Map columnsMap; - if (dbMapping.isMapAll()) { - columnsMap = dbMapping.getAllColumns(); - } else { - columnsMap = dbMapping.getTargetColumns(); - } + Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data.get(0)); StringBuilder insertSql = new StringBuilder(); insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" ("); @@ -148,6 +142,7 @@ public class RdbSyncService { conn.commit(); } } catch (Exception e) { + logger.error(e.getMessage(), e); conn.rollback(); } finally { conn.close(); @@ -180,12 +175,7 @@ public class RdbSyncService { conn.setAutoCommit(false); try { - Map columnsMap; - if (dbMapping.isMapAll()) { - columnsMap = dbMapping.getAllColumns(); - } else { - columnsMap = dbMapping.getTargetColumns(); - } + Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data.get(0)); Map ctype = getTargetColumnType(conn, config); @@ -230,6 +220,7 @@ public class RdbSyncService { conn.commit(); } } catch (Exception e) { + logger.error(e.getMessage(), e); conn.rollback(); } finally { conn.close(); @@ -283,14 +274,13 @@ public class RdbSyncService { conn.commit(); } } catch (Exception e) { + logger.error(e.getMessage(), e); conn.rollback(); } finally { conn.close(); } } - - /** * 获取目标字段类型 * @@ -309,7 +299,7 @@ public class RdbSyncService { columnType = new LinkedHashMap<>(); final Map columnTypeTmp = columnType; String sql = "SELECT * FROM " + dbMapping.getTargetTable() + " WHERE 1=2"; - Util.sqlRS(conn, sql, rs -> { + Util.sqlRS(conn, sql, rs -> { try { ResultSetMetaData rsd = rs.getMetaData(); int columnCount = rsd.getColumnCount(); @@ -524,4 +514,5 @@ public class RdbSyncService { logger.error(e.getMessage(), e); } } + } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java new file mode 100644 index 00000000..5e2c38bb --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java @@ -0,0 +1,39 @@ +package com.alibaba.otter.canal.client.adapter.rdb.support; + +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.Map; + +import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; + +public class SyncUtil { + + public static Map getColumnsMap(MappingConfig.DbMapping dbMapping, Map data) { + return getColumnsMap(dbMapping, data.keySet()); + } + + public static Map getColumnsMap(MappingConfig.DbMapping dbMapping, Collection columns) { + Map columnsMap; + if (dbMapping.isMapAll()) { + columnsMap = new LinkedHashMap<>(); + for (String srcColumn : columns) { + boolean flag = true; + if (dbMapping.getTargetColumns() != null) { + for (Map.Entry entry : dbMapping.getTargetColumns().entrySet()) { + if (srcColumn.equals(entry.getValue())) { + columnsMap.put(entry.getKey(), srcColumn); + flag = false; + break; + } + } + } + if (flag) { + columnsMap.put(srcColumn, srcColumn); + } + } + } else { + columnsMap = dbMapping.getTargetColumns(); + } + return columnsMap; + } +} diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java index dea6cee7..5eeb0878 100644 --- a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java @@ -6,23 +6,21 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; +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.MappingConfigLoader; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; public class ConfigLoadTest { @Before public void before() { - AdapterConfigs.put("oracle", "mytest_user.yml"); // 加载数据源连接池 DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource); } @Test public void testLoad() { - Map configMap = MappingConfigLoader.load("oracle"); + Map configMap = ConfigLoader.load(); Assert.assertFalse(configMap.isEmpty()); } diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java index 4ae04643..5af9fdcb 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 @@ -14,7 +14,7 @@ public class Common { OuterAdapterConfig outerAdapterConfig = new OuterAdapterConfig(); outerAdapterConfig.setName("rdb"); - outerAdapterConfig.setKey("oralce1"); + //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"); 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 5b8a45e3..a00a9ce5 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 @@ -6,7 +6,6 @@ import org.junit.Before; import org.junit.Test; import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; import com.alibaba.otter.canal.client.adapter.support.Dml; public class OracleSyncTest { @@ -15,7 +14,6 @@ public class OracleSyncTest { @Before public void init() { - AdapterConfigs.put("rdb", "mytest_user.yml"); rdbAdapter = Common.init(); } diff --git a/client-adapter/rdb/src/test/resources/oracle/mytest_user.yml b/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml similarity index 100% rename from client-adapter/rdb/src/test/resources/oracle/mytest_user.yml rename to client-adapter/rdb/src/test/resources/rdb/mytest_user.yml