diff --git a/deployer/manager_ddl.sql b/deployer/manager_ddl.sql new file mode 100644 index 00000000..a5e6dbd4 --- /dev/null +++ b/deployer/manager_ddl.sql @@ -0,0 +1,23 @@ +CREATE SCHEMA `canal_manager` DEFAULT CHARACTER SET utf8mb4 ; + +CREATE TABLE `canal_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 ; + +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\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_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 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 fd52f3fa..aa9ac603 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 @@ -7,11 +7,7 @@ import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.common.MQProperties; -import com.alibaba.otter.canal.kafka.CanalKafkaProducer; -import com.alibaba.otter.canal.rocketmq.CanalRocketMQProducer; -import com.alibaba.otter.canal.server.CanalMQStarter; -import com.alibaba.otter.canal.spi.CanalMQProducer; +import com.alibaba.otter.canal.deployer.monitor.ManagerRemoteConfigMonitor; /** * canal独立版本启动的入口类 @@ -21,17 +17,20 @@ import com.alibaba.otter.canal.spi.CanalMQProducer; */ public class CanalLauncher { - private static final String CLASSPATH_URL_PREFIX = "classpath:"; - private static final Logger logger = LoggerFactory.getLogger(CanalLauncher.class); + private static final String CLASSPATH_URL_PREFIX = "classpath:"; + private static final Logger logger = LoggerFactory.getLogger(CanalLauncher.class); + public static volatile boolean running = false; - public static void main(String[] args) throws Throwable { + public static void main(String[] args) { try { + running = true; logger.info("## set default uncaught exception handler"); setGlobalUncaughtExceptionHandler(); logger.info("## load canal configurations"); String conf = System.getProperty("canal.conf", "classpath:canal.properties"); Properties properties = new Properties(); + ManagerRemoteConfigMonitor managerDbConfigMonitor = null; if (conf.startsWith(CLASSPATH_URL_PREFIX)) { conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); properties.load(CanalLauncher.class.getClassLoader().getResourceAsStream(conf)); @@ -39,110 +38,56 @@ public class CanalLauncher { properties.load(new FileInputStream(conf)); } - CanalMQProducer canalMQProducer = null; - String serverMode = CanalController.getProperty(properties, CanalConstants.CANAL_SERVER_MODE); - if (serverMode.equalsIgnoreCase("kafka")) { - canalMQProducer = new CanalKafkaProducer(); - } else if (serverMode.equalsIgnoreCase("rocketmq")) { - canalMQProducer = new CanalRocketMQProducer(); - } - - if (canalMQProducer != null) { - // disable netty - System.setProperty(CanalConstants.CANAL_WITHOUT_NETTY, "true"); - System.setProperty(CanalConstants.CANAL_DESTINATIONS, - properties.getProperty(CanalConstants.CANAL_DESTINATIONS)); - } - - logger.info("## start the canal server."); - final CanalController controller = new CanalController(properties); - controller.start(); - logger.info("## the canal server is running now ......"); - Runtime.getRuntime().addShutdownHook(new Thread() { - - public void run() { - try { - logger.info("## stop the canal server"); - controller.stop(); - } catch (Throwable e) { - logger.warn("##something goes wrong when stopping canal Server:", e); - } finally { - logger.info("## canal server is down."); - } + 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 ManagerRemoteConfigMonitor(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); - if (canalMQProducer != null) { - CanalMQStarter canalMQStarter = new CanalMQStarter(canalMQProducer); - MQProperties mqProperties = buildMQPosition(properties); - canalMQStarter.start(mqProperties); - controller.setCanalMQStarter(canalMQStarter); + if (managerDbConfigMonitor != null) { + managerDbConfigMonitor.start(new ManagerRemoteConfigMonitor.Listener() { + + @Override + public void onChange(Properties properties) { + try { + // 远程配置canal.properties修改重新加载整个应用 + canalStater.destroy(); + canalStater.start(properties); + } catch (Throwable throwable) { + logger.error(throwable.getMessage(), throwable); + } + } + }); + } + + while (running) + ; + + if (managerDbConfigMonitor != null) { + managerDbConfigMonitor.destroy(); } } catch (Throwable e) { logger.error("## Something goes wrong when starting up the canal Server:", e); - System.exit(0); } } - private static MQProperties buildMQPosition(Properties properties) { - MQProperties mqProperties = new MQProperties(); - String servers = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_SERVERS); - if (!StringUtils.isEmpty(servers)) { - mqProperties.setServers(servers); - } - String retires = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_RETRIES); - if (!StringUtils.isEmpty(retires)) { - mqProperties.setRetries(Integer.valueOf(retires)); - } - String batchSize = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_BATCHSIZE); - if (!StringUtils.isEmpty(batchSize)) { - mqProperties.setBatchSize(Integer.valueOf(batchSize)); - } - String lingerMs = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_LINGERMS); - if (!StringUtils.isEmpty(lingerMs)) { - mqProperties.setLingerMs(Integer.valueOf(lingerMs)); - } - String maxRequestSize = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_MAXREQUESTSIZE); - if (!StringUtils.isEmpty(maxRequestSize)) { - mqProperties.setMaxRequestSize(Integer.valueOf(maxRequestSize)); - } - String bufferMemory = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_BUFFERMEMORY); - if (!StringUtils.isEmpty(bufferMemory)) { - mqProperties.setBufferMemory(Long.valueOf(bufferMemory)); - } - String canalBatchSize = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_CANALBATCHSIZE); - if (!StringUtils.isEmpty(canalBatchSize)) { - mqProperties.setCanalBatchSize(Integer.valueOf(canalBatchSize)); - } - String canalGetTimeout = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_CANALGETTIMEOUT); - if (!StringUtils.isEmpty(canalGetTimeout)) { - mqProperties.setCanalGetTimeout(Long.valueOf(canalGetTimeout)); - } - String flatMessage = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_FLATMESSAGE); - if (!StringUtils.isEmpty(flatMessage)) { - mqProperties.setFlatMessage(Boolean.valueOf(flatMessage)); - } - String compressionType = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_COMPRESSION_TYPE); - if (!StringUtils.isEmpty(compressionType)) { - mqProperties.setCompressionType(compressionType); - } - String acks = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_ACKS); - if (!StringUtils.isEmpty(acks)) { - mqProperties.setAcks(acks); - } - - String aliyunAccessKey = CanalController.getProperty(properties, CanalConstants.CANAL_ALIYUN_ACCESSKEY); - if (!StringUtils.isEmpty(aliyunAccessKey)) { - mqProperties.setAliyunAccessKey(aliyunAccessKey); - } - String aliyunSecretKey = CanalController.getProperty(properties, CanalConstants.CANAL_ALIYUN_SECRETKEY); - if (!StringUtils.isEmpty(aliyunSecretKey)) { - mqProperties.setAliyunSecretKey(aliyunSecretKey); - } - return mqProperties; - } - private static void setGlobalUncaughtExceptionHandler() { Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { @@ -152,4 +97,5 @@ public class CanalLauncher { } }); } + } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java new file mode 100644 index 00000000..f8503fa9 --- /dev/null +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java @@ -0,0 +1,165 @@ +package com.alibaba.otter.canal.deployer; + +import java.util.Properties; + +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.alibaba.otter.canal.common.MQProperties; +import com.alibaba.otter.canal.kafka.CanalKafkaProducer; +import com.alibaba.otter.canal.rocketmq.CanalRocketMQProducer; +import com.alibaba.otter.canal.server.CanalMQStarter; +import com.alibaba.otter.canal.spi.CanalMQProducer; + +/** + * Canal server 启动类 + * + * @author rewerma 2018-12-30 下午05:12:16 + * @version 1.0.1 + */ +public class CanalStater { + + private static final Logger logger = LoggerFactory.getLogger(CanalStater.class); + + private CanalController controller = null; + private CanalMQProducer canalMQProducer = null; + private Thread shutdownThread = null; + private CanalMQStarter canalMQStarter = null; + + /** + * 启动方法 + * + * @param properties canal.properties 配置 + * @throws Throwable + */ + synchronized void start(Properties properties) throws Throwable { + String serverMode = CanalController.getProperty(properties, CanalConstants.CANAL_SERVER_MODE); + if (serverMode.equalsIgnoreCase("kafka")) { + canalMQProducer = new CanalKafkaProducer(); + } else if (serverMode.equalsIgnoreCase("rocketmq")) { + canalMQProducer = new CanalRocketMQProducer(); + } + + if (canalMQProducer != null) { + // disable netty + System.setProperty(CanalConstants.CANAL_WITHOUT_NETTY, "true"); + System.setProperty(CanalConstants.CANAL_DESTINATIONS, + properties.getProperty(CanalConstants.CANAL_DESTINATIONS)); + } + + logger.info("## start the canal server."); + controller = new CanalController(properties); + controller.start(); + logger.info("## the canal server is running now ......"); + shutdownThread = new Thread() { + + public void run() { + try { + logger.info("## stop the canal server"); + controller.stop(); + CanalLauncher.running = false; + } catch (Throwable e) { + logger.warn("##something goes wrong when stopping canal Server:", e); + } finally { + logger.info("## canal server is down."); + } + } + + }; + Runtime.getRuntime().addShutdownHook(shutdownThread); + + if (canalMQProducer != null) { + canalMQStarter = new CanalMQStarter(canalMQProducer); + MQProperties mqProperties = buildMQProperties(properties); + canalMQStarter.start(mqProperties); + controller.setCanalMQStarter(canalMQStarter); + } + } + + /** + * 销毁方法，远程配置变更时调用 + * + * @throws Throwable + */ + synchronized void destroy() throws Throwable { + if (controller != null) { + controller.stop(); + controller = null; + } + if (shutdownThread != null) { + Runtime.getRuntime().removeShutdownHook(shutdownThread); + shutdownThread = null; + } + if (canalMQProducer != null && canalMQStarter != null) { + canalMQStarter.destroy(); + canalMQStarter = null; + canalMQProducer = null; + } + } + + /** + * 构造MQ对应的配置 + * + * @param properties canal.properties 配置 + * @return + */ + private static MQProperties buildMQProperties(Properties properties) { + MQProperties mqProperties = new MQProperties(); + String servers = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_SERVERS); + if (!StringUtils.isEmpty(servers)) { + mqProperties.setServers(servers); + } + String retires = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_RETRIES); + if (!StringUtils.isEmpty(retires)) { + mqProperties.setRetries(Integer.valueOf(retires)); + } + String batchSize = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_BATCHSIZE); + if (!StringUtils.isEmpty(batchSize)) { + mqProperties.setBatchSize(Integer.valueOf(batchSize)); + } + String lingerMs = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_LINGERMS); + if (!StringUtils.isEmpty(lingerMs)) { + mqProperties.setLingerMs(Integer.valueOf(lingerMs)); + } + String maxRequestSize = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_MAXREQUESTSIZE); + if (!StringUtils.isEmpty(maxRequestSize)) { + mqProperties.setMaxRequestSize(Integer.valueOf(maxRequestSize)); + } + String bufferMemory = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_BUFFERMEMORY); + if (!StringUtils.isEmpty(bufferMemory)) { + mqProperties.setBufferMemory(Long.valueOf(bufferMemory)); + } + String canalBatchSize = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_CANALBATCHSIZE); + if (!StringUtils.isEmpty(canalBatchSize)) { + mqProperties.setCanalBatchSize(Integer.valueOf(canalBatchSize)); + } + String canalGetTimeout = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_CANALGETTIMEOUT); + if (!StringUtils.isEmpty(canalGetTimeout)) { + mqProperties.setCanalGetTimeout(Long.valueOf(canalGetTimeout)); + } + String flatMessage = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_FLATMESSAGE); + if (!StringUtils.isEmpty(flatMessage)) { + mqProperties.setFlatMessage(Boolean.valueOf(flatMessage)); + } + String compressionType = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_COMPRESSION_TYPE); + if (!StringUtils.isEmpty(compressionType)) { + mqProperties.setCompressionType(compressionType); + } + String acks = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_ACKS); + if (!StringUtils.isEmpty(acks)) { + mqProperties.setAcks(acks); + } + + String aliyunAccessKey = CanalController.getProperty(properties, CanalConstants.CANAL_ALIYUN_ACCESSKEY); + if (!StringUtils.isEmpty(aliyunAccessKey)) { + mqProperties.setAliyunAccessKey(aliyunAccessKey); + } + String aliyunSecretKey = CanalController.getProperty(properties, CanalConstants.CANAL_ALIYUN_SECRETKEY); + if (!StringUtils.isEmpty(aliyunSecretKey)) { + mqProperties.setAliyunSecretKey(aliyunSecretKey); + } + return mqProperties; + } + +} 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 new file mode 100644 index 00000000..de2c3ede --- /dev/null +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerRemoteConfigMonitor.java @@ -0,0 +1,358 @@ +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.*; +import java.util.*; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +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.Joiner; +import com.google.common.collect.MapMaker; + +/** + * 远程配置装载监控 + * + * @author rewerma 2018-12-30 下午05:12:16 + * @version 1.0.1 + */ +public class ManagerRemoteConfigMonitor { + + private static final Logger logger = LoggerFactory.getLogger(ManagerRemoteConfigMonitor.class); + + private Map remoteInstanceConfigs = new MapMaker().makeMap(); + + private Connection conn; + private String jdbcUrl; + private String jdbcUsername; + private String jdbcPassword; + + private long currentConfigTimestamp = 0; + + private long scanIntervalInSecond = 5; + private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2, + new NamedThreadFactory("remote-canal-config-scan")); + + public ManagerRemoteConfigMonitor(String jdbcUrl, String jdbcUsername, String jdbcPassword){ + this.jdbcUrl = jdbcUrl; + this.jdbcUsername = jdbcUsername; + this.jdbcPassword = jdbcPassword; + } + + public void setScanIntervalInSecond(long scanIntervalInSecond) { + this.scanIntervalInSecond = scanIntervalInSecond; + } + + private Connection getConn() throws Exception { + if (conn == null || conn.isClosed()) { + Class.forName("com.mysql.jdbc.Driver"); + conn = DriverManager.getConnection(jdbcUrl, jdbcUsername, jdbcPassword); + } + return conn; + } + + /** + * 加载远程 canal.properties文件 覆盖本地 + * + * @return 远程配置的properties + */ + public Properties loadRemoteConfig() { + Properties properties = null; + try { + // 加载远程canal配置 + ConfigItem configItem = getRemoteCanalConfig(); + if (configItem != null) { + if (configItem.getModifiedTime() != currentConfigTimestamp) { + currentConfigTimestamp = configItem.getModifiedTime(); + 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")); + } + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + return properties; + } + + /** + * 加载远程的instance配置 覆盖本地 + */ + public void loadRemoteInstanceConfigs() { + try { + // 加载远程instance配置 + Map[] modifiedConfigs = getModifiedInstanceConfigs(); + if (modifiedConfigs != null) { + overrideLocalInstanceConfigs(modifiedConfigs); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + /** + * 获取远程canal.properties配置内容 + * + * @return 内容对象 + */ + 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)) { + if (rs.next()) { + ConfigItem configItem = new ConfigItem(); + configItem.setId(1L); + 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; + } + + /** + * 覆盖本地 canal.properties + * + * @param content 远程配置内容文本 + */ + private void overrideLocalCanalConfig(String content) { + try (FileWriter writer = new FileWriter(getConfPath() + "canal.properties")) { + writer.write(content); + writer.flush(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + /** + * 获取远程instance新增、修改、删除配置 + * + * @return Map[0]：新增或修改的instance配置; Map[1]：删除的instance配置 + */ + @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)) { + while (rs.next()) { + 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; + } + + 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 (currentConfig.getModifiedTime() != remoteConfigStat.getModifiedTime()) { + changedIds.add(remoteConfigStat.getId()); + } + } + } + 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)) { + // 删除 + remoteInstanceConfigs.remove(name); + removedInstanceConfig.put(name, null); + } + } + res[1] = removedInstanceConfig.isEmpty() ? null : removedInstanceConfig; + + if (res[0] == null && res[1] == null) { + return null; + } else { + return res; + } + } + + /** + * 覆盖本地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")) { + 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(); + } + } + } + } + + /** + * 监听 canal 主配置和 instance 配置变化 + * + * @param listener 监听回调方法 + */ + public void start(final Listener listener) { + // 监听canal.properties变化 + executor.scheduleWithFixedDelay(new Runnable() { + + public void run() { + try { + Properties properties = loadRemoteConfig(); + if (properties != null) { + listener.onChange(properties); + } + } catch (Throwable e) { + logger.error("scan failed", e); + } + } + + }, 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 void destroy() { + executor.shutdownNow(); + if (conn != null) { + try { + conn.close(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } + } + + /** + * 获取conf文件夹所在路径 + * + * @return 路径地址 + */ + 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; + private String name; + private String content; + private long modifiedTime; + + public Long getId() { + return id; + } + + public void setId(Long id) { + this.id = id; + } + + 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; + } + } + + public interface Listener { + + void onChange(Properties properties); + } +} diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 42fd373d..a7221726 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 diff --git a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java index 0c90f44f..b24bb7cf 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java @@ -21,9 +21,9 @@ import com.alibaba.otter.canal.spi.CanalMQProducer; public class CanalMQStarter { - private static final Logger logger = LoggerFactory.getLogger(CanalMQStarter.class); + private static final Logger logger = LoggerFactory.getLogger(CanalMQStarter.class); - private volatile boolean running = false; + private volatile boolean running = false; private ExecutorService executorService; @@ -33,7 +33,9 @@ public class CanalMQStarter { private CanalServerWithEmbedded canalServer; - private Map canalMQWorks = new ConcurrentHashMap<>(); + private Map canalMQWorks = new ConcurrentHashMap<>(); + + private static Thread shutdownThread = null; public CanalMQStarter(CanalMQProducer canalMQProducer){ this.canalMQProducer = canalMQProducer; @@ -72,7 +74,8 @@ public class CanalMQStarter { running = true; logger.info("## the MQ workers is running now ......"); - Runtime.getRuntime().addShutdownHook(new Thread() { + + shutdownThread = new Thread() { public void run() { try { @@ -87,11 +90,25 @@ public class CanalMQStarter { } } - }); + }; + Runtime.getRuntime().addShutdownHook(shutdownThread); } catch (Throwable e) { logger.error("## Something goes wrong when starting up the canal MQ workers:", e); - System.exit(0); + } + } + + public synchronized void destroy() { + running = false; + if (executorService != null) { + executorService.shutdown(); + } + if (canalMQProducer != null) { + canalMQProducer.stop(); + } + if (shutdownThread != null) { + Runtime.getRuntime().removeShutdownHook(shutdownThread); + shutdownThread = null; } }