mirror of
https://github.com/alibaba/canal.git
synced 2026-11-04 00:30:15 +08:00
rdb专用的的ClassExtensionLoader,默认canal client logger example配置
This commit is contained in:
@@ -40,6 +40,8 @@ public class ExtensionLoader<T> {
|
||||
|
||||
private static final ConcurrentMap<Class<?>, Object> EXTENSION_INSTANCES = new ConcurrentHashMap<>();
|
||||
|
||||
private static final ConcurrentMap<String, Object> EXTENSION_KEY_INSTANCES = new ConcurrentHashMap<>();
|
||||
|
||||
private final Class<?> type;
|
||||
|
||||
private final String classLoaderPolicy;
|
||||
@@ -116,6 +118,30 @@ public class ExtensionLoader<T> {
|
||||
return (T) instance;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public T getExtension(String name, String key) {
|
||||
if (name == null || name.length() == 0) throw new IllegalArgumentException("Extension name == null");
|
||||
if ("true".equals(name)) {
|
||||
return getDefaultExtension();
|
||||
}
|
||||
Holder<Object> holder = cachedInstances.get(name + "-" + key);
|
||||
if (holder == null) {
|
||||
cachedInstances.putIfAbsent(name + "-" + key, new Holder<>());
|
||||
holder = cachedInstances.get(name + "-" + key);
|
||||
}
|
||||
Object instance = holder.get();
|
||||
if (instance == null) {
|
||||
synchronized (holder) {
|
||||
instance = holder.get();
|
||||
if (instance == null) {
|
||||
instance = createExtension(name, key);
|
||||
holder.set(instance);
|
||||
}
|
||||
}
|
||||
}
|
||||
return (T) instance;
|
||||
}
|
||||
|
||||
/**
|
||||
* 返回缺省的扩展,如果没有设置则返回<code>null</code>
|
||||
*/
|
||||
@@ -149,14 +175,19 @@ public class ExtensionLoader<T> {
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public T newInstance(String name) {
|
||||
private T createExtension(String name, String key) {
|
||||
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();
|
||||
T instance = (T) EXTENSION_KEY_INSTANCES.get(name + "-" + key);
|
||||
if (instance == null) {
|
||||
EXTENSION_KEY_INSTANCES.putIfAbsent(name + "-" + key, clazz.newInstance());
|
||||
instance = (T) EXTENSION_KEY_INSTANCES.get(name + "-" + key);
|
||||
}
|
||||
return instance;
|
||||
} catch (Throwable t) {
|
||||
throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type
|
||||
+ ") could not be instantiated: " + t.getMessage(),
|
||||
@@ -164,6 +195,24 @@ public class ExtensionLoader<T> {
|
||||
}
|
||||
}
|
||||
|
||||
@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<String, Class<?>> getExtensionClasses() {
|
||||
Map<String, Class<?>> classes = cachedClasses.get();
|
||||
if (classes == null) {
|
||||
|
||||
@@ -23,7 +23,7 @@
|
||||
</fileSet>
|
||||
<fileSet>
|
||||
<directory>./src/main/resources</directory>
|
||||
<outputDirectory>/config</outputDirectory>
|
||||
<outputDirectory>/conf</outputDirectory>
|
||||
<includes>
|
||||
<include>**/*</include>
|
||||
|
||||
@@ -31,21 +31,21 @@
|
||||
</fileSet>
|
||||
<fileSet>
|
||||
<directory>../elasticsearch/src/main/resources/es</directory>
|
||||
<outputDirectory>/config/es</outputDirectory>
|
||||
<outputDirectory>/conf/es</outputDirectory>
|
||||
<includes>
|
||||
<include>**/*</include>
|
||||
</includes>
|
||||
</fileSet>
|
||||
<fileSet>
|
||||
<directory>../hbase/src/main/resources/hbase</directory>
|
||||
<outputDirectory>/config/hbase</outputDirectory>
|
||||
<outputDirectory>/conf/hbase</outputDirectory>
|
||||
<includes>
|
||||
<include>**/*</include>
|
||||
</includes>
|
||||
</fileSet>
|
||||
<fileSet>
|
||||
<directory>../rdb/src/main/resources/</directory>
|
||||
<outputDirectory>/config</outputDirectory>
|
||||
<outputDirectory>/conf</outputDirectory>
|
||||
<excludes>
|
||||
<exclude>META-INF/**</exclude>
|
||||
</excludes>
|
||||
|
||||
@@ -10,8 +10,6 @@ case "`uname`" in
|
||||
;;
|
||||
esac
|
||||
base=${bin_abs_path}/..
|
||||
#canal_conf=$base/config/canal.properties
|
||||
#logback_configurationFile=$base/conf/logback.xml
|
||||
export LANG=en_US.UTF-8
|
||||
export BASE=$base
|
||||
|
||||
|
||||
@@ -125,7 +125,7 @@ public class CanalAdapterLoader {
|
||||
try {
|
||||
OuterAdapter adapter;
|
||||
if ("rdb".equalsIgnoreCase(config.getName())) {
|
||||
adapter = loader.newInstance(config.getName());
|
||||
adapter = loader.getExtension(config.getName(), config.getKey());
|
||||
} else {
|
||||
adapter = loader.getExtension(config.getName());
|
||||
}
|
||||
|
||||
@@ -11,16 +11,16 @@ spring:
|
||||
time-zone: GMT+8
|
||||
default-property-inclusion: non_null
|
||||
|
||||
#canal.conf:
|
||||
# canalServerHost: 127.0.0.1:11111
|
||||
canal.conf:
|
||||
canalServerHost: 127.0.0.1:11111
|
||||
# zookeeperHosts: slave1:2181
|
||||
# bootstrapServers: slave1:6667 #or rocketmq
|
||||
# flatMessage: true
|
||||
# canalInstances:
|
||||
# - instance: example
|
||||
# groups:
|
||||
# - outAdapters:
|
||||
# - name: logger
|
||||
canalInstances:
|
||||
- instance: example
|
||||
groups:
|
||||
- outAdapters:
|
||||
- name: logger
|
||||
# - name: rdb
|
||||
# key: oracle1
|
||||
# properties:
|
||||
|
||||
Reference in New Issue
Block a user