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; + } +}