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