From 71b7c0b868c508bd130e0a4fe7c49826eedcbfc2 Mon Sep 17 00:00:00 2001 From: machey Date: Sun, 30 Dec 2018 16:44:35 +0800 Subject: [PATCH] =?UTF-8?q?canal=20server=E5=A2=9E=E5=8A=A0=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=BA=93=E8=BF=9C=E7=A8=8B=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deployer/manager_ddl.sql | 15 +- .../otter/canal/deployer/CanalLauncher.java | 38 +++-- .../monitor/ManagerDbConfigMonitor.java | 152 ++++++++++++++---- deployer/src/main/resources/canal.properties | 3 + 4 files changed, 163 insertions(+), 45 deletions(-) diff --git a/deployer/manager_ddl.sql b/deployer/manager_ddl.sql index 84e4dce9..a32e8dd7 100644 --- a/deployer/manager_ddl.sql +++ b/deployer/manager_ddl.sql @@ -7,4 +7,17 @@ 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 DEFAULT CHARSET=utf8mb4 ; + +CREATE TABLE `canal_instance_config` ( + `id` bigint(20) NOT NULL AUTO_INCREMENT, + `name` varchar(45) NOT NULL, + `content` text NOT NULL, + `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 ; + +INSERT INTO `canal_config` VALUES ('1', 'canal.properties', '#################################################\r\n######### common argument ############# \r\n#################################################\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 = false\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 = false\r\ncanal.mq.compressionType = none\r\ncanal.mq.acks = all\r\n', '2018-12-30 14:26:58'); + +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=false\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=root\r\ncanal.instance.dbPassword=121212\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=mytest\\\\..*\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 15:46:15'); \ No newline at end of file diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java index 6ef0f72c..50cc62e2 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java @@ -34,24 +34,30 @@ public class CanalLauncher { if (conf.startsWith(CLASSPATH_URL_PREFIX)) { conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); properties.load(CanalLauncher.class.getClassLoader().getResourceAsStream(conf)); - - String jdbcUrl = properties.getProperty("canal.manager.jdbc.url"); - if (!StringUtils.isEmpty(jdbcUrl)) { - // load remote config - String jdbcUsername = properties.getProperty("canal.manager.jdbc.username"); - String jdbcPassword = properties.getProperty("canal.manager.jdbc.password"); - managerDbConfigMonitor = new ManagerDbConfigMonitor(jdbcUrl, jdbcUsername, jdbcPassword); - Properties remoteConfig = managerDbConfigMonitor.loadRemoteConfig(); - if (remoteConfig != null) { - properties = remoteConfig; - } else { - managerDbConfigMonitor = null; - } - } } else { properties.load(new FileInputStream(conf)); } + String jdbcUrl = properties.getProperty("canal.manager.jdbc.url"); + if (!StringUtils.isEmpty(jdbcUrl)) { + logger.info("## load remote canal configurations"); + // load remote config + String jdbcUsername = properties.getProperty("canal.manager.jdbc.username"); + String jdbcPassword = properties.getProperty("canal.manager.jdbc.password"); + managerDbConfigMonitor = new ManagerDbConfigMonitor(jdbcUrl, jdbcUsername, jdbcPassword); + // 加载远程canal.properties + Properties remoteConfig = managerDbConfigMonitor.loadRemoteConfig(); + // 加载remote instance配置 + managerDbConfigMonitor.loadRemoteInstanceConfigs(); + if (remoteConfig != null) { + properties = remoteConfig; + } else { + managerDbConfigMonitor = null; + } + } else { + logger.info("## load canal configurations"); + } + final CanalStater canalStater = new CanalStater(); canalStater.start(properties); @@ -73,6 +79,10 @@ public class CanalLauncher { while (running) ; + + if (managerDbConfigMonitor != null) { + managerDbConfigMonitor.destroy(); + } } catch (Throwable e) { logger.error("## Something goes wrong when starting up the canal Server:", e); } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerDbConfigMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerDbConfigMonitor.java index 973dc28b..8c8c27dd 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerDbConfigMonitor.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerDbConfigMonitor.java @@ -1,6 +1,7 @@ package com.alibaba.otter.canal.deployer.monitor; import java.io.ByteArrayInputStream; +import java.io.File; import java.io.FileWriter; import java.nio.charset.StandardCharsets; import java.sql.*; @@ -9,14 +10,13 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import com.google.common.collect.MapMaker; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.common.utils.NamedThreadFactory; import com.alibaba.otter.canal.deployer.CanalConstants; -import com.google.common.base.Function; -import com.google.common.collect.MigrateMap; +import com.google.common.base.Joiner; +import com.google.common.collect.MapMaker; public class ManagerDbConfigMonitor { @@ -32,7 +32,7 @@ public class ManagerDbConfigMonitor { private long currentConfigTimestamp = 0; private long scanIntervalInSecond = 5; - private ScheduledExecutorService executor = Executors.newScheduledThreadPool(1, + private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2, new NamedThreadFactory("remote-canal-config-scan")); public ManagerDbConfigMonitor(String jdbcUrl, String jdbcUsername, String jdbcPassword){ @@ -56,6 +56,7 @@ public class ManagerDbConfigMonitor { public Properties loadRemoteConfig() { Properties properties = null; try { + // 加载远程canal配置 ConfigItem configItem = getRemoteCanalConfig(); if (configItem != null) { if (configItem.getModifiedTime() != currentConfigTimestamp) { @@ -73,6 +74,18 @@ public class ManagerDbConfigMonitor { return properties; } + public void loadRemoteInstanceConfigs() { + try { + // 加载远程instance配置 + Map[] modifiedConfigs = getModifiedInstanceConfigs(); + if (modifiedConfigs != null) { + overrideLocalInstanceConfigs(modifiedConfigs); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + private ConfigItem getRemoteCanalConfig() { String sql = "select name, content, modified_time from canal_config where id=1"; try (Statement stmt = getConn().createStatement(); ResultSet rs = stmt.executeQuery(sql)) { @@ -91,7 +104,7 @@ public class ManagerDbConfigMonitor { } private void overrideLocalCanalConfig(String content) { - try (FileWriter writer = new FileWriter("../conf/canal.properties")) { + try (FileWriter writer = new FileWriter(getConfPath() + "canal.properties")) { writer.write(content); writer.flush(); } catch (Exception e) { @@ -100,6 +113,7 @@ public class ManagerDbConfigMonitor { } public void start(final Listener listener) { + // 监听canal.properties变化 executor.scheduleWithFixedDelay(new Runnable() { public void run() { @@ -115,45 +129,99 @@ public class ManagerDbConfigMonitor { } }, 10, scanIntervalInSecond, TimeUnit.SECONDS); + + // 监听instance变化 + executor.scheduleWithFixedDelay(new Runnable() { + + public void run() { + try { + loadRemoteInstanceConfigs(); + } catch (Throwable e) { + logger.error("scan failed", e); + } + } + + }, 10, 3, TimeUnit.SECONDS); } - public interface Listener { - - void onChange(Properties properties); - } - - private Map getChangedInstanceConfigs() { - String sql = "select name, content, modified_time from instance_config"; + @SuppressWarnings("unchecked") + private Map[] getModifiedInstanceConfigs() { + Map[] res = new Map[2]; + Map remoteConfigStatus = new HashMap<>(); + String sql = "select id, name, modified_time from canal_instance_config"; try (Statement stmt = getConn().createStatement(); ResultSet rs = stmt.executeQuery(sql)) { - Map changedInstanceConfig = new HashMap<>(); while (rs.next()) { - ConfigItem configItemNew = new ConfigItem(); - configItemNew.setName(rs.getString("name")); - configItemNew.setContent(rs.getString("content")); - configItemNew.setModifiedTime(rs.getTimestamp("modified_time").getTime()); + ConfigItem configItem = new ConfigItem(); + configItem.setId(rs.getLong("id")); + configItem.setName(rs.getString("name")); + configItem.setModifiedTime(rs.getTimestamp("modified_time").getTime()); + remoteConfigStatus.put(configItem.getName(), configItem); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + return null; + } - ConfigItem configItem = remoteInstanceConfigs.get(configItemNew.getName()); - if (configItem == null) { - remoteInstanceConfigs.put(configItemNew.getName(), configItemNew); - changedInstanceConfig.put(configItemNew.getName(), configItemNew); + if (!remoteConfigStatus.isEmpty()) { + List changedIds = new ArrayList<>(); + + for (ConfigItem remoteConfigStat : remoteConfigStatus.values()) { + ConfigItem currentConfig = remoteInstanceConfigs.get(remoteConfigStat.getName()); + if (currentConfig == null) { + // 新增 + changedIds.add(remoteConfigStat.getId()); } else { - if (configItem.getModifiedTime() != configItemNew.getModifiedTime()) { - remoteInstanceConfigs.put(configItemNew.getName(), configItemNew); - changedInstanceConfig.put(configItemNew.getName(), configItemNew); + // 修改 + if (currentConfig.getModifiedTime() != remoteConfigStat.getModifiedTime()) { + changedIds.add(remoteConfigStat.getId()); } } } - return changedInstanceConfig.isEmpty() ? null : changedInstanceConfig; - } catch (Exception e) { - logger.error(e.getMessage(), e); + if (!changedIds.isEmpty()) { + Map changedInstanceConfig = new HashMap<>(); + String contentsSql = "select id, name, content, modified_time from canal_instance_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.setName(rs.getString("name")); + configItemNew.setContent(rs.getString("content")); + configItemNew.setModifiedTime(rs.getTimestamp("modified_time").getTime()); + + remoteInstanceConfigs.put(configItemNew.getName(), configItemNew); + changedInstanceConfig.put(configItemNew.getName(), configItemNew); + } + + res[0] = changedInstanceConfig.isEmpty() ? null : changedInstanceConfig; + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + } + + Map removedInstanceConfig = new HashMap<>(); + for (String name : remoteInstanceConfigs.keySet()) { + if (!remoteConfigStatus.containsKey(name)) { + // 删除 + removedInstanceConfig.put(name, null); + } + } + res[1] = removedInstanceConfig.isEmpty() ? null : removedInstanceConfig; + + if (res[0] == null && res[1] == null) { + return null; + } else { + return res; } - return null; } - private void overrideLocalInstanceConfigs(Map changedInstanceConfigs) { + private void overrideLocalInstanceConfigs(Map[] modifiedInstanceConfigs) { + Map changedInstanceConfigs = modifiedInstanceConfigs[0]; if (changedInstanceConfigs != null) { for (ConfigItem configItem : changedInstanceConfigs.values()) { - try (FileWriter writer = new FileWriter("../conf/" + configItem.getName() + "/instance.properties")) { + try (FileWriter writer = new FileWriter( + getConfPath() + configItem.getName() + "/instance.properties")) { writer.write(configItem.getContent()); writer.flush(); } catch (Exception e) { @@ -161,9 +229,19 @@ public class ManagerDbConfigMonitor { } } } + Map removedInstanceConfigs = modifiedInstanceConfigs[1]; + if (removedInstanceConfigs != null) { + for (String name : removedInstanceConfigs.keySet()) { + File file = new File(getConfPath() + name + "/"); + if (file.exists()) { + file.delete(); + } + } + } } public void destroy() { + executor.shutdownNow(); if (conn != null) { try { conn.close(); @@ -173,6 +251,16 @@ public class ManagerDbConfigMonitor { } } + private String getConfPath() { + String classpath = this.getClass().getResource("/").getPath(); + String confPath = classpath + ".." + File.separator + "conf" + File.separator; + if (new File(confPath).exists()) { + return confPath; + } else { + return classpath; + } + } + public static class ConfigItem { private Long id; @@ -213,4 +301,8 @@ public class ManagerDbConfigMonitor { } } + public interface Listener { + + void onChange(Properties properties); + } } diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 5d7b9d06..eba01959 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -1,6 +1,9 @@ ################################################# ######### common argument ############# ################################################# +#canal.manager.jdbc.url=jdbc:mysql://127.0.0.1:3306/canal_manager?useUnicode=true&characterEncoding=UTF-8 +#canal.manager.jdbc.username=root +#canal.manager.jdbc.password=121212 canal.id = 1 canal.ip = canal.port = 11111