chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

修改yml配置读取绑定，遵循spring yml加载方式，支持environment级占位符替换

Browse Source
This commit is contained in:
mcy
2019-02-01 10:11:52 +08:00
parent a20960834b
commit 8211ef35e6
45 changed files with 4195 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java
View File

@@ -2,10 +2,7 @@ package com.alibaba.otter.canal.adapter.launcher.loader;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -13,7 +10,12 @@ import java.util.concurrent.Future;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.env.EnumerablePropertySource;
import org.springframework.core.env.Environment;
import org.springframework.core.env.PropertySource;
import org.springframework.core.env.StandardEnvironment;
import com.alibaba.otter.canal.adapter.launcher.config.SpringContext;
import com.alibaba.otter.canal.client.adapter.OuterAdapter;
import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader;
@@ -62,7 +64,7 @@ public class CanalAdapterLoader {
for (CanalClientConfig.Group connectorGroup : canalAdapter.getGroups()) {
List<OuterAdapter> canalOutConnectors = new ArrayList<>();
for (OuterAdapterConfig c : connectorGroup.getOuterAdapters()) {
loadConnector(c, canalOutConnectors);
loadAdapter(c, canalOutConnectors);
}
canalOuterAdapterGroups.add(canalOutConnectors);
}
@@ -91,7 +93,7 @@ public class CanalAdapterLoader {
List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
List<OuterAdapter> canalOuterAdapters = new ArrayList<>();
for (OuterAdapterConfig config : group.getOuterAdapters()) {
loadConnector(config, canalOuterAdapters);
loadAdapter(config, canalOuterAdapters);
}
canalOuterAdapterGroups.add(canalOuterAdapters);
@@ -103,8 +105,8 @@ public class CanalAdapterLoader {
canalClientConfig.getFlatMessage());
canalMQWorker.put(canalAdapter.getInstance() + "-kafka-" + group.getGroupId(), canalKafkaWorker);
canalKafkaWorker.start();
logger.info("Start adapter for canal-client mq topic: {} succeed", canalAdapter.getInstance() + "-"
+ group.getGroupId());
logger.info("Start adapter for canal-client mq topic: {} succeed",
canalAdapter.getInstance() + "-" + group.getGroupId());
}
}
} else if ("rocketMQ".equalsIgnoreCase(canalClientConfig.getMode())) {
@@ -114,7 +116,7 @@ public class CanalAdapterLoader {
List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
List<OuterAdapter> canalOuterAdapters = new ArrayList<>();
for (OuterAdapterConfig config : group.getOuterAdapters()) {
loadConnector(config, canalOuterAdapters);
loadAdapter(config, canalOuterAdapters);
}
canalOuterAdapterGroups.add(canalOuterAdapters);
CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker(canalClientConfig,
@@ -128,14 +130,14 @@ public class CanalAdapterLoader {
canalMQWorker.put(canalAdapter.getInstance() + "-rocketmq-" + group.getGroupId(), rocketMQWorker);
rocketMQWorker.start();
logger.info("Start adapter for canal-client mq topic: {} succeed", canalAdapter.getInstance() + "-"
+ group.getGroupId());
logger.info("Start adapter for canal-client mq topic: {} succeed",
canalAdapter.getInstance() + "-" + group.getGroupId());
}
}
}
}
private void loadConnector(OuterAdapterConfig config, List<OuterAdapter> canalOutConnectors) {
private void loadAdapter(OuterAdapterConfig config, List<OuterAdapter> canalOutConnectors) {
try {
OuterAdapter adapter;
adapter = loader.getExtension(config.getName(), StringUtils.trimToEmpty(config.getKey()));
@@ -143,7 +145,23 @@ public class CanalAdapterLoader {
ClassLoader cl = Thread.currentThread().getContextClassLoader();
// 替换ClassLoader
Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader());
adapter.init(config);
Environment env = (Environment) SpringContext.getBean(Environment.class);
Properties evnProperties = null;
if (env instanceof StandardEnvironment) {
evnProperties = new Properties();
for (PropertySource<?> propertySource : ((StandardEnvironment) env).getPropertySources()) {
if (propertySource instanceof EnumerablePropertySource) {
String[] names = ((EnumerablePropertySource) propertySource).getPropertyNames();
for (String name : names) {
Object val = propertySource.getProperty(name);
if (val != null) {
evnProperties.put(name, val);
}
}
}
}
}
adapter.init(config, evnProperties);
Thread.currentThread().setContextClassLoader(cl);
canalOutConnectors.add(adapter);
logger.info("Load canal adapter: {} succeed", config.getName());