From 661752096c101f8f823d43b74f300b96b4c13bde Mon Sep 17 00:00:00 2001 From: machey Date: Sun, 30 Dec 2018 17:04:15 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=B3=A8=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/deployer/CanalLauncher.java | 10 +- ...r.java => ManagerRemoteConfigMonitor.java} | 124 ++++++++++++------ 2 files changed, 91 insertions(+), 43 deletions(-) rename deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/{ManagerDbConfigMonitor.java => ManagerRemoteConfigMonitor.java} (87%) 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 50cc62e2..aa9ac603 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,7 @@ import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.deployer.monitor.ManagerDbConfigMonitor; +import com.alibaba.otter.canal.deployer.monitor.ManagerRemoteConfigMonitor; /** * canal独立版本启动的入口类 @@ -30,7 +30,7 @@ public class CanalLauncher { logger.info("## load canal configurations"); String conf = System.getProperty("canal.conf", "classpath:canal.properties"); Properties properties = new Properties(); - ManagerDbConfigMonitor managerDbConfigMonitor = null; + ManagerRemoteConfigMonitor managerDbConfigMonitor = null; if (conf.startsWith(CLASSPATH_URL_PREFIX)) { conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); properties.load(CanalLauncher.class.getClassLoader().getResourceAsStream(conf)); @@ -44,7 +44,7 @@ public class CanalLauncher { // load remote config String jdbcUsername = properties.getProperty("canal.manager.jdbc.username"); String jdbcPassword = properties.getProperty("canal.manager.jdbc.password"); - managerDbConfigMonitor = new ManagerDbConfigMonitor(jdbcUrl, jdbcUsername, jdbcPassword); + managerDbConfigMonitor = new ManagerRemoteConfigMonitor(jdbcUrl, jdbcUsername, jdbcPassword); // 加载远程canal.properties Properties remoteConfig = managerDbConfigMonitor.loadRemoteConfig(); // 加载remote instance配置 @@ -62,12 +62,12 @@ public class CanalLauncher { canalStater.start(properties); if (managerDbConfigMonitor != null) { - managerDbConfigMonitor.start(new ManagerDbConfigMonitor.Listener() { + managerDbConfigMonitor.start(new ManagerRemoteConfigMonitor.Listener() { @Override public void onChange(Properties properties) { try { - // 远程配置canal.properties修改重新启动 + // 远程配置canal.properties修改重新加载整个应用 canalStater.destroy(); canalStater.start(properties); } catch (Throwable throwable) { diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerDbConfigMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerRemoteConfigMonitor.java similarity index 87% rename from deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerDbConfigMonitor.java rename to deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerRemoteConfigMonitor.java index 8c8c27dd..d64cffd8 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerDbConfigMonitor.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerRemoteConfigMonitor.java @@ -18,11 +18,17 @@ import com.alibaba.otter.canal.deployer.CanalConstants; import com.google.common.base.Joiner; import com.google.common.collect.MapMaker; -public class ManagerDbConfigMonitor { +/** + * 远程配置装载监控 + * + * @author rewerma 2018-12-30 下午05:12:16 + * @version 1.0.1 + */ +public class ManagerRemoteConfigMonitor { - private static final Logger logger = LoggerFactory.getLogger(ManagerDbConfigMonitor.class); + private static final Logger logger = LoggerFactory.getLogger(ManagerRemoteConfigMonitor.class); - private Map remoteInstanceConfigs = new MapMaker().makeMap(); + private Map remoteInstanceConfigs = new MapMaker().makeMap(); private Connection conn; private String jdbcUrl; @@ -35,7 +41,7 @@ public class ManagerDbConfigMonitor { private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2, new NamedThreadFactory("remote-canal-config-scan")); - public ManagerDbConfigMonitor(String jdbcUrl, String jdbcUsername, String jdbcPassword){ + public ManagerRemoteConfigMonitor(String jdbcUrl, String jdbcUsername, String jdbcPassword){ this.jdbcUrl = jdbcUrl; this.jdbcUsername = jdbcUsername; this.jdbcPassword = jdbcPassword; @@ -53,6 +59,11 @@ public class ManagerDbConfigMonitor { return conn; } + /** + * 加载远程 canal.properties文件 覆盖本地 + * + * @return 远程配置的properties + */ public Properties loadRemoteConfig() { Properties properties = null; try { @@ -74,6 +85,9 @@ public class ManagerDbConfigMonitor { return properties; } + /** + * 加载远程的instance配置 覆盖本地 + */ public void loadRemoteInstanceConfigs() { try { // 加载远程instance配置 @@ -86,6 +100,11 @@ public class ManagerDbConfigMonitor { } } + /** + * 获取远程canal.properties配置内容 + * + * @return 内容对象 + */ 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)) { @@ -103,6 +122,11 @@ public class ManagerDbConfigMonitor { return null; } + /** + * 覆盖本地 canal.properties + * + * @param content 远程配置内容文本 + */ private void overrideLocalCanalConfig(String content) { try (FileWriter writer = new FileWriter(getConfPath() + "canal.properties")) { writer.write(content); @@ -112,38 +136,11 @@ public class ManagerDbConfigMonitor { } } - public void start(final Listener listener) { - // 监听canal.properties变化 - executor.scheduleWithFixedDelay(new Runnable() { - - public void run() { - try { - Properties properties = loadRemoteConfig(); - if (properties != null) { - // 重启整个进程 - listener.onChange(properties); - } - } catch (Throwable e) { - logger.error("scan failed", e); - } - } - - }, 10, scanIntervalInSecond, TimeUnit.SECONDS); - - // 监听instance变化 - executor.scheduleWithFixedDelay(new Runnable() { - - public void run() { - try { - loadRemoteInstanceConfigs(); - } catch (Throwable e) { - logger.error("scan failed", e); - } - } - - }, 10, 3, TimeUnit.SECONDS); - } - + /** + * 获取远程instance新增、修改、删除配置 + * + * @return Map[0]：新增或修改的instance配置; Map[1]：删除的instance配置 + */ @SuppressWarnings("unchecked") private Map[] getModifiedInstanceConfigs() { Map[] res = new Map[2]; @@ -216,12 +213,16 @@ public class ManagerDbConfigMonitor { } } + /** + * 覆盖本地instance配置 + * + * @param modifiedInstanceConfigs 有变更的配置项 + */ private void overrideLocalInstanceConfigs(Map[] modifiedInstanceConfigs) { Map changedInstanceConfigs = modifiedInstanceConfigs[0]; if (changedInstanceConfigs != null) { for (ConfigItem configItem : changedInstanceConfigs.values()) { - try (FileWriter writer = new FileWriter( - getConfPath() + configItem.getName() + "/instance.properties")) { + try (FileWriter writer = new FileWriter(getConfPath() + configItem.getName() + "/instance.properties")) { writer.write(configItem.getContent()); writer.flush(); } catch (Exception e) { @@ -240,6 +241,45 @@ public class ManagerDbConfigMonitor { } } + /** + * 监听 canal 主配置和 instance 配置变化 + * + * @param listener 监听回调方法 + */ + public void start(final Listener listener) { + // 监听canal.properties变化 + executor.scheduleWithFixedDelay(new Runnable() { + + public void run() { + try { + Properties properties = loadRemoteConfig(); + if (properties != null) { + listener.onChange(properties); + } + } catch (Throwable e) { + logger.error("scan failed", e); + } + } + + }, 10, scanIntervalInSecond, TimeUnit.SECONDS); + + // 监听instance变化 + executor.scheduleWithFixedDelay(new Runnable() { + + public void run() { + try { + loadRemoteInstanceConfigs(); + } catch (Throwable e) { + logger.error("scan failed", e); + } + } + + }, 10, 3, TimeUnit.SECONDS); + } + + /** + * 销毁 + */ public void destroy() { executor.shutdownNow(); if (conn != null) { @@ -251,6 +291,11 @@ public class ManagerDbConfigMonitor { } } + /** + * 获取conf文件夹所在路径 + * + * @return 路径地址 + */ private String getConfPath() { String classpath = this.getClass().getResource("/").getPath(); String confPath = classpath + ".." + File.separator + "conf" + File.separator; @@ -261,6 +306,9 @@ public class ManagerDbConfigMonitor { } } + /** + * 配置对应对象 + */ public static class ConfigItem { private Long id;