adapter 主配置application.yml动态加载

client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Constant.java

@@ -0,0 +1,6 @@
+package com.alibaba.otter.canal.client.adapter.support;
+
+public class Constant {
+
+    public static final String CONF_DIR = "conf";
+}

client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java

@@ -15,7 +15,7 @@ public class MappingConfigsLoader {
         Map<String, String> configContentMap = new HashMap<>();
 
         // 先取本地文件，再取类路径
-        File configDir = new File("../conf/" + name);
+        File configDir = new File(".." + File.separator + Constant.CONF_DIR + File.separator + name);
         if (!configDir.exists()) {
             URL url = MappingConfigsLoader.class.getClassLoader().getResource("");
             if (url != null) {

client-adapter/launcher/pom.xml

@@ -49,6 +49,11 @@
             <artifactId>spring-boot-configuration-processor</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-context</artifactId>
+            <version>2.0.0.RELEASE</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.curator</groupId>
             <artifactId>curator-recipes</artifactId>

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java

@@ -1,5 +1,7 @@
 package com.alibaba.otter.canal.adapter.launcher;
 
+import org.springframework.boot.Banner;
+import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 
@@ -13,6 +15,8 @@ import org.springframework.boot.builder.SpringApplicationBuilder;
 public class CanalAdapterApplication {
 
     public static void main(String[] args) {
-        new SpringApplicationBuilder(CanalAdapterApplication.class).run(args);
+        SpringApplication application = new SpringApplication(CanalAdapterApplication.class);
+        application.setBannerMode(Banner.Mode.OFF);
+        application.run(args);
     }
 }

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/RefresherConfig.java

@@ -0,0 +1,22 @@
+package com.alibaba.otter.canal.adapter.launcher.config;
+
+import org.springframework.cloud.context.refresh.ContextRefresher;
+import org.springframework.cloud.context.scope.refresh.RefreshScope;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class RefresherConfig {
+
+    @Bean
+    public RefreshScope refreshScope() {
+        return new RefreshScope();
+    }
+
+    @Bean
+    public ContextRefresher contextRefresher(ConfigurableApplicationContext configurableApplicationContext,
+                                             RefreshScope refreshScope) {
+        return new ContextRefresher(configurableApplicationContext, refreshScope);
+    }
+}

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java

@@ -6,8 +6,10 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
@@ -136,7 +138,6 @@ public class CanalAdapterLoader {
             logger.info("Load canal adapter: {} succeed", config.getName());
         } catch (Exception e) {
             logger.error("Load canal adapter: {} failed", config.getName(), e);
-            System.exit(0);
         }
     }
 
@@ -144,19 +145,41 @@ public class CanalAdapterLoader {
      * 销毁所有适配器 为防止canal实例太多造成销毁阻塞, 并行销毁
      */
     public void destroy() {
-        if (canalWorkers.size() > 0) {
+        if (!canalWorkers.isEmpty()) {
             ExecutorService stopExecutorService = Executors.newFixedThreadPool(canalWorkers.size());
+            List<Future<Boolean>> futures = new ArrayList<>();
             for (CanalAdapterWorker canalAdapterWorker : canalWorkers.values()) {
-                stopExecutorService.submit(canalAdapterWorker::stop);
+                futures.add(stopExecutorService.submit(() -> {
+                    canalAdapterWorker.stop();
+                    return true;
+                }));
             }
+            futures.forEach(future -> {
+                try {
+                    future.get();
+                } catch (Exception e) {
+                    // ignore
+                }
+            });
             stopExecutorService.shutdown();
         }
 
-        if (canalMQWorker.size() > 0) {
+        if (!canalMQWorker.isEmpty()) {
             ExecutorService stopMQWorkerService = Executors.newFixedThreadPool(canalMQWorker.size());
+            List<Future<Boolean>> futures = new ArrayList<>();
             for (AbstractCanalAdapterWorker canalAdapterMQWorker : canalMQWorker.values()) {
-                stopMQWorkerService.submit(canalAdapterMQWorker::stop);
+                futures.add(stopMQWorkerService.submit(() -> {
+                    canalAdapterMQWorker.stop();
+                    return true;
+                }));
             }
+            futures.forEach(future -> {
+                try {
+                    future.get();
+                } catch (Exception e) {
+                    // ignore
+                }
+            });
             stopMQWorkerService.shutdown();
         }
         logger.info("All canal adapters destroyed");

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java

@@ -4,14 +4,15 @@ import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 import javax.annotation.Resource;
 
-import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch;
-import com.alibaba.otter.canal.adapter.launcher.config.SpringContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.cloud.context.config.annotation.RefreshScope;
 import org.springframework.stereotype.Component;
 
 import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch;
 import com.alibaba.otter.canal.adapter.launcher.config.AdapterCanalConfig;
+import com.alibaba.otter.canal.adapter.launcher.config.SpringContext;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 
 /**
@@ -21,42 +22,51 @@ import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
  * @version 1.0.0
  */
 @Component
+@RefreshScope
 public class CanalAdapterService {
 
-    private static final Logger       logger = LoggerFactory.getLogger(CanalAdapterService.class);
+    private static final Logger logger  = LoggerFactory.getLogger(CanalAdapterService.class);
 
-    private static CanalAdapterLoader adapterLoader;
+    private CanalAdapterLoader  adapterLoader;
 
     @Resource
-    private AdapterCanalConfig        adapterCanalConfig;
+    private AdapterCanalConfig  adapterCanalConfig;
 
     // 注入bean保证优先注册
     @Resource
-    private SpringContext             springContext;
+    private SpringContext       springContext;
     @Resource
-    private SyncSwitch                syncSwitch;
+    private SyncSwitch          syncSwitch;
+
+    private volatile boolean    running = false;
 
     @PostConstruct
-    public void init() {
+    public synchronized void init() {
-        if (adapterLoader == null) {
+        if (running) {
-            try {
+            return;
-                logger.info("## start the canal client adapters.");
+        }
-                adapterLoader = new CanalAdapterLoader(adapterCanalConfig);
+        try {
-                adapterLoader.init();
+            logger.info("## start the canal client adapters.");
-                logger.info("## the canal client adapters are running now ......");
+            adapterLoader = new CanalAdapterLoader(adapterCanalConfig);
-            } catch (Throwable e) {
+            adapterLoader.init();
-                logger.error("## something goes wrong when starting up the canal client adapters:", e);
+            running = true;
-                System.exit(0);
+            logger.info("## the canal client adapters are running now ......");
-            }
+        } catch (Exception e) {
+            logger.error("## something goes wrong when starting up the canal client adapters:", e);
         }
     }
 
     @PreDestroy
-    public void destroy() {
+    public synchronized void destroy() {
+        if (!running) {
+            return;
+        }
         try {
+            running = false;
             logger.info("## stop the canal client adapters");
             if (adapterLoader != null) {
                 adapterLoader.destroy();
+                adapterLoader = null;
             }
             for (DruidDataSource druidDataSource : DatasourceConfig.DATA_SOURCES.values()) {
                 try {
@@ -65,6 +75,7 @@ public class CanalAdapterService {
                     logger.error(e.getMessage(), e);
                 }
             }
+            DatasourceConfig.DATA_SOURCES.clear();
         } catch (Throwable e) {
             logger.warn("## something goes wrong when stopping canal client adapters:", e);
         } finally {

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationRunningMonitor.java

@@ -0,0 +1,94 @@
+package com.alibaba.otter.canal.adapter.launcher.monitor;
+
+import java.io.File;
+import java.io.FileReader;
+import java.net.URL;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import javax.annotation.Resource;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.cloud.context.refresh.ContextRefresher;
+import org.springframework.stereotype.Component;
+import org.yaml.snakeyaml.Yaml;
+
+import com.alibaba.otter.canal.adapter.launcher.loader.CanalAdapterService;
+import com.alibaba.otter.canal.client.adapter.support.Constant;
+
+@Component
+public class ApplicationRunningMonitor {
+
+    private static final Logger      logger                = LoggerFactory.getLogger(ApplicationRunningMonitor.class);
+
+    @Resource
+    private ContextRefresher         contextRefresher;
+
+    @Resource
+    private CanalAdapterService      canalAdapterService;
+
+    private ScheduledExecutorService scheduled             = Executors.newScheduledThreadPool(1);
+
+    private static volatile long     applicationLastModify = -1L;
+
+    @PostConstruct
+    public void init() {
+        URL url = this.getClass().getClassLoader().getResource("");
+        String path;
+        if (url != null) {
+            path = url.getPath();
+        } else {
+            path = new File("").getAbsolutePath();
+        }
+        File file = null;
+        if (path != null) {
+            file = new File(path + ".." + File.separator + Constant.CONF_DIR + File.separator + "application.yml");
+            if (!file.exists()) {
+                file = new File(path + "application.yml");
+            }
+        }
+        if (file == null || !file.exists()) {
+            throw new RuntimeException("application.yml config file not found.");
+        }
+        File configFile = file;
+        scheduled.scheduleWithFixedDelay(() -> {
+            if (applicationLastModify == -1L) {
+                applicationLastModify = configFile.lastModified();
+            } else {
+                if (configFile.lastModified() != applicationLastModify) {
+                    applicationLastModify = configFile.lastModified();
+
+                    try {
+                        // 检查yml格式
+                        new Yaml().loadAs(new FileReader(configFile), Map.class);
+
+                        canalAdapterService.destroy();
+
+                        // refresh context
+                        contextRefresher.refresh();
+
+                        try {
+                            Thread.sleep(2000);
+                        } catch (InterruptedException e) {
+                            // ignore
+                        }
+                        canalAdapterService.init();
+                        logger.info("## adapter application config reloaded.");
+                    } catch (Exception e) {
+                        logger.error(e.getMessage(), e);
+                    }
+                }
+            }
+        }, 5, 5, TimeUnit.SECONDS);
+    }
+
+    @PreDestroy
+    public void destroy() {
+        scheduled.shutdown();
+    }
+}

client-adapter/launcher/src/main/resources/application.yml

@@ -2,6 +2,7 @@ server:
   port: 8081
 logging:
   level:
+#    org.springframework: WARN
     com.alibaba.otter.canal.client.adapter.hbase: DEBUG
     com.alibaba.otter.canal.client.adapter.es: DEBUG
     com.alibaba.otter.canal.client.adapter.rdb: DEBUG