From a7467539c7b2ceb094baf22325cdb3d75227a3f3 Mon Sep 17 00:00:00 2001 From: mcy Date: Fri, 25 Jan 2019 17:50:50 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E6=95=B4=E7=90=86server=E7=AB=AF=E8=BF=9C?= =?UTF-8?q?=E7=A8=8B=E9=85=8D=E7=BD=AE,=20=E7=BB=9F=E4=B8=80=E6=89=A9?= =?UTF-8?q?=E5=B1=95=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/kafka/KafkaCanalConnector.java | 6 +- .../otter/canal/deployer/CanalLauncher.java | 31 +-- .../deployer/monitor/remote/ConfigItem.java | 47 ++++ .../DbRemoteConfigLoader.java} | 242 +++++++----------- .../remote/RemoteCanalConfigMonitor.java | 13 + .../monitor/remote/RemoteConfigLoader.java | 36 +++ .../remote/RemoteConfigLoaderFactory.java | 35 +++ .../monitor/remote/RemoteInstanceMonitor.java | 31 +++ 8 files changed, 274 insertions(+), 167 deletions(-) create mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/ConfigItem.java rename deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/{ManagerRemoteConfigMonitor.java => remote/DbRemoteConfigLoader.java} (59%) create mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteCanalConfigMonitor.java create mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoader.java create mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoaderFactory.java create mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitor.java diff --git a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java index 9b8b263a..f3bae97d 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java @@ -1,8 +1,10 @@ package com.alibaba.otter.canal.client.kafka; import java.util.*; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import com.google.common.collect.MapMaker; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; @@ -31,7 +33,7 @@ import com.google.common.collect.Lists; public class KafkaCanalConnector implements CanalMQConnector { protected KafkaConsumer kafkaConsumer; - protected KafkaConsumer kafkaConsumer2; // 用于扁平message的数据消费 + protected KafkaConsumer kafkaConsumer2; // 用于扁平message的数据消费 protected String topic; protected Integer partition; protected Properties properties; @@ -39,7 +41,7 @@ public class KafkaCanalConnector implements CanalMQConnector { protected volatile boolean running = false; protected boolean flatMessage; - private Map currentOffsets = new HashMap<>(); + private Map currentOffsets = new ConcurrentHashMap<>(); public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize, boolean flatMessage){ diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java index 839daa99..aab5cda6 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java @@ -7,7 +7,9 @@ import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.deployer.monitor.ManagerRemoteConfigMonitor; +import com.alibaba.otter.canal.deployer.monitor.remote.RemoteConfigLoader; +import com.alibaba.otter.canal.deployer.monitor.remote.RemoteConfigLoaderFactory; +import com.alibaba.otter.canal.deployer.monitor.remote.RemoteCanalConfigMonitor; /** * canal独立版本启动的入口类 @@ -30,7 +32,7 @@ public class CanalLauncher { logger.info("## load canal configurations"); String conf = System.getProperty("canal.conf", "classpath:canal.properties"); Properties properties = new Properties(); - ManagerRemoteConfigMonitor managerDbConfigMonitor = null; + RemoteConfigLoader remoteConfigLoader = null; if (conf.startsWith(CLASSPATH_URL_PREFIX)) { conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); properties.load(CanalLauncher.class.getClassLoader().getResourceAsStream(conf)); @@ -38,31 +40,24 @@ public class CanalLauncher { properties.load(new FileInputStream(conf)); } - String jdbcUrl = properties.getProperty("canal.manager.jdbc.url"); - if (!StringUtils.isEmpty(jdbcUrl)) { - logger.info("## load remote canal configurations"); - // load remote config - String jdbcUsername = properties.getProperty("canal.manager.jdbc.username"); - String jdbcPassword = properties.getProperty("canal.manager.jdbc.password"); - managerDbConfigMonitor = new ManagerRemoteConfigMonitor(jdbcUrl, jdbcUsername, jdbcPassword); + remoteConfigLoader = RemoteConfigLoaderFactory.getRemoteConfigLoader(properties); + if (remoteConfigLoader != null) { // 加载远程canal.properties - Properties remoteConfig = managerDbConfigMonitor.loadRemoteConfig(); + Properties remoteConfig = remoteConfigLoader.loadRemoteConfig(); // 加载remote instance配置 - managerDbConfigMonitor.loadRemoteInstanceConfigs(); + remoteConfigLoader.loadRemoteInstanceConfigs(); if (remoteConfig != null) { properties = remoteConfig; } else { - managerDbConfigMonitor = null; + remoteConfigLoader = null; } - } else { - logger.info("## load canal configurations"); } final CanalStater canalStater = new CanalStater(); canalStater.start(properties); - if (managerDbConfigMonitor != null) { - managerDbConfigMonitor.start(new ManagerRemoteConfigMonitor.Listener() { + if (remoteConfigLoader != null) { + remoteConfigLoader.startMonitor(new RemoteCanalConfigMonitor() { @Override public void onChange(Properties properties) { @@ -81,8 +76,8 @@ public class CanalLauncher { Thread.sleep(1000); } - if (managerDbConfigMonitor != null) { - managerDbConfigMonitor.destroy(); + if (remoteConfigLoader != null) { + remoteConfigLoader.destroy(); } } catch (Throwable e) { logger.error("## Something goes wrong when starting up the canal Server:", e); diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/ConfigItem.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/ConfigItem.java new file mode 100644 index 00000000..5fb9b4c6 --- /dev/null +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/ConfigItem.java @@ -0,0 +1,47 @@ +package com.alibaba.otter.canal.deployer.monitor.remote; + +/** + * 配置对应对象 + * + * @author rewerma 2019-01-25 下午05:20:16 + * @version 1.0.0 + */ +public class ConfigItem { + + private Long id; + private String name; + private String content; + private long modifiedTime; + + public Long getId() { + return id; + } + + public void setId(Long id) { + this.id = id; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getContent() { + return content; + } + + public void setContent(String content) { + this.content = content; + } + + public long getModifiedTime() { + return modifiedTime; + } + + public void setModifiedTime(long modifiedTime) { + this.modifiedTime = modifiedTime; + } +} diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerRemoteConfigMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/DbRemoteConfigLoader.java similarity index 59% rename from deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerRemoteConfigMonitor.java rename to deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/DbRemoteConfigLoader.java index 6cb0ebe2..7cf7fef0 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerRemoteConfigMonitor.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/DbRemoteConfigLoader.java @@ -1,39 +1,41 @@ -package com.alibaba.otter.canal.deployer.monitor; +package com.alibaba.otter.canal.deployer.monitor.remote; import java.io.ByteArrayInputStream; import java.io.File; import java.io.FileWriter; import java.nio.charset.StandardCharsets; -import java.sql.*; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; import java.util.*; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.common.utils.NamedThreadFactory; import com.alibaba.otter.canal.deployer.CanalConstants; import com.google.common.base.Joiner; import com.google.common.collect.MapMaker; /** - * 远程配置装载监控 + * 基于数据库的远程配置装载器 * - * @author rewerma 2018-12-30 下午05:12:16 - * @version 1.0.1 + * @author rewerma 2019-01-25 下午05:20:16 + * @version 1.0.0 */ -public class ManagerRemoteConfigMonitor { +public class DbRemoteConfigLoader implements RemoteConfigLoader { - private static final Logger logger = LoggerFactory.getLogger(ManagerRemoteConfigMonitor.class); + private static final Logger logger = LoggerFactory.getLogger(DbRemoteConfigLoader.class); private Map remoteInstanceConfigs = new MapMaker().makeMap(); - private Connection conn; - private String jdbcUrl; - private String jdbcUsername; - private String jdbcPassword; + private DruidDataSource dataSource; private long currentConfigTimestamp = 0; @@ -41,29 +43,36 @@ public class ManagerRemoteConfigMonitor { private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2, new NamedThreadFactory("remote-canal-config-scan")); - public ManagerRemoteConfigMonitor(String jdbcUrl, String jdbcUsername, String jdbcPassword){ - this.jdbcUrl = jdbcUrl; - this.jdbcUsername = jdbcUsername; - this.jdbcPassword = jdbcPassword; - } + private RemoteInstanceMonitor remoteInstanceMonitor = new RemoteInstanceMonitorImpl(); - public void setScanIntervalInSecond(long scanIntervalInSecond) { - this.scanIntervalInSecond = scanIntervalInSecond; - } - - private Connection getConn() throws Exception { - if (conn == null || conn.isClosed()) { - Class.forName("com.mysql.jdbc.Driver"); - conn = DriverManager.getConnection(jdbcUrl, jdbcUsername, jdbcPassword); + public DbRemoteConfigLoader(String driverName, String jdbcUrl, String jdbcUsername, String jdbcPassword){ + dataSource = new DruidDataSource(); + if (StringUtils.isEmpty(driverName)) { + driverName = "com.mysql.jdbc.Driver"; + } + dataSource.setDriverClassName(driverName); + dataSource.setUrl(jdbcUrl); + dataSource.setUsername(jdbcUsername); + dataSource.setPassword(jdbcPassword); + dataSource.setInitialSize(1); + dataSource.setMinIdle(1); + dataSource.setMaxActive(1); + dataSource.setMaxWait(60000); + dataSource.setTimeBetweenEvictionRunsMillis(60000); + dataSource.setMinEvictableIdleTimeMillis(300000); + try { + dataSource.init(); + } catch (SQLException e) { + throw new RuntimeException(e.getMessage(), e); } - return conn; } /** - * 加载远程 canal.properties文件 覆盖本地 + * 加载远程 canal.properties文件 * * @return 远程配置的properties */ + @Override public Properties loadRemoteConfig() { Properties properties = null; try { @@ -87,15 +96,14 @@ public class ManagerRemoteConfigMonitor { } /** - * 加载远程的instance配置 覆盖本地 + * 覆盖本地 canal.properties + * + * @param content 远程配置内容文本 */ - public void loadRemoteInstanceConfigs() { - try { - // 加载远程instance配置 - Map[] modifiedConfigs = getModifiedInstanceConfigs(); - if (modifiedConfigs != null) { - overrideLocalInstanceConfigs(modifiedConfigs); - } + private void overrideLocalCanalConfig(String content) { + try (FileWriter writer = new FileWriter(getConfPath() + "canal.properties")) { + writer.write(content); + writer.flush(); } catch (Exception e) { logger.error(e.getMessage(), e); } @@ -108,7 +116,9 @@ public class ManagerRemoteConfigMonitor { */ private ConfigItem getRemoteCanalConfig() { String sql = "select name, content, modified_time from canal_config where id=1"; - try (Statement stmt = getConn().createStatement(); ResultSet rs = stmt.executeQuery(sql)) { + try (Connection conn = dataSource.getConnection(); + Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery(sql)) { if (rs.next()) { ConfigItem configItem = new ConfigItem(); configItem.setId(1L); @@ -124,30 +134,27 @@ public class ManagerRemoteConfigMonitor { } /** - * 覆盖本地 canal.properties - * - * @param content 远程配置内容文本 + * 加载远程的instance配置 */ - private void overrideLocalCanalConfig(String content) { - try (FileWriter writer = new FileWriter(getConfPath() + "canal.properties")) { - writer.write(content); - writer.flush(); + @Override + public void loadRemoteInstanceConfigs() { + try { + // 加载远程instance配置 + loadModifiedInstanceConfigs(); } catch (Exception e) { logger.error(e.getMessage(), e); } } /** - * 获取远程instance新增、修改、删除配置 - * - * @return Map[0]：新增或修改的instance配置; Map[1]：删除的instance配置 + * 加载远程instance新增、修改、删除配置 */ - @SuppressWarnings("unchecked") - private Map[] getModifiedInstanceConfigs() { - Map[] res = new Map[2]; + private void loadModifiedInstanceConfigs() { Map remoteConfigStatus = new HashMap<>(); String sql = "select id, name, modified_time from canal_instance_config"; - try (Statement stmt = getConn().createStatement(); ResultSet rs = stmt.executeQuery(sql)) { + try (Connection conn = dataSource.getConnection(); + Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery(sql)) { while (rs.next()) { ConfigItem configItem = new ConfigItem(); configItem.setId(rs.getLong("id")); @@ -157,7 +164,6 @@ public class ManagerRemoteConfigMonitor { } } catch (Exception e) { logger.error(e.getMessage(), e); - return null; } if (!remoteConfigStatus.isEmpty()) { @@ -176,10 +182,11 @@ public class ManagerRemoteConfigMonitor { } } if (!changedIds.isEmpty()) { - Map changedInstanceConfig = new HashMap<>(); String contentsSql = "select id, name, content, modified_time from canal_instance_config where id in (" + Joiner.on(",").join(changedIds) + ")"; - try (Statement stmt = getConn().createStatement(); ResultSet rs = stmt.executeQuery(contentsSql)) { + try (Connection conn = dataSource.getConnection(); + Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery(contentsSql)) { while (rs.next()) { ConfigItem configItemNew = new ConfigItem(); configItemNew.setId(rs.getLong("id")); @@ -188,64 +195,20 @@ public class ManagerRemoteConfigMonitor { configItemNew.setModifiedTime(rs.getTimestamp("modified_time").getTime()); remoteInstanceConfigs.put(configItemNew.getName(), configItemNew); - changedInstanceConfig.put(configItemNew.getName(), configItemNew); + remoteInstanceMonitor.onModify(configItemNew); } - res[0] = changedInstanceConfig.isEmpty() ? null : changedInstanceConfig; } catch (Exception e) { logger.error(e.getMessage(), e); } } } - Map removedInstanceConfig = new HashMap<>(); for (String name : remoteInstanceConfigs.keySet()) { if (!remoteConfigStatus.containsKey(name)) { // 删除 remoteInstanceConfigs.remove(name); - removedInstanceConfig.put(name, null); - } - } - res[1] = removedInstanceConfig.isEmpty() ? null : removedInstanceConfig; - - if (res[0] == null && res[1] == null) { - return null; - } else { - return res; - } - } - - /** - * 覆盖本地instance配置 - * - * @param modifiedInstanceConfigs 有变更的配置项 - */ - private void overrideLocalInstanceConfigs(Map[] modifiedInstanceConfigs) { - Map changedInstanceConfigs = modifiedInstanceConfigs[0]; - if (changedInstanceConfigs != null) { - for (ConfigItem configItem : changedInstanceConfigs.values()) { - File instanceDir = new File(getConfPath() + configItem.getName()); - if (!instanceDir.exists()) { - instanceDir.mkdirs(); - } - try (FileWriter writer = new FileWriter( - getConfPath() + configItem.getName() + "/instance.properties")) { - writer.write(configItem.getContent()); - writer.flush(); - logger.info("## Loaded remote instance config: {}/instance.properties ", configItem.getName()); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - } - Map removedInstanceConfigs = modifiedInstanceConfigs[1]; - if (removedInstanceConfigs != null) { - for (String name : removedInstanceConfigs.keySet()) { - File file = new File(getConfPath() + name + "/"); - if (file.exists()) { - deleteDir(file); - logger.info("## Deleted and loaded remote instance config: {} ", name); - } + remoteInstanceMonitor.onDelete(name); } } } @@ -273,9 +236,9 @@ public class ManagerRemoteConfigMonitor { /** * 监听 canal 主配置和 instance 配置变化 * - * @param listener 监听回调方法 + * @param remoteCanalConfigMonitor 监听回调方法 */ - public void start(final Listener listener) { + public void startMonitor(final RemoteCanalConfigMonitor remoteCanalConfigMonitor) { // 监听canal.properties变化 executor.scheduleWithFixedDelay(new Runnable() { @@ -283,10 +246,10 @@ public class ManagerRemoteConfigMonitor { try { Properties properties = loadRemoteConfig(); if (properties != null) { - listener.onChange(properties); + remoteCanalConfigMonitor.onChange(properties); } } catch (Throwable e) { - logger.error("scan failed", e); + logger.error("Scan remote canal config failed", e); } } @@ -299,7 +262,7 @@ public class ManagerRemoteConfigMonitor { try { loadRemoteInstanceConfigs(); } catch (Throwable e) { - logger.error("scan failed", e); + logger.error("Scan remote instance config failed", e); } } @@ -311,12 +274,10 @@ public class ManagerRemoteConfigMonitor { */ public void destroy() { executor.shutdownNow(); - if (conn != null) { - try { - conn.close(); - } catch (SQLException e) { - logger.error(e.getMessage(), e); - } + try { + dataSource.close(); + } catch (Exception e) { + logger.error(e.getMessage(), e); } } @@ -336,50 +297,37 @@ public class ManagerRemoteConfigMonitor { } /** - * 配置对应对象 + * 远程xxx/instance.properties配置监听器实现 */ - public static class ConfigItem { + private class RemoteInstanceMonitorImpl implements RemoteInstanceMonitor { - private Long id; - private String name; - private String content; - private long modifiedTime; - - public Long getId() { - return id; + @Override + public void onAdd(ConfigItem configItem) { + this.onModify(configItem); } - public void setId(Long id) { - this.id = id; + @Override + public void onModify(ConfigItem configItem) { + File instanceDir = new File(getConfPath() + configItem.getName()); + if (!instanceDir.exists()) { + instanceDir.mkdirs(); + } + try (FileWriter writer = new FileWriter(getConfPath() + configItem.getName() + "/instance.properties")) { + writer.write(configItem.getContent()); + writer.flush(); + logger.info("## Loaded remote instance config: {}/instance.properties ", configItem.getName()); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } } - public String getName() { - return name; + @Override + public void onDelete(String instanceName) { + File file = new File(getConfPath() + instanceName + "/"); + if (file.exists()) { + deleteDir(file); + logger.info("## Deleted and loaded remote instance config: {} ", instanceName); + } } - - public void setName(String name) { - this.name = name; - } - - public String getContent() { - return content; - } - - public void setContent(String content) { - this.content = content; - } - - public long getModifiedTime() { - return modifiedTime; - } - - public void setModifiedTime(long modifiedTime) { - this.modifiedTime = modifiedTime; - } - } - - public interface Listener { - - void onChange(T properties); } } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteCanalConfigMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteCanalConfigMonitor.java new file mode 100644 index 00000000..4c0ab955 --- /dev/null +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteCanalConfigMonitor.java @@ -0,0 +1,13 @@ +package com.alibaba.otter.canal.deployer.monitor.remote; + +import java.util.Properties; + +/** + * 远程canal.properties配置监听器接口 + * + * @author rewerma 2019-01-25 下午05:20:16 + * @version 1.0.0 + */ +public interface RemoteCanalConfigMonitor { + void onChange(Properties properties); +} diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoader.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoader.java new file mode 100644 index 00000000..6b7444c0 --- /dev/null +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoader.java @@ -0,0 +1,36 @@ +package com.alibaba.otter.canal.deployer.monitor.remote; + +import java.util.Properties; + +/** + * 远程配置装载器接口 + * + * @author rewerma 2019-01-25 下午05:20:16 + * @version 1.0.0 + */ +public interface RemoteConfigLoader { + + /** + * 加载远程 canal.properties文件 + * + * @return 远程配置的properties + */ + Properties loadRemoteConfig(); + + /** + * 加载远程的instance配置 + */ + void loadRemoteInstanceConfigs(); + + /** + * 启动监听 canal 主配置和 instance 配置变化 + * + * @param remoteCanalConfigMonitor 监听回调方法 + */ + void startMonitor(final RemoteCanalConfigMonitor remoteCanalConfigMonitor); + + /** + * 销毁 + */ + void destroy(); +} diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoaderFactory.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoaderFactory.java new file mode 100644 index 00000000..9256a38a --- /dev/null +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoaderFactory.java @@ -0,0 +1,35 @@ +package com.alibaba.otter.canal.deployer.monitor.remote; + +import java.util.Properties; + +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 远程配置装载器工厂类 + * + * @author rewerma 2019-01-25 下午05:20:16 + * @version 1.0.0 + */ +public class RemoteConfigLoaderFactory { + + private static final Logger logger = LoggerFactory.getLogger(RemoteConfigLoaderFactory.class); + + public static RemoteConfigLoader getRemoteConfigLoader(Properties localProperties) { + String jdbcUrl = localProperties.getProperty("canal.manager.jdbc.url"); + if (!StringUtils.isEmpty(jdbcUrl)) { + logger.info("## load remote canal configurations"); + // load remote config + String driverName = localProperties.getProperty("canal.manager.jdbc.driverName"); + String jdbcUsername = localProperties.getProperty("canal.manager.jdbc.username"); + String jdbcPassword = localProperties.getProperty("canal.manager.jdbc.password"); + return new DbRemoteConfigLoader(driverName, jdbcUrl, jdbcUsername, jdbcPassword); + } + // 可扩展其它远程配置加载器 + + logger.info("## load local canal configurations"); + + return null; + } +} diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitor.java new file mode 100644 index 00000000..28c97788 --- /dev/null +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitor.java @@ -0,0 +1,31 @@ +package com.alibaba.otter.canal.deployer.monitor.remote; + +/** + * 远程xxx/instance.properties配置监听器接口 + * + * @author rewerma 2019-01-25 下午05:20:16 + * @version 1.0.0 + */ +public interface RemoteInstanceMonitor { + + /** + * 新增配置事件 + * + * @param configItem 配置项 + */ + void onAdd(ConfigItem configItem); + + /** + * 修改配置事件 + * + * @param configItem 配置项 + */ + void onModify(ConfigItem configItem); + + /** + * 删除配置事件 + * + * @param instanceName 实例名 + */ + void onDelete(String instanceName); +} From 9a2e890f5741ffc1d8016d08f4bf1893f4487e09 Mon Sep 17 00:00:00 2001 From: mcy Date: Fri, 25 Jan 2019 18:50:30 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=20=E6=95=B4=E7=90=86adapter=E7=AB=AF?= =?UTF-8?q?=E8=BF=9C=E7=A8=8B=E9=85=8D=E7=BD=AE,=20=E7=BB=9F=E4=B8=80?= =?UTF-8?q?=E6=89=A9=E5=B1=95=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../config/BootstrapConfiguration.java | 41 ++- .../launcher/loader/CanalAdapterService.java | 23 +- .../launcher/monitor/remote/ConfigItem.java | 56 ++++ .../DbRemoteConfigLoader.java} | 259 +++++++----------- .../monitor/remote/RemoteAdapterMonitor.java | 31 +++ .../monitor/remote/RemoteConfigLoader.java | 30 ++ .../remote/RemoteConfigLoaderFactory.java | 34 +++ 7 files changed, 280 insertions(+), 194 deletions(-) create mode 100644 client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/ConfigItem.java rename client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/{AdapterRemoteConfigMonitor.java => remote/DbRemoteConfigLoader.java} (58%) create mode 100644 client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitor.java create mode 100644 client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteConfigLoader.java create mode 100644 client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteConfigLoaderFactory.java diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/BootstrapConfiguration.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/BootstrapConfiguration.java index 3f13c00c..ffae1b14 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/BootstrapConfiguration.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/BootstrapConfiguration.java @@ -1,14 +1,14 @@ package com.alibaba.otter.canal.adapter.launcher.config; import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; -import com.alibaba.otter.canal.adapter.launcher.monitor.AdapterRemoteConfigMonitor; -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.core.env.Environment; +import com.alibaba.otter.canal.adapter.launcher.monitor.remote.RemoteConfigLoader; +import com.alibaba.otter.canal.adapter.launcher.monitor.remote.RemoteConfigLoaderFactory; + /** * Bootstrap级别配置加载 * @@ -17,28 +17,25 @@ import org.springframework.core.env.Environment; */ public class BootstrapConfiguration { - private static final Logger logger = LoggerFactory.getLogger(BootstrapConfiguration.class); - @Autowired - private Environment env; + private Environment env; + + private RemoteConfigLoader remoteConfigLoader = null; @PostConstruct public void loadRemoteConfig() { - try { - // 加载远程配置 - String jdbcUrl = env.getProperty("canal.manager.jdbc.url"); - if (StringUtils.isNotEmpty(jdbcUrl)) { - String jdbcUsername = env.getProperty("canal.manager.jdbc.username"); - String jdbcPassword = env.getProperty("canal.manager.jdbc.password"); - AdapterRemoteConfigMonitor configMonitor = new AdapterRemoteConfigMonitor(jdbcUrl, - jdbcUsername, - jdbcPassword); - configMonitor.loadRemoteConfig(); - configMonitor.loadRemoteAdapterConfigs(); - configMonitor.start(); // 启动监听 - } - } catch (Exception e) { - logger.error(e.getMessage(), e); + remoteConfigLoader = RemoteConfigLoaderFactory.getRemoteConfigLoader(env); + if (remoteConfigLoader != null) { + remoteConfigLoader.loadRemoteConfig(); + remoteConfigLoader.loadRemoteAdapterConfigs(); + remoteConfigLoader.startMonitor(); // 启动监听 + } + } + + @PreDestroy + public synchronized void destroy() { + if (remoteConfigLoader != null) { + remoteConfigLoader.destroy(); } } } 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 e2f03470..48cad990 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 @@ -4,8 +4,6 @@ import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import javax.annotation.Resource; -import com.alibaba.otter.canal.adapter.launcher.monitor.AdapterRemoteConfigMonitor; -import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.cloud.context.config.annotation.RefreshScope; @@ -29,27 +27,25 @@ import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; @RefreshScope public class CanalAdapterService { - private static final Logger logger = LoggerFactory.getLogger(CanalAdapterService.class); + private static final Logger logger = LoggerFactory.getLogger(CanalAdapterService.class); - private CanalAdapterLoader adapterLoader; + private CanalAdapterLoader adapterLoader; @Resource - private ContextRefresher contextRefresher; + private ContextRefresher contextRefresher; @Resource - private AdapterCanalConfig adapterCanalConfig; + private AdapterCanalConfig adapterCanalConfig; @Resource - private Environment env; + private Environment env; // 注入bean保证优先注册 @Resource - private SpringContext springContext; + private SpringContext springContext; @Resource - private SyncSwitch syncSwitch; + private SyncSwitch syncSwitch; - private volatile boolean running = false; - - private AdapterRemoteConfigMonitor configMonitor = null; + private volatile boolean running = false; @PostConstruct public synchronized void init() { @@ -75,9 +71,6 @@ public class CanalAdapterService { try { running = false; logger.info("## stop the canal client adapters"); - if (configMonitor != null) { - configMonitor.destroy(); - } if (adapterLoader != null) { adapterLoader.destroy(); diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/ConfigItem.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/ConfigItem.java new file mode 100644 index 00000000..0074fbbe --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/ConfigItem.java @@ -0,0 +1,56 @@ +package com.alibaba.otter.canal.adapter.launcher.monitor.remote; + +/** + * 配置对应对象 + * + * @author rewerma 2019-01-25 下午05:20:16 + * @version 1.0.0 + */ +public class ConfigItem { + + private Long id; + private String category; + private String name; + private String content; + private long modifiedTime; + + public Long getId() { + return id; + } + + public void setId(Long id) { + this.id = id; + } + + public String getCategory() { + return category; + } + + public void setCategory(String category) { + this.category = category; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getContent() { + return content; + } + + public void setContent(String content) { + this.content = content; + } + + public long getModifiedTime() { + return modifiedTime; + } + + public void setModifiedTime(long modifiedTime) { + this.modifiedTime = modifiedTime; + } +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/AdapterRemoteConfigMonitor.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java similarity index 58% rename from client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/AdapterRemoteConfigMonitor.java rename to client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java index 1609a249..824ddcc9 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/AdapterRemoteConfigMonitor.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java @@ -1,10 +1,11 @@ -package com.alibaba.otter.canal.adapter.launcher.monitor; +package com.alibaba.otter.canal.adapter.launcher.monitor.remote; import java.io.File; -import java.io.FileInputStream; import java.io.FileWriter; -import java.net.URL; -import java.sql.*; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -13,54 +14,61 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +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.Constant; -import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader; +import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.common.utils.NamedThreadFactory; import com.google.common.base.Joiner; import com.google.common.collect.MapMaker; /** - * 远程配置装载、监控类 + * 基于数据库的远程配置装载器 * - * @author rewerma @ 2019-01-05 + * @author rewerma 2019-01-25 下午05:20:16 * @version 1.0.0 */ -public class AdapterRemoteConfigMonitor { +public class DbRemoteConfigLoader implements RemoteConfigLoader { - private static final Logger logger = LoggerFactory.getLogger(AdapterRemoteConfigMonitor.class); + private static final Logger logger = LoggerFactory.getLogger(DbRemoteConfigLoader.class); - private Connection conn; - private String jdbcUrl; - private String jdbcUsername; - private String jdbcPassword; + private DruidDataSource dataSource; - private long currentConfigTimestamp = 0; + private static volatile long currentConfigTimestamp = 0; private Map remoteAdapterConfigs = new MapMaker().makeMap(); private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2, new NamedThreadFactory("remote-adapter-config-scan")); - public AdapterRemoteConfigMonitor(String jdbcUrl, String jdbcUsername, String jdbcPassword){ - this.jdbcUrl = jdbcUrl; - this.jdbcUsername = jdbcUsername; - this.jdbcPassword = jdbcPassword; - } + private RemoteAdapterMonitor remoteAdapterMonitor = new RemoteAdapterMonitorImpl(); - private Connection getConn() throws Exception { - if (conn == null || conn.isClosed()) { - Class.forName("com.mysql.jdbc.Driver"); - conn = DriverManager.getConnection(jdbcUrl, jdbcUsername, jdbcPassword); + public DbRemoteConfigLoader(String driverName, String jdbcUrl, String jdbcUsername, String jdbcPassword){ + dataSource = new DruidDataSource(); + if (StringUtils.isEmpty(driverName)) { + driverName = "com.mysql.jdbc.Driver"; + } + dataSource.setDriverClassName(driverName); + dataSource.setUrl(jdbcUrl); + dataSource.setUsername(jdbcUsername); + dataSource.setPassword(jdbcPassword); + dataSource.setInitialSize(1); + dataSource.setMinIdle(1); + dataSource.setMaxActive(1); + dataSource.setMaxWait(60000); + dataSource.setTimeBetweenEvictionRunsMillis(60000); + dataSource.setMinEvictableIdleTimeMillis(300000); + try { + dataSource.init(); + } catch (SQLException e) { + throw new RuntimeException(e.getMessage(), e); } - return conn; } /** - * 加载远程application.yml配置到本地 + * 加载远程application.yml配置 */ + @Override public void loadRemoteConfig() { try { // 加载远程adapter配置 @@ -84,7 +92,9 @@ public class AdapterRemoteConfigMonitor { */ private ConfigItem getRemoteAdapterConfig() { String sql = "select name, content, modified_time from canal_config where id=2"; - try (Statement stmt = getConn().createStatement(); ResultSet rs = stmt.executeQuery(sql)) { + try (Connection conn = dataSource.getConnection(); + Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery(sql)) { if (rs.next()) { ConfigItem configItem = new ConfigItem(); configItem.setId(2L); @@ -114,54 +124,29 @@ public class AdapterRemoteConfigMonitor { } /** - * 启动监听数据库变化 - */ - public void start() { - // 监听application.yml变化 - executor.scheduleWithFixedDelay(() -> { - try { - loadRemoteConfig(); - } catch (Throwable e) { - logger.error("scan remote application.yml failed", e); - } - }, 10, 3, TimeUnit.SECONDS); - - // 监听adapter变化 - executor.scheduleWithFixedDelay(() -> { - try { - loadRemoteAdapterConfigs(); - } catch (Throwable e) { - logger.error("scan remote adapter configs failed", e); - } - }, 10, 3, TimeUnit.SECONDS); - } - - /** - * 加载adapter配置到本地 + * 加载adapter配置 */ + @Override public void loadRemoteAdapterConfigs() { try { // 加载远程adapter配置 - Map[] modifiedConfigs = getModifiedAdapterConfigs(); - if (modifiedConfigs != null) { - overrideLocalAdapterConfigs(modifiedConfigs); - } + loadModifiedAdapterConfigs(); } catch (Exception e) { logger.error(e.getMessage(), e); } } /** - * 获取有变动的adapter配置 - * - * @return Map[0]: 新增修改的配置, Map[1]: 删除的配置 + * 加载有变动的adapter配置 */ @SuppressWarnings("unchecked") - private Map[] getModifiedAdapterConfigs() { + private void loadModifiedAdapterConfigs() { Map[] res = new Map[2]; Map remoteConfigStatus = new HashMap<>(); String sql = "select id, category, name, modified_time from canal_adapter_config"; - try (Statement stmt = getConn().createStatement(); ResultSet rs = stmt.executeQuery(sql)) { + try (Connection conn = dataSource.getConnection(); + Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery(sql)) { while (rs.next()) { ConfigItem configItem = new ConfigItem(); configItem.setId(rs.getLong("id")); @@ -172,7 +157,6 @@ public class AdapterRemoteConfigMonitor { } } catch (Exception e) { logger.error(e.getMessage(), e); - return null; } if (!remoteConfigStatus.isEmpty()) { @@ -192,10 +176,11 @@ public class AdapterRemoteConfigMonitor { } } if (!changedIds.isEmpty()) { - Map changedAdapterConfig = new HashMap<>(); String contentsSql = "select id, category, name, content, modified_time from canal_adapter_config where id in (" + Joiner.on(",").join(changedIds) + ")"; - try (Statement stmt = getConn().createStatement(); ResultSet rs = stmt.executeQuery(contentsSql)) { + try (Connection conn = dataSource.getConnection(); + Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery(contentsSql)) { while (rs.next()) { ConfigItem configItemNew = new ConfigItem(); configItemNew.setId(rs.getLong("id")); @@ -206,63 +191,20 @@ public class AdapterRemoteConfigMonitor { remoteAdapterConfigs.put(configItemNew.getCategory() + "/" + configItemNew.getName(), configItemNew); - changedAdapterConfig.put(configItemNew.getCategory() + "/" + configItemNew.getName(), - configItemNew); + remoteAdapterMonitor.onModify(configItemNew); } - res[0] = changedAdapterConfig.isEmpty() ? null : changedAdapterConfig; } catch (Exception e) { logger.error(e.getMessage(), e); } } } - Map removedAdapterConfig = new HashMap<>(); for (ConfigItem configItem : remoteAdapterConfigs.values()) { if (!remoteConfigStatus.containsKey(configItem.getCategory() + "/" + configItem.getName())) { // 删除 remoteAdapterConfigs.remove(configItem.getCategory() + "/" + configItem.getName()); - removedAdapterConfig.put(configItem.getCategory() + "/" + configItem.getName(), null); - } - } - res[1] = removedAdapterConfig.isEmpty() ? null : removedAdapterConfig; - - if (res[0] == null && res[1] == null) { - return null; - } else { - return res; - } - } - - /** - * 覆盖adapter配置到本地 - * - * @param modifiedAdapterConfigs 变动的配置集合 - */ - private void overrideLocalAdapterConfigs(Map[] modifiedAdapterConfigs) { - Map changedAdapterConfigs = modifiedAdapterConfigs[0]; - if (changedAdapterConfigs != null) { - for (ConfigItem configItem : changedAdapterConfigs.values()) { - try (FileWriter writer = new FileWriter( - getConfPath() + configItem.getCategory() + "/" + configItem.getName())) { - writer.write(configItem.getContent()); - writer.flush(); - logger.info("## Loaded remote adapter config: {}/{}", - configItem.getCategory(), - configItem.getName()); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - } - Map removedAdapterConfigs = modifiedAdapterConfigs[1]; - if (removedAdapterConfigs != null) { - for (String name : removedAdapterConfigs.keySet()) { - File file = new File(getConfPath() + name); - if (file.exists()) { - deleteDir(file); - logger.info("## Deleted and reloaded remote adapter config: {}", name); - } + remoteAdapterMonitor.onDelete(configItem.getCategory() + "/" + configItem.getName()); } } } @@ -302,66 +244,69 @@ public class AdapterRemoteConfigMonitor { } } - public void destroy() { - executor.shutdownNow(); - if (conn != null) { + /** + * 启动监听数据库变化 + */ + @Override + public void startMonitor() { + // 监听application.yml变化 + executor.scheduleWithFixedDelay(() -> { try { - conn.close(); - } catch (SQLException e) { - logger.error(e.getMessage(), e); + loadRemoteConfig(); + } catch (Throwable e) { + logger.error("scan remote application.yml failed", e); } - } + }, 10, 3, TimeUnit.SECONDS); + + // 监听adapter变化 + executor.scheduleWithFixedDelay(() -> { + try { + loadRemoteAdapterConfigs(); + } catch (Throwable e) { + logger.error("scan remote adapter configs failed", e); + } + }, 10, 3, TimeUnit.SECONDS); } /** - * 配置对应对象 + * 销毁 */ - public static class ConfigItem { + @Override + public void destroy() { + executor.shutdownNow(); + try { + dataSource.close(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } - private Long id; - private String category; - private String name; - private String content; - private long modifiedTime; + private class RemoteAdapterMonitorImpl implements RemoteAdapterMonitor { - public Long getId() { - return id; + @Override + public void onAdd(ConfigItem configItem) { + this.onModify(configItem); } - public void setId(Long id) { - this.id = id; + @Override + public void onModify(ConfigItem configItem) { + try (FileWriter writer = new FileWriter( + getConfPath() + configItem.getCategory() + "/" + configItem.getName())) { + writer.write(configItem.getContent()); + writer.flush(); + logger.info("## Loaded remote adapter config: {}/{}", configItem.getCategory(), configItem.getName()); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } } - public String getCategory() { - return category; - } - - public void setCategory(String category) { - this.category = category; - } - - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - - public String getContent() { - return content; - } - - public void setContent(String content) { - this.content = content; - } - - public long getModifiedTime() { - return modifiedTime; - } - - public void setModifiedTime(long modifiedTime) { - this.modifiedTime = modifiedTime; + @Override + public void onDelete(String name) { + File file = new File(getConfPath() + name); + if (file.exists()) { + deleteDir(file); + logger.info("## Deleted and reloaded remote adapter config: {}", name); + } } } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitor.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitor.java new file mode 100644 index 00000000..b0907314 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitor.java @@ -0,0 +1,31 @@ +package com.alibaba.otter.canal.adapter.launcher.monitor.remote; + +/** + * 远程配置监听器接口 + * + * @author rewerma 2019-01-25 下午05:20:16 + * @version 1.0.0 + */ +public interface RemoteAdapterMonitor { + + /** + * 新增配置事件 + * + * @param configItem 配置项 + */ + void onAdd(ConfigItem configItem); + + /** + * 修改配置事件 + * + * @param configItem 配置项 + */ + void onModify(ConfigItem configItem); + + /** + * 删除配置事件 + * + * @param name 配置名 + */ + void onDelete(String name); +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteConfigLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteConfigLoader.java new file mode 100644 index 00000000..36af05bd --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteConfigLoader.java @@ -0,0 +1,30 @@ +package com.alibaba.otter.canal.adapter.launcher.monitor.remote; + +/** + * 远程配置装载器接口 + * + * @author rewerma 2019-01-25 下午05:20:16 + * @version 1.0.0 + */ +public interface RemoteConfigLoader { + + /** + * 加载远程application.yml配置到本地 + */ + void loadRemoteConfig(); + + /** + * 加载adapter配置 + */ + void loadRemoteAdapterConfigs(); + + /** + * 启动监听数据库变化 + */ + void startMonitor(); + + /** + * 销毁 + */ + void destroy(); +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteConfigLoaderFactory.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteConfigLoaderFactory.java new file mode 100644 index 00000000..355d3df5 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteConfigLoaderFactory.java @@ -0,0 +1,34 @@ +package com.alibaba.otter.canal.adapter.launcher.monitor.remote; + +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.core.env.Environment; + +/** + * 远程配置装载器工厂类 + * + * @author rewerma 2019-01-25 下午05:20:16 + * @version 1.0.0 + */ +public class RemoteConfigLoaderFactory { + + private static final Logger logger = LoggerFactory.getLogger(RemoteConfigLoaderFactory.class); + + public static RemoteConfigLoader getRemoteConfigLoader(Environment env) { + try { + String jdbcUrl = env.getProperty("canal.manager.jdbc.url"); + if (!StringUtils.isEmpty(jdbcUrl)) { + // load remote config + String driverName = env.getProperty("canal.manager.jdbc.driverName"); + String jdbcUsername = env.getProperty("canal.manager.jdbc.username"); + String jdbcPassword = env.getProperty("canal.manager.jdbc.password"); + return new DbRemoteConfigLoader(driverName, jdbcUrl, jdbcUsername, jdbcPassword); + } + // 可扩展其它远程配置加载器 + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + return null; + } +}