diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Constant.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Constant.java new file mode 100644 index 00000000..9a75f765 --- /dev/null +++ b/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"; +} diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java index 711a52b6..c437731b 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java @@ -15,7 +15,7 @@ public class MappingConfigsLoader { Map 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) { diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index df31b8c8..6a313934 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -49,6 +49,11 @@ spring-boot-configuration-processor provided + + org.springframework.cloud + spring-cloud-context + 2.0.0.RELEASE + org.apache.curator curator-recipes diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java index 99140b0c..83f4b350 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java +++ b/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); } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/RefresherConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/RefresherConfig.java new file mode 100644 index 00000000..127b7ceb --- /dev/null +++ b/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); + } +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index e2d9bea8..900d1207 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/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> 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> 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"); diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java index 15dafa79..79eb0be1 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java @@ -4,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() { - if (adapterLoader == null) { - try { - logger.info("## start the canal client adapters."); - adapterLoader = new CanalAdapterLoader(adapterCanalConfig); - adapterLoader.init(); - logger.info("## the canal client adapters are running now ......"); - } catch (Throwable e) { - logger.error("## something goes wrong when starting up the canal client adapters:", e); - System.exit(0); - } + public synchronized void init() { + if (running) { + return; + } + try { + logger.info("## start the canal client adapters."); + adapterLoader = new CanalAdapterLoader(adapterCanalConfig); + adapterLoader.init(); + running = true; + 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 { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationRunningMonitor.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationRunningMonitor.java new file mode 100644 index 00000000..b1be1b76 --- /dev/null +++ b/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(); + } +} diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index e5d4f507..f862bb0e 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/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