From ae474d67ad874241e9e224ec19000fccbb2a6af4 Mon Sep 17 00:00:00 2001 From: "wurongjun@duiba.com.cn" <775105396@qq.com> Date: Mon, 25 Feb 2019 17:53:30 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=BD=93=E9=80=82=E9=85=8D?= =?UTF-8?q?=E5=99=A8mapping=E6=96=87=E4=BB=B6=E4=BF=AE=E6=94=B9=E6=97=B6?= =?UTF-8?q?=EF=BC=8C=E6=97=A0=E6=B3=95=E7=94=9F=E6=95=88=E7=9A=84=E9=97=AE?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../adapter/rdb/monitor/RdbConfigMonitor.java | 338 +++++++++--------- 1 file changed, 169 insertions(+), 169 deletions(-) diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java index 72529a37..779dc3b0 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java @@ -1,169 +1,169 @@ -package com.alibaba.otter.canal.client.adapter.rdb.monitor; - -import java.io.File; -import java.util.HashMap; -import java.util.Map; - -import org.apache.commons.io.filefilter.FileFilterUtils; -import org.apache.commons.io.monitor.FileAlterationListenerAdaptor; -import org.apache.commons.io.monitor.FileAlterationMonitor; -import org.apache.commons.io.monitor.FileAlterationObserver; -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.rdb.RdbAdapter; -import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; -import com.alibaba.otter.canal.client.adapter.rdb.config.MirrorDbConfig; -import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader; -import com.alibaba.otter.canal.client.adapter.support.Util; - -public class RdbConfigMonitor { - - private static final Logger logger = LoggerFactory.getLogger(RdbConfigMonitor.class); - - private static final String adapterName = "rdb"; - - private String key; - - private RdbAdapter rdbAdapter; - - private FileAlterationMonitor fileMonitor; - - public void init(String key, RdbAdapter rdbAdapter) { - this.key = key; - this.rdbAdapter = rdbAdapter; - File confDir = Util.getConfDirPath(adapterName); - try { - FileAlterationObserver observer = new FileAlterationObserver(confDir, - FileFilterUtils.and(FileFilterUtils.fileFileFilter(), FileFilterUtils.suffixFileFilter("yml"))); - FileListener listener = new FileListener(); - observer.addListener(listener); - fileMonitor = new FileAlterationMonitor(3000, observer); - fileMonitor.start(); - - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - public void destroy() { - try { - fileMonitor.stop(); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - private class FileListener extends FileAlterationListenerAdaptor { - - @Override - public void onFileCreate(File file) { - super.onFileCreate(file); - try { - // 加载新增的配置文件 - String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName()); - MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class); - config.validate(); - if ((key == null && config.getOuterAdapterKey() == null) - || (key != null && key.equals(config.getOuterAdapterKey()))) { - addConfigToCache(file, config); - - logger.info("Add a new rdb mapping config: {} to canal adapter", file.getName()); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - @Override - public void onFileChange(File file) { - super.onFileChange(file); - - try { - if (rdbAdapter.getRdbMapping().containsKey(file.getName())) { - // 加载配置文件 - String configContent = MappingConfigsLoader - .loadConfig(adapterName + File.separator + file.getName()); - if (configContent == null) { - onFileDelete(file); - return; - } - MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class); - config.validate(); - if ((key == null && config.getOuterAdapterKey() == null) - || (key != null && key.equals(config.getOuterAdapterKey()))) { - if (rdbAdapter.getRdbMapping().containsKey(file.getName())) { - deleteConfigFromCache(file); - } - addConfigToCache(file, config); - } else { - // 不能修改outerAdapterKey - throw new RuntimeException("Outer adapter key not allowed modify"); - } - logger.info("Change a rdb mapping config: {} of canal adapter", file.getName()); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - @Override - public void onFileDelete(File file) { - super.onFileDelete(file); - - try { - if (rdbAdapter.getRdbMapping().containsKey(file.getName())) { - deleteConfigFromCache(file); - - logger.info("Delete a rdb mapping config: {} of canal adapter", file.getName()); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - private void addConfigToCache(File file, MappingConfig mappingConfig) { - if (mappingConfig == null || mappingConfig.getDbMapping() == null) { - return; - } - rdbAdapter.getRdbMapping().put(file.getName(), mappingConfig); - if (!mappingConfig.getDbMapping().getMirrorDb()) { - Map configMap = rdbAdapter.getMappingConfigCache() - .computeIfAbsent(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." - + mappingConfig.getDbMapping().getDatabase() + "." - + mappingConfig.getDbMapping().getTable(), - k1 -> new HashMap<>()); - configMap.put(file.getName(), mappingConfig); - } else { - Map mirrorDbConfigCache = rdbAdapter.getMirrorDbConfigCache(); - mirrorDbConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." - + mappingConfig.getDbMapping().getDatabase(), - MirrorDbConfig.create(file.getName(), mappingConfig)); - } - } - - private void deleteConfigFromCache(File file) { - MappingConfig mappingConfig = rdbAdapter.getRdbMapping().remove(file.getName()); - - if (mappingConfig == null || mappingConfig.getDbMapping() == null) { - return; - } - if (!mappingConfig.getDbMapping().getMirrorDb()) { - for (Map configMap : rdbAdapter.getMappingConfigCache().values()) { - if (configMap != null) { - configMap.remove(file.getName()); - } - } - } else { - rdbAdapter.getMirrorDbConfigCache().forEach((key, mirrorDbConfig) -> { - if (mirrorDbConfig.getFileName().equals(file.getName())) { - rdbAdapter.getMirrorDbConfigCache().remove(key); - } - }); - } - - } - } -} +package com.alibaba.otter.canal.client.adapter.rdb.monitor; + +import java.io.File; +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.io.filefilter.FileFilterUtils; +import org.apache.commons.io.monitor.FileAlterationListenerAdaptor; +import org.apache.commons.io.monitor.FileAlterationMonitor; +import org.apache.commons.io.monitor.FileAlterationObserver; +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.rdb.RdbAdapter; +import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; +import com.alibaba.otter.canal.client.adapter.rdb.config.MirrorDbConfig; +import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader; +import com.alibaba.otter.canal.client.adapter.support.Util; + +public class RdbConfigMonitor { + + private static final Logger logger = LoggerFactory.getLogger(RdbConfigMonitor.class); + + private static final String adapterName = "rdb"; + + private String key; + + private RdbAdapter rdbAdapter; + + private FileAlterationMonitor fileMonitor; + + public void init(String key, RdbAdapter rdbAdapter) { + this.key = key; + this.rdbAdapter = rdbAdapter; + File confDir = Util.getConfDirPath(adapterName); + try { + FileAlterationObserver observer = new FileAlterationObserver(confDir, + FileFilterUtils.and(FileFilterUtils.fileFileFilter(), FileFilterUtils.suffixFileFilter("yml"))); + FileListener listener = new FileListener(); + observer.addListener(listener); + fileMonitor = new FileAlterationMonitor(3000, observer); + fileMonitor.start(); + + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + public void destroy() { + try { + fileMonitor.stop(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + private class FileListener extends FileAlterationListenerAdaptor { + + @Override + public void onFileCreate(File file) { + super.onFileCreate(file); + try { + // 加载新增的配置文件 + String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName()); + MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class); + config.validate(); + if ((key == null && config.getOuterAdapterKey() == null) + || (key != null && key.equals(config.getOuterAdapterKey()))) { + addConfigToCache(file, config); + + logger.info("Add a new rdb mapping config: {} to canal adapter", file.getName()); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + @Override + public void onFileChange(File file) { + super.onFileChange(file); + + try { + if (rdbAdapter.getRdbMapping().containsKey(file.getName())) { + // 加载配置文件 + String configContent = MappingConfigsLoader + .loadConfig(adapterName + File.separator + file.getName()); + if (configContent == null) { + onFileDelete(file); + return; + } + MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class); + config.validate(); + if ((key == null && config.getOuterAdapterKey() == null) + || (key != null && key.equals(config.getOuterAdapterKey()))) { + if (rdbAdapter.getRdbMapping().containsKey(file.getName())) { + deleteConfigFromCache(file); + } + addConfigToCache(file, config); + } else { + // 不能修改outerAdapterKey + throw new RuntimeException("Outer adapter key not allowed modify"); + } + logger.info("Change a rdb mapping config: {} of canal adapter", file.getName()); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + @Override + public void onFileDelete(File file) { + super.onFileDelete(file); + + try { + if (rdbAdapter.getRdbMapping().containsKey(file.getName())) { + deleteConfigFromCache(file); + + logger.info("Delete a rdb mapping config: {} of canal adapter", file.getName()); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + private void addConfigToCache(File file, MappingConfig mappingConfig) { + if (mappingConfig == null || mappingConfig.getDbMapping() == null) { + return; + } + rdbAdapter.getRdbMapping().put(file.getName(), mappingConfig); + if (!mappingConfig.getDbMapping().getMirrorDb()) { + Map configMap = rdbAdapter.getMappingConfigCache() + .computeIfAbsent(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "_" + + mappingConfig.getDbMapping().getDatabase() + "-" + + mappingConfig.getDbMapping().getTable(), + k1 -> new HashMap<>()); + configMap.put(file.getName(), mappingConfig); + } else { + Map mirrorDbConfigCache = rdbAdapter.getMirrorDbConfigCache(); + mirrorDbConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "." + + mappingConfig.getDbMapping().getDatabase(), + MirrorDbConfig.create(file.getName(), mappingConfig)); + } + } + + private void deleteConfigFromCache(File file) { + MappingConfig mappingConfig = rdbAdapter.getRdbMapping().remove(file.getName()); + + if (mappingConfig == null || mappingConfig.getDbMapping() == null) { + return; + } + if (!mappingConfig.getDbMapping().getMirrorDb()) { + for (Map configMap : rdbAdapter.getMappingConfigCache().values()) { + if (configMap != null) { + configMap.remove(file.getName()); + } + } + } else { + rdbAdapter.getMirrorDbConfigCache().forEach((key, mirrorDbConfig) -> { + if (mirrorDbConfig.getFileName().equals(file.getName())) { + rdbAdapter.getMirrorDbConfigCache().remove(key); + } + }); + } + + } + } +}