整理server端远程配置, 统一扩展接口

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;
@@ -39,7 +41,7 @@ public class KafkaCanalConnector implements CanalMQConnector {
     protected volatile boolean               running        = false;
     protected boolean                        flatMessage;
 
-    private Map<Integer, Long>               currentOffsets = new HashMap<>();
+    private Map<Integer, Long>               currentOffsets = new ConcurrentHashMap<>();
 
     public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize,
                                boolean flatMessage){

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");
+            remoteConfigLoader = RemoteConfigLoaderFactory.getRemoteConfigLoader(properties);
-            if (!StringUtils.isEmpty(jdbcUrl)) {
+            if (remoteConfigLoader != null) {
-                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);
                 // 加载远程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) {
+            if (remoteConfigLoader != null) {
-                managerDbConfigMonitor.start(new ManagerRemoteConfigMonitor.Listener<Properties>() {
+                remoteConfigLoader.startMonitor(new RemoteCanalConfigMonitor() {
 
                     @Override
                     public void onChange(Properties properties) {
@@ -81,8 +76,8 @@ public class CanalLauncher {
                 Thread.sleep(1000);
             }
 
-            if (managerDbConfigMonitor != null) {
+            if (remoteConfigLoader != null) {
-                managerDbConfigMonitor.destroy();
+                remoteConfigLoader.destroy();
             }
         } catch (Throwable e) {
             logger.error("## Something goes wrong when starting up the canal Server:", e);

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

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
+ * @author rewerma 2019-01-25 下午05:20:16
- * @version 1.0.1
+ * @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<String, ConfigItem>  remoteInstanceConfigs  = new MapMaker().makeMap();
 
-    private Connection               conn;
+    private DruidDataSource          dataSource;
-    private String                   jdbcUrl;
-    private String                   jdbcUsername;
-    private String                   jdbcPassword;
 
     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){
+    private RemoteInstanceMonitor    remoteInstanceMonitor  = new RemoteInstanceMonitorImpl();
-        this.jdbcUrl = jdbcUrl;
-        this.jdbcUsername = jdbcUsername;
-        this.jdbcPassword = jdbcPassword;
-    }
 
-    public void setScanIntervalInSecond(long scanIntervalInSecond) {
+    public DbRemoteConfigLoader(String driverName, String jdbcUrl, String jdbcUsername, String jdbcPassword){
-        this.scanIntervalInSecond = scanIntervalInSecond;
+        dataSource = new DruidDataSource();
+        if (StringUtils.isEmpty(driverName)) {
+            driverName = "com.mysql.jdbc.Driver";
         }
-
+        dataSource.setDriverClassName(driverName);
-    private Connection getConn() throws Exception {
+        dataSource.setUrl(jdbcUrl);
-        if (conn == null || conn.isClosed()) {
+        dataSource.setUsername(jdbcUsername);
-            Class.forName("com.mysql.jdbc.Driver");
+        dataSource.setPassword(jdbcPassword);
-            conn = DriverManager.getConnection(jdbcUrl, jdbcUsername, 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() {
+    private void overrideLocalCanalConfig(String content) {
-        try {
+        try (FileWriter writer = new FileWriter(getConfPath() + "canal.properties")) {
-            // 加载远程instance配置
+            writer.write(content);
-            Map<String, ConfigItem>[] modifiedConfigs = getModifiedInstanceConfigs();
+            writer.flush();
-            if (modifiedConfigs != null) {
-                overrideLocalInstanceConfigs(modifiedConfigs);
-            }
         } 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
+     * 加载远程的instance配置
-     *
-     * @param content 远程配置内容文本
      */
-    private void overrideLocalCanalConfig(String content) {
+    @Override
-        try (FileWriter writer = new FileWriter(getConfPath() + "canal.properties")) {
+    public void loadRemoteInstanceConfigs() {
-            writer.write(content);
+        try {
-            writer.flush();
+            // 加载远程instance配置
+            loadModifiedInstanceConfigs();
         } catch (Exception e) {
             logger.error(e.getMessage(), e);
         }
     }
 
     /**
-     * 获取远程instance新增、修改、删除配置
+     * 加载远程instance新增、修改、删除配置
-     *
-     * @return Map[0]：新增或修改的instance配置; Map[1]：删除的instance配置
      */
-    @SuppressWarnings("unchecked")
+    private void loadModifiedInstanceConfigs() {
-    private Map<String, ConfigItem>[] getModifiedInstanceConfigs() {
-        Map<String, ConfigItem>[] res = new Map[2];
         Map<String, ConfigItem> 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<String, ConfigItem> 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<String, ConfigItem> removedInstanceConfig = new HashMap<>();
         for (String name : remoteInstanceConfigs.keySet()) {
             if (!remoteConfigStatus.containsKey(name)) {
                 // 删除
                 remoteInstanceConfigs.remove(name);
-                removedInstanceConfig.put(name, null);
+                remoteInstanceMonitor.onDelete(name);
-            }
-        }
-        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<String, ConfigItem>[] modifiedInstanceConfigs) {
-        Map<String, ConfigItem> 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<String, ConfigItem> 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);
-                }
             }
         }
     }
@@ -273,9 +236,9 @@ public class ManagerRemoteConfigMonitor {
     /**
      * 监听 canal 主配置和 instance 配置变化
      *
-     * @param listener 监听回调方法
+     * @param remoteCanalConfigMonitor 监听回调方法
      */
-    public void start(final Listener<Properties> 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,14 +274,12 @@ public class ManagerRemoteConfigMonitor {
      */
     public void destroy() {
         executor.shutdownNow();
-        if (conn != null) {
         try {
-                conn.close();
+            dataSource.close();
-            } catch (SQLException e) {
+        } catch (Exception e) {
             logger.error(e.getMessage(), e);
         }
     }
-    }
 
     /**
      * 获取conf文件夹所在路径
@@ -336,50 +297,37 @@ public class ManagerRemoteConfigMonitor {
     }
 
     /**
-     * 配置对应对象
+     * 远程xxx/instance.properties配置监听器实现
      */
-    public static class ConfigItem {
+    private class RemoteInstanceMonitorImpl implements RemoteInstanceMonitor {
 
-        private Long   id;
+        @Override
-        private String name;
+        public void onAdd(ConfigItem configItem) {
-        private String content;
+            this.onModify(configItem);
-        private long   modifiedTime;
-
-        public Long getId() {
-            return id;
         }
 
-        public void setId(Long id) {
+        @Override
-            this.id = id;
+        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")) {
-        public String getName() {
+                writer.write(configItem.getContent());
-            return name;
+                writer.flush();
-        }
+                logger.info("## Loaded remote instance config: {}/instance.properties ", configItem.getName());
-
+            } catch (Exception e) {
-        public void setName(String name) {
+                logger.error(e.getMessage(), e);
-            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<T> {
+        @Override
-
+        public void onDelete(String instanceName) {
-        void onChange(T properties);
+            File file = new File(getConfPath() + instanceName + "/");
+            if (file.exists()) {
+                deleteDir(file);
+                logger.info("## Deleted and loaded remote instance config: {} ", instanceName);
+            }
+        }
     }
 }

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

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

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

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