From 77646c018562470f1b5d43f7756156d9c248c5a8 Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 14 Jan 2019 10:58:33 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E6=8B=BC=E5=86=99=E9=94=99=E8=AF=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deployer/manager_ddl.sql | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/deployer/manager_ddl.sql b/deployer/manager_ddl.sql index c91b8ce0..c175c24e 100644 --- a/deployer/manager_ddl.sql +++ b/deployer/manager_ddl.sql @@ -33,5 +33,5 @@ INSERT INTO `canal_config` VALUES ('2', 'application.yml', 'server:\n port: 808 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 ('2', 'hbase', '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'); From f26adff7c79f16dd5e858444ea59b4882df20edc Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 14 Jan 2019 11:16:33 +0800 Subject: [PATCH 2/2] fixed #1416 --- .../monitor/AdapterRemoteConfigMonitor.java | 65 +++++++++++++------ .../monitor/ManagerRemoteConfigMonitor.java | 64 +++++++++++------- 2 files changed, 87 insertions(+), 42 deletions(-) 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 index 1b9b5e95..1609a249 100644 --- 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 @@ -69,6 +69,7 @@ public class AdapterRemoteConfigMonitor { if (configItem.getModifiedTime() != currentConfigTimestamp) { currentConfigTimestamp = configItem.getModifiedTime(); overrideLocalCanalConfig(configItem.getContent()); + logger.info("## Loaded remote adapter config: application.yml"); } } } catch (Exception e) { @@ -78,7 +79,7 @@ public class AdapterRemoteConfigMonitor { /** * 获取远程application.yml配置 - * + * * @return 配置对象 */ private ConfigItem getRemoteAdapterConfig() { @@ -100,7 +101,7 @@ public class AdapterRemoteConfigMonitor { /** * 覆盖本地application.yml文件 - * + * * @param content 文件内容 */ private void overrideLocalCanalConfig(String content) { @@ -140,7 +141,7 @@ public class AdapterRemoteConfigMonitor { */ public void loadRemoteAdapterConfigs() { try { - // 加载远程instance配置 + // 加载远程adapter配置 Map[] modifiedConfigs = getModifiedAdapterConfigs(); if (modifiedConfigs != null) { overrideLocalAdapterConfigs(modifiedConfigs); @@ -152,7 +153,7 @@ public class AdapterRemoteConfigMonitor { /** * 获取有变动的adapter配置 - * + * * @return Map[0]: 新增修改的配置, Map[1]: 删除的配置 */ @SuppressWarnings("unchecked") @@ -191,7 +192,7 @@ public class AdapterRemoteConfigMonitor { } } if (!changedIds.isEmpty()) { - Map changedInstanceConfig = new HashMap<>(); + Map 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)) { @@ -205,26 +206,26 @@ public class AdapterRemoteConfigMonitor { remoteAdapterConfigs.put(configItemNew.getCategory() + "/" + configItemNew.getName(), configItemNew); - changedInstanceConfig.put(configItemNew.getCategory() + "/" + configItemNew.getName(), + changedAdapterConfig.put(configItemNew.getCategory() + "/" + configItemNew.getName(), configItemNew); } - res[0] = changedInstanceConfig.isEmpty() ? null : changedInstanceConfig; + res[0] = changedAdapterConfig.isEmpty() ? null : changedAdapterConfig; } catch (Exception e) { logger.error(e.getMessage(), e); } } } - Map removedInstanceConfig = new HashMap<>(); + Map removedAdapterConfig = 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); + removedAdapterConfig.put(configItem.getCategory() + "/" + configItem.getName(), null); } } - res[1] = removedInstanceConfig.isEmpty() ? null : removedInstanceConfig; + res[1] = removedAdapterConfig.isEmpty() ? null : removedAdapterConfig; if (res[0] == null && res[1] == null) { return null; @@ -235,33 +236,57 @@ public class AdapterRemoteConfigMonitor { /** * 覆盖adapter配置到本地 - * - * @param modifiedInstanceConfigs 变动的配置集合 + * + * @param modifiedAdapterConfigs 变动的配置集合 */ - private void overrideLocalAdapterConfigs(Map[] modifiedInstanceConfigs) { - Map changedInstanceConfigs = modifiedInstanceConfigs[0]; - if (changedInstanceConfigs != null) { - for (ConfigItem configItem : changedInstanceConfigs.values()) { + private void overrideLocalAdapterConfigs(Map[] modifiedAdapterConfigs) { + Map 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 removedInstanceConfigs = modifiedInstanceConfigs[1]; - if (removedInstanceConfigs != null) { - for (String name : removedInstanceConfigs.keySet()) { + Map removedAdapterConfigs = modifiedAdapterConfigs[1]; + if (removedAdapterConfigs != null) { + for (String name : removedAdapterConfigs.keySet()) { File file = new File(getConfPath() + name); if (file.exists()) { - file.delete(); + deleteDir(file); + logger.info("## Deleted and reloaded remote adapter config: {}", name); } } } } + private static boolean deleteDir(File dirFile) { + if (!dirFile.exists()) { + return false; + } + + if (dirFile.isFile()) { + return dirFile.delete(); + } else { + File[] files = dirFile.listFiles(); + if (files == null || files.length == 0) { + return dirFile.delete(); + } + for (File file : files) { + deleteDir(file); + } + } + + return dirFile.delete(); + } + /** * 获取conf文件夹所在路径 * diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerRemoteConfigMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerRemoteConfigMonitor.java index 2f906db8..6cb0ebe2 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerRemoteConfigMonitor.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerRemoteConfigMonitor.java @@ -4,16 +4,8 @@ import java.io.ByteArrayInputStream; import java.io.File; import java.io.FileWriter; import java.nio.charset.StandardCharsets; -import java.sql.Connection; -import java.sql.DriverManager; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.Statement; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Properties; +import java.sql.*; +import java.util.*; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -28,7 +20,7 @@ import com.google.common.collect.MapMaker; /** * 远程配置装载监控 - * + * * @author rewerma 2018-12-30 下午05:12:16 * @version 1.0.1 */ @@ -47,7 +39,7 @@ public class ManagerRemoteConfigMonitor { private long scanIntervalInSecond = 5; private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2, - new NamedThreadFactory("remote-canal-config-scan")); + new NamedThreadFactory("remote-canal-config-scan")); public ManagerRemoteConfigMonitor(String jdbcUrl, String jdbcUsername, String jdbcPassword){ this.jdbcUrl = jdbcUrl; @@ -69,7 +61,7 @@ public class ManagerRemoteConfigMonitor { /** * 加载远程 canal.properties文件 覆盖本地 - * + * * @return 远程配置的properties */ public Properties loadRemoteConfig() { @@ -83,8 +75,9 @@ public class ManagerRemoteConfigMonitor { overrideLocalCanalConfig(configItem.getContent()); properties = new Properties(); properties.load(new ByteArrayInputStream(configItem.getContent().getBytes(StandardCharsets.UTF_8))); - scanIntervalInSecond = Integer.valueOf(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL, - "5")); + scanIntervalInSecond = Integer + .valueOf(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL, "5")); + logger.info("## Loaded remote canal config: canal.properties "); } } } catch (Exception e) { @@ -110,7 +103,7 @@ public class ManagerRemoteConfigMonitor { /** * 获取远程canal.properties配置内容 - * + * * @return 内容对象 */ private ConfigItem getRemoteCanalConfig() { @@ -132,7 +125,7 @@ public class ManagerRemoteConfigMonitor { /** * 覆盖本地 canal.properties - * + * * @param content 远程配置内容文本 */ private void overrideLocalCanalConfig(String content) { @@ -224,16 +217,22 @@ public class ManagerRemoteConfigMonitor { /** * 覆盖本地instance配置 - * + * * @param modifiedInstanceConfigs 有变更的配置项 */ private void overrideLocalInstanceConfigs(Map[] modifiedInstanceConfigs) { Map changedInstanceConfigs = modifiedInstanceConfigs[0]; if (changedInstanceConfigs != null) { for (ConfigItem configItem : changedInstanceConfigs.values()) { - try (FileWriter writer = new FileWriter(getConfPath() + configItem.getName() + "/instance.properties")) { + 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); } @@ -244,15 +243,36 @@ public class ManagerRemoteConfigMonitor { for (String name : removedInstanceConfigs.keySet()) { File file = new File(getConfPath() + name + "/"); if (file.exists()) { - file.delete(); + deleteDir(file); + logger.info("## Deleted and loaded remote instance config: {} ", name); } } } } + private static boolean deleteDir(File dirFile) { + if (!dirFile.exists()) { + return false; + } + + if (dirFile.isFile()) { + return dirFile.delete(); + } else { + File[] files = dirFile.listFiles(); + if (files == null || files.length == 0) { + return dirFile.delete(); + } + for (File file : files) { + deleteDir(file); + } + } + + return dirFile.delete(); + } + /** * 监听 canal 主配置和 instance 配置变化 - * + * * @param listener 监听回调方法 */ public void start(final Listener listener) { @@ -302,7 +322,7 @@ public class ManagerRemoteConfigMonitor { /** * 获取conf文件夹所在路径 - * + * * @return 路径地址 */ private String getConfPath() {