Merge pull request #1475 from rewerma/master

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

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 {
+        remoteConfigLoader = RemoteConfigLoaderFactory.getRemoteConfigLoader(env);
-            // 加载远程配置
+        if (remoteConfigLoader != null) {
-            String jdbcUrl = env.getProperty("canal.manager.jdbc.url");
+            remoteConfigLoader.loadRemoteConfig();
-            if (StringUtils.isNotEmpty(jdbcUrl)) {
+            remoteConfigLoader.loadRemoteAdapterConfigs();
-                String jdbcUsername = env.getProperty("canal.manager.jdbc.username");
+            remoteConfigLoader.startMonitor(); // 启动监听
-                String jdbcPassword = env.getProperty("canal.manager.jdbc.password");
+        }
-                AdapterRemoteConfigMonitor configMonitor = new AdapterRemoteConfigMonitor(jdbcUrl,
+    }
-                    jdbcUsername,
+
-                    jdbcPassword);
+    @PreDestroy
-                configMonitor.loadRemoteConfig();
+    public synchronized void destroy() {
-                configMonitor.loadRemoteAdapterConfigs();
+        if (remoteConfigLoader != null) {
-                configMonitor.start(); // 启动监听
+            remoteConfigLoader.destroy();
-            }
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
         }
     }
 }

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 volatile boolean    running = false;
-
-    private AdapterRemoteConfigMonitor configMonitor = null;
 
     @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();

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

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.Connection;
-import java.sql.*;
+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.druid.pool.DruidDataSource;
-import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
 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 DruidDataSource          dataSource;
-    private String                   jdbcUrl;
-    private String                   jdbcUsername;
-    private String                   jdbcPassword;
 
-    private long                     currentConfigTimestamp = 0;
+    private static volatile long     currentConfigTimestamp = 0;
     private Map<String, ConfigItem>  remoteAdapterConfigs   = new MapMaker().makeMap();
 
     private ScheduledExecutorService executor               = Executors.newScheduledThreadPool(2,
         new NamedThreadFactory("remote-adapter-config-scan"));
 
-    public AdapterRemoteConfigMonitor(String jdbcUrl, String jdbcUsername, String jdbcPassword){
+    private RemoteAdapterMonitor     remoteAdapterMonitor   = new RemoteAdapterMonitorImpl();
-        this.jdbcUrl = jdbcUrl;
-        this.jdbcUsername = jdbcUsername;
-        this.jdbcPassword = jdbcPassword;
-    }
 
-    private Connection getConn() throws Exception {
+    public DbRemoteConfigLoader(String driverName, String jdbcUrl, String jdbcUsername, String jdbcPassword){
-        if (conn == null || conn.isClosed()) {
+        dataSource = new DruidDataSource();
-            Class.forName("com.mysql.jdbc.Driver");
+        if (StringUtils.isEmpty(driverName)) {
-            conn = DriverManager.getConnection(jdbcUrl, jdbcUsername, jdbcPassword);
+            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 {
     }
 
     /**
-     * 启动监听数据库变化
+     * 加载adapter配置
-     */
-    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配置到本地
      */
+    @Override
     public void loadRemoteAdapterConfigs() {
         try {
             // 加载远程adapter配置
-            Map<String, ConfigItem>[] modifiedConfigs = getModifiedAdapterConfigs();
+            loadModifiedAdapterConfigs();
-            if (modifiedConfigs != null) {
-                overrideLocalAdapterConfigs(modifiedConfigs);
-            }
         } catch (Exception e) {
             logger.error(e.getMessage(), e);
         }
     }
 
     /**
-     * 获取有变动的adapter配置
+     * 加载有变动的adapter配置
-     *
-     * @return Map[0]: 新增修改的配置, Map[1]: 删除的配置
      */
     @SuppressWarnings("unchecked")
-    private Map<String, ConfigItem>[] getModifiedAdapterConfigs() {
+    private void loadModifiedAdapterConfigs() {
         Map<String, ConfigItem>[] res = new Map[2];
         Map<String, ConfigItem> 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<String, ConfigItem> 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(),
+                        remoteAdapterMonitor.onModify(configItemNew);
-                            configItemNew);
                     }
 
-                    res[0] = changedAdapterConfig.isEmpty() ? null : changedAdapterConfig;
                 } catch (Exception e) {
                     logger.error(e.getMessage(), e);
                 }
             }
         }
 
-        Map<String, ConfigItem> 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);
+                remoteAdapterMonitor.onDelete(configItem.getCategory() + "/" + configItem.getName());
-            }
-        }
-        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<String, ConfigItem>[] modifiedAdapterConfigs) {
-        Map<String, ConfigItem> 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<String, ConfigItem> 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);
-                }
             }
         }
     }
@@ -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();
+                loadRemoteConfig();
-            } catch (SQLException e) {
+            } catch (Throwable e) {
-                logger.error(e.getMessage(), 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 class RemoteAdapterMonitorImpl implements RemoteAdapterMonitor {
-        private String category;
-        private String name;
-        private String content;
-        private long   modifiedTime;
 
-        public Long getId() {
+        @Override
-            return id;
+        public void onAdd(ConfigItem configItem) {
+            this.onModify(configItem);
         }
 
-        public void setId(Long id) {
+        @Override
-            this.id = id;
+        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() {
+        @Override
-            return category;
+        public void onDelete(String name) {
-        }
+            File file = new File(getConfPath() + name);
-
+            if (file.exists()) {
-        public void setCategory(String category) {
+                deleteDir(file);
-            this.category = category;
+                logger.info("## Deleted and reloaded remote adapter config: {}", name);
-        }
+            }
-
-        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;
         }
     }
 }

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

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

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

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<String, Message> kafkaConsumer;
-    protected KafkaConsumer<String, String>  kafkaConsumer2;                  // 用于扁平message的数据消费
+    protected KafkaConsumer<String, String>  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<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";
-    private Connection getConn() throws Exception {
+        }
-        if (conn == null || conn.isClosed()) {
+        dataSource.setDriverClassName(driverName);
-            Class.forName("com.mysql.jdbc.Driver");
+        dataSource.setUrl(jdbcUrl);
-            conn = DriverManager.getConnection(jdbcUrl, jdbcUsername, jdbcPassword);
+        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() {
+    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,12 +274,10 @@ public class ManagerRemoteConfigMonitor {
      */
     public void destroy() {
         executor.shutdownNow();
-        if (conn != null) {
+        try {
-            try {
+            dataSource.close();
-                conn.close();
+        } catch (Exception e) {
-            } catch (SQLException e) {
+            logger.error(e.getMessage(), 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;
+        @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")) {
+                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() {
+        @Override
-            return name;
+        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<T> {
-
-        void onChange(T properties);
     }
 }

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