wyl/canal
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

简化adapter配置

Browse Source
This commit is contained in:
mcy
2018-11-12 17:30:40 +08:00
parent 66912d3d05
commit 2e5b7b89af
33 changed files with 260 additions and 536 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java
-41
View File
@@ -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<String, Set<String>> configs = new HashMap<>();
public static void put(String key, String value) {
Set<String> values = configs.get(key);
if (values == null) {
values = new LinkedHashSet<>();
}
values.add(value);
configs.put(key, values);
}
public static Set<String> get(String key) {
return configs.get(key);
}
public static void clear() {
configs.clear();
}
}
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java
+46
View File
@@ -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<String, String> loadConfigs(String name) {
Map<String, String> 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;
}
}
client-adapter/elasticsearch/pom.xml
+20
View File
@@ -71,6 +71,26 @@
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-antrun-plugin</artifactId>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>run</goal>
</goals>
<configuration>
<tasks>
<copy todir="${project.basedir}/../launcher/target/classes/es" overwrite="true" >
<fileset dir="${project.basedir}/target/classes/es" erroronmissingdir="true">
<include name="*.yml"/>
</fileset>
</copy>
</tasks>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
+3 -3
View File
@@ -63,12 +63,12 @@ public class ESAdapter implements OuterAdapter {
@Override
public void init(OuterAdapterConfig configuration) {
try {
SPI spi = this.getClass().getAnnotation(SPI.class);
Map<String, ESSyncConfig> esSyncConfigTmp = ESSyncConfigLoader.load(spi.value());
Map<String, ESSyncConfig> esSyncConfigTmp = ESSyncConfigLoader.load();
// 过滤不匹配的key的配置
esSyncConfigTmp.forEach((key, config) -> {
if ((config.getOuterAdapterKey() == null && configuration.getKey() == null)
|| config.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey())) {
|| (config.getOuterAdapterKey() != null
&& config.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
esSyncConfig.put(key, config);
}
});
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java
+9 -71
View File
@@ -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,75 +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);
public static synchronized Map<String, ESSyncConfig> load(String name) {
public static synchronized Map<String, ESSyncConfig> load() {
logger.info("## Start loading es mapping config ... ");
Map<String, ESSyncConfig> esSyncConfig = new LinkedHashMap<>();
Collection<String> configs = AdapterConfigs.get("es");
if (configs == null) {
return esSyncConfig;
}
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(name + "/" + c);
}
config = new Yaml().loadAs(configContent, ESSyncConfig.class);
Map<String, String> configContentMap = MappingConfigsLoader.loadConfigs("es");
configContentMap.forEach((fileName, content) -> {
ESSyncConfig config = new Yaml().loadAs(content, ESSyncConfig.class);
try {
config.validate();
} 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");
return esSyncConfig;
}
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
}
}
}
}
client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java
+2 -4
View File
@@ -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,21 +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() {
Map<String, ESSyncConfig> configMap = ESSyncConfigLoader.load("es");
Map<String, ESSyncConfig> configMap = ESSyncConfigLoader.load();
ESSyncConfig config = configMap.get("mytest_user.yml");
Assert.assertNotNull(config);
Assert.assertEquals("defaultDS", config.getDataSourceKey());
client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java
+4 -7
View File
@@ -2,28 +2,25 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync;
import java.util.*;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
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();
}
client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java
+4 -7
View File
@@ -2,28 +2,25 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync;
import java.util.*;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
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();
}
client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java
+4 -7
View File
@@ -2,28 +2,25 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync;
import java.util.*;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
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();
}
client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java
+2 -3
View File
@@ -4,14 +4,13 @@ import java.util.*;
import javax.sql.DataSource;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
import org.elasticsearch.action.get.GetResponse;
import org.junit.Assert;
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;
@@ -21,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();
}
client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java
+2 -5
View File
@@ -4,16 +4,13 @@ import java.util.*;
import javax.sql.DataSource;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
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;
@@ -23,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();
}
client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java
+2 -6
View File
@@ -2,17 +2,13 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync;
import java.util.*;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
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 {
@@ -21,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();
}
client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one.yml
-10
View File
@@ -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
client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one2.yml
-10
View File
@@ -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
client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub.yml
-11
View File
@@ -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
client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub2.yml
-11
View File
@@ -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
client-adapter/elasticsearch/src/test/resources/es/mytest_user_single.yml → client-adapter/elasticsearch/src/test/resources/es/mytest_user_single.yml_
View File
client-adapter/hbase/pom.xml
+20
View File
@@ -62,6 +62,26 @@
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-antrun-plugin</artifactId>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>run</goal>
</goals>
<configuration>
<tasks>
<copy todir="${project.basedir}/../launcher/target/classes/hbase" overwrite="true" >
<fileset dir="${project.basedir}/target/classes/hbase" erroronmissingdir="true">
<include name="*.yml"/>
</fileset>
</copy>
</tasks>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java
+3 -3
View File
@@ -47,12 +47,12 @@ public class HbaseAdapter implements OuterAdapter {
@Override
public void init(OuterAdapterConfig configuration) {
try {
SPI spi = this.getClass().getAnnotation(SPI.class);
Map<String, MappingConfig> hbaseMappingTmp = MappingConfigLoader.load(spi.value());
Map<String, MappingConfig> hbaseMappingTmp = MappingConfigLoader.load();
// 过滤不匹配的key的配置
hbaseMappingTmp.forEach((key, mappingConfig) -> {
if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null)
|| mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey())) {
|| (mappingConfig.getOuterAdapterKey() != null
&& mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
hbaseMapping.put(key, mappingConfig);
}
});
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java
+9 -114
View File
@@ -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,128 +17,30 @@ import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
*/
public class MappingConfigLoader {
private static Logger logger = LoggerFactory.getLogger(MappingConfigLoader.class);
private static Logger logger = LoggerFactory.getLogger(MappingConfigLoader.class);
/**
* 加载HBase表映射配置
*
* @return 配置名/配置文件名--对象
*/
public static Map<String, MappingConfig> load(String name) {
public static Map<String, MappingConfig> load() {
logger.info("## Start loading hbase mapping config ... ");
Map<String, MappingConfig> result = new LinkedHashMap<>();
Collection<String> 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;
}
MappingConfig config;
String configContent = null;
if (c.endsWith(".yml")) {
configContent = readConfigContent(name + "/" + 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<String, String> 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
}
}
}
}
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java
+40 -1
View File
@@ -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<String> DESTINATIONS = new LinkedHashSet<>();
public final Set<String> DESTINATIONS = new LinkedHashSet<>();
private Map<String, DatasourceConfig> srcDataSources;
@Override
public void setCanalInstances(List<CanalInstance> canalInstances) {
@@ -48,4 +54,37 @@ public class AdapterCanalConfig extends CanalClientConfig {
}
}
}
public Map<String, DatasourceConfig> getSrcDataSources() {
return srcDataSources;
}
public void setSrcDataSources(Map<String, DatasourceConfig> srcDataSources) {
this.srcDataSources = srcDataSources;
if (srcDataSources != null) {
for (Map.Entry<String, DatasourceConfig> 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);
}
}
}
}
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java
-84
View File
@@ -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<String, DatasourceConfig> datasourceConfigs;
private List<String> adapterConfigs;
public List<String> getAdapterConfigs() {
return adapterConfigs;
}
public Map<String, DatasourceConfig> getDatasourceConfigs() {
return datasourceConfigs;
}
public void setDatasourceConfigs(Map<String, DatasourceConfig> datasourceConfigs) {
this.datasourceConfigs = datasourceConfigs;
if (datasourceConfigs != null) {
for (Map.Entry<String, DatasourceConfig> 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<String> 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);
}
}
}
}
}
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java
-3
View File
@@ -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;
client-adapter/launcher/src/main/resources/application.yml
+8 -21
View File
@@ -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
client-adapter/rdb/pom.xml
+20
View File
@@ -80,6 +80,26 @@
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-antrun-plugin</artifactId>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>run</goal>
</goals>
<configuration>
<tasks>
<copy todir="${project.basedir}/../launcher/target/classes/rdb" overwrite="true" >
<fileset dir="${project.basedir}/target/classes/rdb" erroronmissingdir="true">
<include name="*.yml"/>
</fileset>
</copy>
</tasks>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java
+9 -4
View File
@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.client.adapter.rdb;
import java.io.File;
import java.sql.Connection;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
@@ -14,7 +15,7 @@ import org.slf4j.LoggerFactory;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.otter.canal.client.adapter.OuterAdapter;
import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfigLoader;
import com.alibaba.otter.canal.client.adapter.rdb.config.ConfigLoader;
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.*;
@@ -33,12 +34,16 @@ public class RdbAdapter implements OuterAdapter {
@Override
public void init(OuterAdapterConfig configuration) {
SPI spi = this.getClass().getAnnotation(SPI.class);
Map<String, MappingConfig> rdbMappingTmp = MappingConfigLoader.load(spi.value());
System.out.println("xxxxx: " + this.getClass().getClassLoader().getResource("").getPath());
File file = new File(this.getClass().getClassLoader().getResource("").getPath() + "rdb" + File.separator);
System.out.println(file.getAbsolutePath());
Map<String, MappingConfig> rdbMappingTmp = ConfigLoader.load();
// 过滤不匹配的key的配置
rdbMappingTmp.forEach((key, mappingConfig) -> {
if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null)
|| mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey())) {
|| (mappingConfig.getOuterAdapterKey() != null
&& mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
rdbMapping.put(key, mappingConfig);
}
});
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java
+46
View File
@@ -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<String, MappingConfig> load() {
logger.info("## Start loading rdb mapping config ... ");
Map<String, MappingConfig> result = new LinkedHashMap<>();
Map<String, String> 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;
}
}
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java
-100
View File
@@ -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<String, MappingConfig> load(String name) {
logger.info("## Start loading rdb mapping config ... ");
Map<String, MappingConfig> result = new LinkedHashMap<>();
Collection<String> 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
}
}
}
}
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
+4 -3
View File
@@ -148,6 +148,7 @@ public class RdbSyncService {
conn.commit();
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
conn.rollback();
} finally {
conn.close();
@@ -230,6 +231,7 @@ public class RdbSyncService {
conn.commit();
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
conn.rollback();
} finally {
conn.close();
@@ -283,14 +285,13 @@ public class RdbSyncService {
conn.commit();
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
conn.rollback();
} finally {
conn.close();
}
}
/**
* 获取目标字段类型
*
@@ -309,7 +310,7 @@ public class RdbSyncService {
columnType = new LinkedHashMap<>();
final Map<String, Integer> 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();
client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java
+2 -4
View File
@@ -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<String, MappingConfig> configMap = MappingConfigLoader.load("oracle");
Map<String, MappingConfig> configMap = ConfigLoader.load();
Assert.assertFalse(configMap.isEmpty());
}
client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java
+1 -1
View File
@@ -14,7 +14,7 @@ public class Common {
OuterAdapterConfig outerAdapterConfig = new OuterAdapterConfig();
outerAdapterConfig.setName("rdb");
outerAdapterConfig.setKey("oralce1");
//outerAdapterConfig.setKey("oralce1");
Map<String, String> properties = new HashMap<>();
properties.put("jdbc.driveClassName", "oracle.jdbc.OracleDriver");
properties.put("jdbc.url", "jdbc:oracle:thin:@127.0.0.1:49161:XE");
client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java
-2
View File
@@ -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();
}
client-adapter/rdb/src/test/resources/oracle/mytest_user.yml → client-adapter/rdb/src/test/resources/rdb/mytest_user.yml
View File