diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java index 3b19b39e..566a9fe0 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java @@ -84,6 +84,10 @@ public class ESConfigMonitor { // 加载配置文件 String configContent = MappingConfigsLoader .loadConfig(adapterName + File.separator + file.getName()); + if (configContent == null) { + onFileDelete(file); + return; + } ESSyncConfig config = new Yaml().loadAs(configContent, ESSyncConfig.class); config.validate(); if (esAdapter.getEsSyncConfig().containsKey(file.getName())) { diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/monitor/HbaseConfigMonitor.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/monitor/HbaseConfigMonitor.java index ae6ddef2..da16ee2e 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/monitor/HbaseConfigMonitor.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/monitor/HbaseConfigMonitor.java @@ -60,6 +60,10 @@ public class HbaseConfigMonitor { try { // 加载新增的配置文件 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(); addConfigToCache(file, config); diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/BootstrapConfiguration.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/BootstrapConfiguration.java new file mode 100644 index 00000000..3f13c00c --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/BootstrapConfiguration.java @@ -0,0 +1,44 @@ +package com.alibaba.otter.canal.adapter.launcher.config; + +import javax.annotation.PostConstruct; + +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; + +/** + * Bootstrap级别配置加载 + * + * @author rewerma @ 2019-01-05 + * @version 1.0.0 + */ +public class BootstrapConfiguration { + + private static final Logger logger = LoggerFactory.getLogger(BootstrapConfiguration.class); + + @Autowired + private Environment env; + + @PostConstruct + public void loadRemoteConfig() { + try { + // 加载远程配置 + String jdbcUrl = env.getProperty("canal.manager.jdbc.url"); + if (StringUtils.isNotEmpty(jdbcUrl)) { + String jdbcUsername = env.getProperty("canal.manager.jdbc.username"); + String jdbcPassword = env.getProperty("canal.manager.jdbc.password"); + AdapterRemoteConfigMonitor configMonitor = new AdapterRemoteConfigMonitor(jdbcUrl, + jdbcUsername, + jdbcPassword); + configMonitor.loadRemoteConfig(); + configMonitor.loadRemoteAdapterConfigs(); + configMonitor.start(); // 启动监听 + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } +} 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 79eb0be1..e2f03470 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,9 +4,13 @@ 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; +import org.springframework.cloud.context.refresh.ContextRefresher; +import org.springframework.core.env.Environment; import org.springframework.stereotype.Component; import com.alibaba.druid.pool.DruidDataSource; @@ -25,20 +29,27 @@ 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 AdapterCanalConfig adapterCanalConfig; + private ContextRefresher contextRefresher; + + @Resource + private AdapterCanalConfig adapterCanalConfig; + @Resource + 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() { @@ -64,6 +75,10 @@ public class CanalAdapterService { try { running = false; logger.info("## stop the canal client adapters"); + if (configMonitor != null) { + configMonitor.destroy(); + } + if (adapterLoader != null) { adapterLoader.destroy(); adapterLoader = null; diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/AdapterRemoteConfigMonitor.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/AdapterRemoteConfigMonitor.java new file mode 100644 index 00000000..1b9b5e95 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/AdapterRemoteConfigMonitor.java @@ -0,0 +1,342 @@ +package com.alibaba.otter.canal.adapter.launcher.monitor; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileWriter; +import java.net.URL; +import java.sql.*; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +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.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 + * @version 1.0.0 + */ +public class AdapterRemoteConfigMonitor { + + private static final Logger logger = LoggerFactory.getLogger(AdapterRemoteConfigMonitor.class); + + private Connection conn; + private String jdbcUrl; + private String jdbcUsername; + private String jdbcPassword; + + private long currentConfigTimestamp = 0; + private Map remoteAdapterConfigs = new MapMaker().makeMap(); + + private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2, + new NamedThreadFactory("remote-adapter-config-scan")); + + public AdapterRemoteConfigMonitor(String jdbcUrl, String jdbcUsername, String jdbcPassword){ + this.jdbcUrl = jdbcUrl; + this.jdbcUsername = jdbcUsername; + this.jdbcPassword = jdbcPassword; + } + + private Connection getConn() throws Exception { + if (conn == null || conn.isClosed()) { + Class.forName("com.mysql.jdbc.Driver"); + conn = DriverManager.getConnection(jdbcUrl, jdbcUsername, jdbcPassword); + } + return conn; + } + + /** + * 加载远程application.yml配置到本地 + */ + public void loadRemoteConfig() { + try { + // 加载远程adapter配置 + ConfigItem configItem = getRemoteAdapterConfig(); + if (configItem != null) { + if (configItem.getModifiedTime() != currentConfigTimestamp) { + currentConfigTimestamp = configItem.getModifiedTime(); + overrideLocalCanalConfig(configItem.getContent()); + } + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + /** + * 获取远程application.yml配置 + * + * @return 配置对象 + */ + 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)) { + if (rs.next()) { + ConfigItem configItem = new ConfigItem(); + configItem.setId(2L); + configItem.setName(rs.getString("name")); + configItem.setContent(rs.getString("content")); + configItem.setModifiedTime(rs.getTimestamp("modified_time").getTime()); + return configItem; + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + return null; + } + + /** + * 覆盖本地application.yml文件 + * + * @param content 文件内容 + */ + private void overrideLocalCanalConfig(String content) { + try (FileWriter writer = new FileWriter(getConfPath() + "application.yml")) { + writer.write(content); + writer.flush(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + /** + * 启动监听数据库变化 + */ + 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配置到本地 + */ + public void loadRemoteAdapterConfigs() { + try { + // 加载远程instance配置 + Map[] modifiedConfigs = getModifiedAdapterConfigs(); + if (modifiedConfigs != null) { + overrideLocalAdapterConfigs(modifiedConfigs); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + /** + * 获取有变动的adapter配置 + * + * @return Map[0]: 新增修改的配置, Map[1]: 删除的配置 + */ + @SuppressWarnings("unchecked") + private Map[] getModifiedAdapterConfigs() { + Map[] res = new Map[2]; + Map 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)) { + while (rs.next()) { + ConfigItem configItem = new ConfigItem(); + configItem.setId(rs.getLong("id")); + configItem.setCategory(rs.getString("category")); + configItem.setName(rs.getString("name")); + configItem.setModifiedTime(rs.getTimestamp("modified_time").getTime()); + remoteConfigStatus.put(configItem.getCategory() + "/" + configItem.getName(), configItem); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + return null; + } + + if (!remoteConfigStatus.isEmpty()) { + List changedIds = new ArrayList<>(); + + for (ConfigItem remoteConfigStat : remoteConfigStatus.values()) { + ConfigItem currentConfig = remoteAdapterConfigs + .get(remoteConfigStat.getCategory() + "/" + remoteConfigStat.getName()); + if (currentConfig == null) { + // 新增 + changedIds.add(remoteConfigStat.getId()); + } else { + // 修改 + if (currentConfig.getModifiedTime() != remoteConfigStat.getModifiedTime()) { + changedIds.add(remoteConfigStat.getId()); + } + } + } + if (!changedIds.isEmpty()) { + Map changedInstanceConfig = 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)) { + while (rs.next()) { + ConfigItem configItemNew = new ConfigItem(); + configItemNew.setId(rs.getLong("id")); + configItemNew.setCategory(rs.getString("category")); + configItemNew.setName(rs.getString("name")); + configItemNew.setContent(rs.getString("content")); + configItemNew.setModifiedTime(rs.getTimestamp("modified_time").getTime()); + + remoteAdapterConfigs.put(configItemNew.getCategory() + "/" + configItemNew.getName(), + configItemNew); + changedInstanceConfig.put(configItemNew.getCategory() + "/" + configItemNew.getName(), + configItemNew); + } + + res[0] = changedInstanceConfig.isEmpty() ? null : changedInstanceConfig; + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + } + + Map removedInstanceConfig = new HashMap<>(); + for (ConfigItem configItem : remoteAdapterConfigs.values()) { + if (!remoteConfigStatus.containsKey(configItem.getCategory() + "/" + configItem.getName())) { + // 删除 + remoteAdapterConfigs.remove(configItem.getCategory() + "/" + configItem.getName()); + removedInstanceConfig.put(configItem.getCategory() + "/" + configItem.getName(), null); + } + } + res[1] = removedInstanceConfig.isEmpty() ? null : removedInstanceConfig; + + if (res[0] == null && res[1] == null) { + return null; + } else { + return res; + } + } + + /** + * 覆盖adapter配置到本地 + * + * @param modifiedInstanceConfigs 变动的配置集合 + */ + private void overrideLocalAdapterConfigs(Map[] modifiedInstanceConfigs) { + Map changedInstanceConfigs = modifiedInstanceConfigs[0]; + if (changedInstanceConfigs != null) { + for (ConfigItem configItem : changedInstanceConfigs.values()) { + try (FileWriter writer = new FileWriter( + getConfPath() + configItem.getCategory() + "/" + configItem.getName())) { + writer.write(configItem.getContent()); + writer.flush(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + } + Map removedInstanceConfigs = modifiedInstanceConfigs[1]; + if (removedInstanceConfigs != null) { + for (String name : removedInstanceConfigs.keySet()) { + File file = new File(getConfPath() + name); + if (file.exists()) { + file.delete(); + } + } + } + } + + /** + * 获取conf文件夹所在路径 + * + * @return 路径地址 + */ + private String getConfPath() { + String classpath = this.getClass().getResource("/").getPath(); + String confPath = classpath + "../conf/"; + if (new File(confPath).exists()) { + return confPath; + } else { + return classpath; + } + } + + public void destroy() { + executor.shutdownNow(); + if (conn != null) { + try { + conn.close(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } + } + + /** + * 配置对应对象 + */ + public static 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; + } + } +} diff --git a/client-adapter/launcher/src/main/resources/META-INF/spring.factories b/client-adapter/launcher/src/main/resources/META-INF/spring.factories new file mode 100644 index 00000000..29018217 --- /dev/null +++ b/client-adapter/launcher/src/main/resources/META-INF/spring.factories @@ -0,0 +1,3 @@ +# Bootstrap Configuration +org.springframework.cloud.bootstrap.BootstrapConfiguration=\ + com.alibaba.otter.canal.adapter.launcher.config.BootstrapConfiguration \ No newline at end of file diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 6478db45..f08b1df9 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -39,7 +39,7 @@ canal.conf: # key: mysql1 # properties: # jdbc.driverClassName: com.mysql.jdbc.Driver -# jdbc.url: jdbc:mysql://192.168.0.36/mytest?useUnicode=true +# jdbc.url: jdbc:mysql://127.0.0.1:3306/mytest2?useUnicode=true # jdbc.username: root # jdbc.password: 121212 # - name: rdb diff --git a/client-adapter/launcher/src/main/resources/bootstrap.yml b/client-adapter/launcher/src/main/resources/bootstrap.yml new file mode 100644 index 00000000..c36f305f --- /dev/null +++ b/client-adapter/launcher/src/main/resources/bootstrap.yml @@ -0,0 +1,6 @@ +#canal: +# manager: +# jdbc: +# url: jdbc:mysql://127.0.0.1:3306/canal_manager?useUnicode=true&characterEncoding=UTF-8 +# username: root +# password: 121212 \ No newline at end of file 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 e337b829..ddf47138 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 @@ -4,6 +4,7 @@ import java.io.File; import java.util.HashMap; import java.util.Map; +import com.alibaba.otter.canal.client.adapter.rdb.config.MirrorDbConfig; import org.apache.commons.io.filefilter.FileFilterUtils; import org.apache.commons.io.monitor.FileAlterationListenerAdaptor; import org.apache.commons.io.monitor.FileAlterationMonitor; @@ -86,6 +87,10 @@ public class RdbConfigMonitor { // 加载配置文件 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) @@ -121,39 +126,44 @@ public class RdbConfigMonitor { } private void addConfigToCache(File file, MappingConfig mappingConfig) { - MappingConfig.DbMapping dbMapping = mappingConfig.getDbMapping(); - if (dbMapping != null) { - if (!dbMapping.getMirrorDb()) { - rdbAdapter.getRdbMapping().put(file.getName(), mappingConfig); - 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)); - } + 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()); - rdbAdapter.getRdbMapping().remove(file.getName()); - for (Map configMap : rdbAdapter.getMappingConfigCache().values()) { - if (configMap != null) { - configMap.remove(file.getName()); - } + if (mappingConfig == null || mappingConfig.getDbMapping() == null) { + return; } - - rdbAdapter.getMirrorDbConfigCache().forEach((key, mirrorDbConfig) -> { - if (mirrorDbConfig.getFileName().equals(file.getName())) { - rdbAdapter.getMirrorDbConfigCache().remove(key); + 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); + } + }); + } } } diff --git a/deployer/manager_ddl.sql b/deployer/manager_ddl.sql index a5e6dbd4..c91b8ce0 100644 --- a/deployer/manager_ddl.sql +++ b/deployer/manager_ddl.sql @@ -7,7 +7,7 @@ CREATE TABLE `canal_config` ( `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`), UNIQUE KEY `name_UNIQUE` (`name`) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 ; +) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 ; CREATE TABLE `canal_instance_config` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, @@ -16,8 +16,22 @@ CREATE TABLE `canal_instance_config` ( `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`), UNIQUE KEY `name_UNIQUE` (`name`) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 ; +) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 ; + +CREATE TABLE `canal_adapter_config` ( + `id` bigint(20) NOT NULL AUTO_INCREMENT, + `category` varchar(45) NOT NULL, + `name` varchar(45) NOT NULL, + `content` text NOT NULL, + `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`) +) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 ; INSERT INTO `canal_config` VALUES ('1', 'canal.properties', '#################################################\r\n######### common argument ############# \r\n#################################################\r\n#canal.manager.jdbc.url=jdbc:mysql://127.0.0.1:3306/canal_manager?useUnicode=true&characterEncoding=UTF-8\r\n#canal.manager.jdbc.username=root\r\n#canal.manager.jdbc.password=121212\r\ncanal.id = 1\r\ncanal.ip =\r\ncanal.port = 11111\r\ncanal.metrics.pull.port = 11112\r\ncanal.zkServers =\r\n# flush data to zk\r\ncanal.zookeeper.flush.period = 1000\r\ncanal.withoutNetty = false\r\n# tcp, kafka, RocketMQ\r\ncanal.serverMode = tcp\r\n# flush meta cursor/parse position to file\r\ncanal.file.data.dir = ${canal.conf.dir}\r\ncanal.file.flush.period = 1000\r\n## memory store RingBuffer size, should be Math.pow(2,n)\r\ncanal.instance.memory.buffer.size = 16384\r\n## memory store RingBuffer used memory unit size , default 1kb\r\ncanal.instance.memory.buffer.memunit = 1024 \r\n## meory store gets mode used MEMSIZE or ITEMSIZE\r\ncanal.instance.memory.batch.mode = MEMSIZE\r\ncanal.instance.memory.rawEntry = true\r\n\r\n## detecing config\r\ncanal.instance.detecting.enable = false\r\n#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()\r\ncanal.instance.detecting.sql = select 1\r\ncanal.instance.detecting.interval.time = 3\r\ncanal.instance.detecting.retry.threshold = 3\r\ncanal.instance.detecting.heartbeatHaEnable = false\r\n\r\n# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery\r\ncanal.instance.transaction.size = 1024\r\n# mysql fallback connected to new master should fallback times\r\ncanal.instance.fallbackIntervalInSeconds = 60\r\n\r\n# network config\r\ncanal.instance.network.receiveBufferSize = 16384\r\ncanal.instance.network.sendBufferSize = 16384\r\ncanal.instance.network.soTimeout = 30\r\n\r\n# binlog filter config\r\ncanal.instance.filter.druid.ddl = true\r\ncanal.instance.filter.query.dcl = false\r\ncanal.instance.filter.query.dml = false\r\ncanal.instance.filter.query.ddl = false\r\ncanal.instance.filter.table.error = false\r\ncanal.instance.filter.rows = false\r\ncanal.instance.filter.transaction.entry = false\r\n\r\n# binlog format/image check\r\ncanal.instance.binlog.format = ROW,STATEMENT,MIXED \r\ncanal.instance.binlog.image = FULL,MINIMAL,NOBLOB\r\n\r\n# binlog ddl isolation\r\ncanal.instance.get.ddl.isolation = false\r\n\r\n# parallel parser config\r\ncanal.instance.parser.parallel = true\r\n## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()\r\n#canal.instance.parser.parallelThreadSize = 16\r\n## disruptor ringbuffer size, must be power of 2\r\ncanal.instance.parser.parallelBufferSize = 256\r\n\r\n# table meta tsdb info\r\ncanal.instance.tsdb.enable = true\r\ncanal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}\r\ncanal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;\r\ncanal.instance.tsdb.dbUsername = canal\r\ncanal.instance.tsdb.dbPassword = canal\r\n# dump snapshot interval, default 24 hour\r\ncanal.instance.tsdb.snapshot.interval = 24\r\n# purge snapshot expire , default 360 hour(15 days)\r\ncanal.instance.tsdb.snapshot.expire = 360\r\n\r\n# aliyun ak/sk , support rds/mq\r\ncanal.aliyun.accesskey =\r\ncanal.aliyun.secretkey =\r\n\r\n#################################################\r\n######### destinations ############# \r\n#################################################\r\ncanal.destinations = example\r\n# conf root dir\r\ncanal.conf.dir = ../conf\r\n# auto scan instance dir add/remove and start/stop instance\r\ncanal.auto.scan = true\r\ncanal.auto.scan.interval = 5\r\n\r\ncanal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml\r\n#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml\r\n\r\ncanal.instance.global.mode = spring\r\ncanal.instance.global.lazy = false\r\n#canal.instance.global.manager.address = 127.0.0.1:1099\r\n#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml\r\ncanal.instance.global.spring.xml = classpath:spring/file-instance.xml\r\n#canal.instance.global.spring.xml = classpath:spring/default-instance.xml\r\n\r\n##################################################\r\n######### MQ #############\r\n##################################################\r\ncanal.mq.servers = 127.0.0.1:6667\r\ncanal.mq.retries = 0\r\ncanal.mq.batchSize = 16384\r\ncanal.mq.maxRequestSize = 1048576\r\ncanal.mq.lingerMs = 1\r\ncanal.mq.bufferMemory = 33554432\r\ncanal.mq.canalBatchSize = 50\r\ncanal.mq.canalGetTimeout = 100\r\ncanal.mq.flatMessage = true\r\ncanal.mq.compressionType = none\r\ncanal.mq.acks = all\r\n', '2018-12-30 16:46:00'); +INSERT INTO `canal_config` VALUES ('2', 'application.yml', 'server:\n port: 8081\nlogging:\n level:\n org.springframework: WARN\n com.alibaba.otter.canal.client.adapter.hbase: DEBUG\n com.alibaba.otter.canal.client.adapter.es: DEBUG\n com.alibaba.otter.canal.client.adapter.rdb: DEBUG\nspring:\n jackson:\n date-format: yyyy-MM-dd HH:mm:ss\n time-zone: GMT+8\n default-property-inclusion: non_null\n\ncanal.conf:\n canalServerHost: 127.0.0.1:11111\n# zookeeperHosts: slave1:2181\n# mqServers: 127.0.0.1:9092 #or rocketmq\n# flatMessage: true\n batchSize: 500\n syncBatchSize: 1000\n retries: 0\n timeout:\n accessKey:\n secretKey:\n mode: tcp # kafka rocketMQ\n# srcDataSources:\n# defaultDS:\n# url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true\n# username: root\n# password: 121212\n canalAdapters:\n - instance: example # canal instance Name or mq topic name\n groups:\n - groupId: g1\n outerAdapters:\n - name: logger\n# - name: rdb\n# key: mysql1\n# properties:\n# jdbc.driverClassName: com.mysql.jdbc.Driver\n# jdbc.url: jdbc:mysql://127.0.0.1:3306/mytest2?useUnicode=true\n# jdbc.username: root\n# jdbc.password: 121212\n# - name: rdb\n# key: oracle1\n# properties:\n# jdbc.driverClassName: oracle.jdbc.OracleDriver\n# jdbc.url: jdbc:oracle:thin:@localhost:49161:XE\n# jdbc.username: mytest\n# jdbc.password: m121212\n# - name: rdb\n# key: postgres1\n# properties:\n# jdbc.driverClassName: org.postgresql.Driver\n# jdbc.url: jdbc:postgresql://localhost:5432/postgres\n# jdbc.username: postgres\n# jdbc.password: 121212\n# threads: 1\n# commitSize: 3000\n# - name: hbase\n# properties:\n# hbase.zookeeper.quorum: 127.0.0.1\n# hbase.zookeeper.property.clientPort: 2181\n# zookeeper.znode.parent: /hbase\n# - name: es\n# hosts: 127.0.0.1:9300\n# properties:\n# cluster.name: elasticsearch\n', '2019-01-05 01:37:43'); -INSERT INTO `canal_instance_config` VALUES ('1', 'example', '#################################################\r\n## mysql serverId , v1.0.26+ will autoGen\r\n# canal.instance.mysql.slaveId=0\r\n\r\n# enable gtid use true/false\r\ncanal.instance.gtidon=false\r\n\r\n# position info\r\ncanal.instance.master.address=127.0.0.1:3306\r\ncanal.instance.master.journal.name=\r\ncanal.instance.master.position=\r\ncanal.instance.master.timestamp=\r\ncanal.instance.master.gtid=\r\n\r\n# rds oss binlog\r\ncanal.instance.rds.accesskey=\r\ncanal.instance.rds.secretkey=\r\ncanal.instance.rds.instanceId=\r\n\r\n# table meta tsdb info\r\ncanal.instance.tsdb.enable=true\r\n#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb\r\n#canal.instance.tsdb.dbUsername=canal\r\n#canal.instance.tsdb.dbPassword=canal\r\n\r\n#canal.instance.standby.address =\r\n#canal.instance.standby.journal.name =\r\n#canal.instance.standby.position =\r\n#canal.instance.standby.timestamp =\r\n#canal.instance.standby.gtid=\r\n\r\n# username/password\r\ncanal.instance.dbUsername=canal\r\ncanal.instance.dbPassword=canal\r\ncanal.instance.connectionCharset = UTF-8\r\ncanal.instance.defaultDatabaseName =test\r\n# enable druid Decrypt database password\r\ncanal.instance.enableDruid=false\r\n#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==\r\n\r\n# table regex\r\ncanal.instance.filter.regex=.*\\\\..*\r\n# table black regex\r\ncanal.instance.filter.black.regex=\r\n\r\n# mq config\r\ncanal.mq.topic=example\r\n# 动态topic, 需mq支持动态创建topic\r\n#canal.mq.dynamicTopic=.*,mytest\\\\..*,mytest2.user\r\ncanal.mq.partition=0\r\n# hash partition config\r\n#canal.mq.partitionsNum=3\r\n#canal.mq.partitionHash=test.table:id^name,.*\\\\..*\r\n#################################################\r\n', '2018-12-30 16:46:16'); \ No newline at end of file +INSERT INTO `canal_instance_config` VALUES ('1', 'example', '#################################################\r\n## mysql serverId , v1.0.26+ will autoGen\r\n# canal.instance.mysql.slaveId=0\r\n\r\n# enable gtid use true/false\r\ncanal.instance.gtidon=false\r\n\r\n# position info\r\ncanal.instance.master.address=127.0.0.1:3306\r\ncanal.instance.master.journal.name=\r\ncanal.instance.master.position=\r\ncanal.instance.master.timestamp=\r\ncanal.instance.master.gtid=\r\n\r\n# rds oss binlog\r\ncanal.instance.rds.accesskey=\r\ncanal.instance.rds.secretkey=\r\ncanal.instance.rds.instanceId=\r\n\r\n# table meta tsdb info\r\ncanal.instance.tsdb.enable=true\r\n#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb\r\n#canal.instance.tsdb.dbUsername=canal\r\n#canal.instance.tsdb.dbPassword=canal\r\n\r\n#canal.instance.standby.address =\r\n#canal.instance.standby.journal.name =\r\n#canal.instance.standby.position =\r\n#canal.instance.standby.timestamp =\r\n#canal.instance.standby.gtid=\r\n\r\n# username/password\r\ncanal.instance.dbUsername=canal\r\ncanal.instance.dbPassword=canal\r\ncanal.instance.connectionCharset = UTF-8\r\ncanal.instance.defaultDatabaseName =test\r\n# enable druid Decrypt database password\r\ncanal.instance.enableDruid=false\r\n#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==\r\n\r\n# table regex\r\ncanal.instance.filter.regex=.*\\\\..*\r\n# table black regex\r\ncanal.instance.filter.black.regex=\r\n\r\n# mq config\r\ncanal.mq.topic=example\r\n# 动态topic, 需mq支持动态创建topic\r\n#canal.mq.dynamicTopic=.*,mytest\\\\..*,mytest2.user\r\ncanal.mq.partition=0\r\n# hash partition config\r\n#canal.mq.partitionsNum=3\r\n#canal.mq.partitionHash=test.table:id^name,.*\\\\..*\r\n#################################################\r\n', '2018-12-30 16:46:16'); + +INSERT INTO `canal_adapter_config` VALUES ('1', 'es', 'mytest_user.yml', 'dataSourceKey: defaultDS\ndestination: example\nesMapping:\n _index: mytest_user\n _type: _doc\n _id: _id\n# pk: id\n sql: \"select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name,\n a.c_time as _c_time, c.labels as _labels from user a\n left join role b on b.id=a.role_id\n left join (select user_id, group_concat(label order by id desc separator \';\') as labels from label\n group by user_id) c on c.user_id=a.id\"\n# objFields:\n# _labels: array:;\n etlCondition: \"where a.c_time>=\'{0}\'\"\n commitBatch: 3000\n', '2019-01-01 17:05:43'); +INSERT INTO `canal_adapter_config` VALUES ('2', 'hhase', 'mytest_person2.yml', 'dataSourceKey: defaultDS\ndestination: example\nhbaseMapping:\n mode: STRING #NATIVE #PHOENIX\n database: mytest # 数据库名\n table: person2 # 数据库表名\n hbaseTable: MYTEST.PERSON2 # HBase表名\n family: CF # 默认统一Family名称\n uppercaseQualifier: true # 字段名转大写, 默认为true\n commitBatch: 3000 # 批量提交的大小\n #rowKey: id,type # 复合字段rowKey不能和columns中的rowKey重复\n columns:\n # 数据库字段:HBase对应字段\n id: ROWKEY LEN:15\n name: NAME\n email: EMAIL\n type:\n c_time: C_TIME\n birthday: BIRTHDAY\n# excludeColumns:\n# - lat # 忽略字段\n\n# -- NATIVE类型\n# $DEFAULT\n# $STRING\n# $INTEGER\n# $LONG\n# $SHORT\n# $BOOLEAN\n# $FLOAT\n# $DOUBLE\n# $BIGDECIMAL\n# $DATE\n# $BYTE\n# $BYTES\n\n# -- PHOENIX类型\n# $DEFAULT 对应PHOENIX里的VARCHAR\n# $UNSIGNED_INT 对应PHOENIX里的UNSIGNED_INT 4字节\n# $UNSIGNED_LONG 对应PHOENIX里的UNSIGNED_LONG 8字节\n# $UNSIGNED_TINYINT 对应PHOENIX里的UNSIGNED_TINYINT 1字节\n# $UNSIGNED_SMALLINT 对应PHOENIX里的UNSIGNED_SMALLINT 2字节\n# $UNSIGNED_FLOAT 对应PHOENIX里的UNSIGNED_FLOAT 4字节\n# $UNSIGNED_DOUBLE 对应PHOENIX里的UNSIGNED_DOUBLE 8字节\n# $INTEGER 对应PHOENIX里的INTEGER 4字节\n# $BIGINT 对应PHOENIX里的BIGINT 8字节\n# $TINYINT 对应PHOENIX里的TINYINT 1字节\n# $SMALLINT 对应PHOENIX里的SMALLINT 2字节\n# $FLOAT 对应PHOENIX里的FLOAT 4字节\n# $DOUBLE 对应PHOENIX里的DOUBLE 8字节\n# $BOOLEAN 对应PHOENIX里的BOOLEAN 1字节\n# $TIME 对应PHOENIX里的TIME 8字节\n# $DATE 对应PHOENIX里的DATE 8字节\n# $TIMESTAMP 对应PHOENIX里的TIMESTAMP 12字节\n# $UNSIGNED_TIME 对应PHOENIX里的UNSIGNED_TIME 8字节\n# $UNSIGNED_DATE 对应PHOENIX里的UNSIGNED_DATE 8字节\n# $UNSIGNED_TIMESTAMP 对应PHOENIX里的UNSIGNED_TIMESTAMP 12字节\n# $VARCHAR 对应PHOENIX里的VARCHAR 动态长度\n# $VARBINARY 对应PHOENIX里的VARBINARY 动态长度\n# $DECIMAL 对应PHOENIX里的DECIMAL 动态长度', '2019-01-01 17:08:08'); +INSERT INTO `canal_adapter_config` VALUES ('3', 'rdb', 'mytest_user.yml', 'dataSourceKey: defaultDS\ndestination: example\nouterAdapterKey: oracle1\nconcurrent: true\ndbMapping:\n database: mytest\n table: user\n targetTable: mytest.tb_user\n targetPk:\n id: id\n# mapAll: true\n targetColumns:\n id:\n name:\n role_id:\n c_time:\n test1:\n\n\n# Mirror schema synchronize config\n#dataSourceKey: defaultDS\n#destination: example\n#outerAdapterKey: mysql1\n#concurrent: true\n#dbMapping:\n# mirrorDb: true\n# database: mytest\n', '2019-01-01 17:08:50');