chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed issue #2099 , support remote config poll

Browse Source
This commit is contained in:
agapple
2019-08-27 16:11:20 +08:00
parent 2e1b7a5093
commit bf89974f7c
48 changed files with 1255 additions and 1025 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
deployer/manager_ddl.sql
View File

@@ -1,37 +0,0 @@
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 AUTO_INCREMENT=3 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 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');
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', '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');

1
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java
View File

@@ -19,6 +19,7 @@ public class CanalConstants {
public static final String CANAL_USER = ROOT + "." + "user";
public static final String CANAL_PASSWD = ROOT + "." + "passwd";
public static final String CANAL_METRICS_PULL_PORT = ROOT + "." + "metrics.pull.port";
public static final String CANAL_ADMIN_MANAGER = ROOT + "." + "admin.manager";
public static final String CANAL_ADMIN_PORT = ROOT + "." + "admin.port";
public static final String CANAL_ADMIN_USER = ROOT + "." + "admin.user";
public static final String CANAL_ADMIN_PASSWD = ROOT + "." + "admin.passwd";

84
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
View File

@@ -12,11 +12,7 @@ import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import com.alibaba.otter.canal.common.CanalException;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
@@ -31,10 +27,9 @@ import com.alibaba.otter.canal.deployer.monitor.ManagerInstanceConfigMonitor;
import com.alibaba.otter.canal.deployer.monitor.SpringInstanceConfigMonitor;
import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator;
import com.alibaba.otter.canal.instance.manager.CanalConfigClient;
import com.alibaba.otter.canal.instance.manager.ManagerCanalInstanceGenerator;
import com.alibaba.otter.canal.instance.manager.PlainCanalInstanceGenerator;
import com.alibaba.otter.canal.instance.manager.plain.PlainCanalConfigClient;
import com.alibaba.otter.canal.instance.spring.SpringCanalInstanceGenerator;
import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.server.CanalMQStarter;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.alibaba.otter.canal.server.exception.CanalServerException;
@@ -56,10 +51,11 @@ public class CanalController {
private String ip;
private String registerIp;
private int port;
private int adminPort;
// 默认使用spring的方式载入
private Map<String, InstanceConfig> instanceConfigs;
private InstanceConfig globalInstanceConfig;
private Map<String, CanalConfigClient> managerClients;
private Map<String, PlainCanalConfigClient> managerClients;
// 监听instance config的变化
private boolean autoScan = true;
private InstanceAction defaultAction;
@@ -71,15 +67,17 @@ public class CanalController {
private ZkClientx zkclientx;
private CanalMQStarter canalMQStarter;
private String adminUser;
private String adminPasswd;
public CanalController(){
this(System.getProperties());
}
public CanalController(final Properties properties){
managerClients = MigrateMap.makeComputingMap(new Function<String, CanalConfigClient>() {
managerClients = MigrateMap.makeComputingMap(new Function<String, PlainCanalConfigClient>() {
public CanalConfigClient apply(String managerAddress) {
public PlainCanalConfigClient apply(String managerAddress) {
return getManagerClient(managerAddress);
}
});
@@ -111,6 +109,7 @@ public class CanalController {
ip = getProperty(properties, CanalConstants.CANAL_IP);
registerIp = getProperty(properties, CanalConstants.CANAL_REGISTER_IP);
port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT));
adminPort = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_ADMIN_PORT));
embededCanalServer = CanalServerWithEmbedded.instance();
embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator
try {
@@ -121,6 +120,8 @@ public class CanalController {
embededCanalServer.setMetricsPort(11112);
}
this.adminUser = getProperty(properties, CanalConstants.CANAL_ADMIN_USER);
this.adminPasswd = getProperty(properties, CanalConstants.CANAL_ADMIN_PASSWD);
embededCanalServer.setUser(getProperty(properties, CanalConstants.CANAL_USER));
embededCanalServer.setPasswd(getProperty(properties, CanalConstants.CANAL_PASSWD));
@@ -254,6 +255,8 @@ public class CanalController {
runningMonitor.start();
}
}
logger.info("auto notify start {} successful.", destination);
}
public void stop(String destination) {
@@ -266,12 +269,16 @@ public class CanalController {
runningMonitor.stop();
}
}
logger.info("auto notify stop {} successful.", destination);
}
public void reload(String destination) {
// 目前任何配置变化，直接重启，简单处理
stop(destination);
start(destination);
logger.info("auto notify reload {} successful.", destination);
}
};
@@ -298,7 +305,14 @@ public class CanalController {
}
return monitor;
} else if (mode.isManager()) {
return new ManagerInstanceConfigMonitor();
ManagerInstanceConfigMonitor monitor = new ManagerInstanceConfigMonitor();
monitor.setScanIntervalInSecond(scanInterval);
monitor.setDefaultAction(defaultAction);
String managerAddress = getProperty(properties, CanalConstants.CANAL_ADMIN_MANAGER);
monitor.setConfigClient(getManagerClient(managerAddress));
monitor.setIp(registerIp);
monitor.setPort(adminPort);
return monitor;
} else {
throw new UnsupportedOperationException("unknow mode :" + mode + " for monitor");
}
@@ -308,9 +322,13 @@ public class CanalController {
}
private InstanceConfig initGlobalConfig(Properties properties) {
String adminManagerAddress = getProperty(properties, CanalConstants.CANAL_ADMIN_MANAGER);
InstanceConfig globalConfig = new InstanceConfig();
String modeStr = getProperty(properties, CanalConstants.getInstanceModeKey(CanalConstants.GLOBAL_NAME));
if (StringUtils.isNotEmpty(modeStr)) {
if (StringUtils.isNotEmpty(adminManagerAddress)) {
// 如果指定了manager地址,则强制适用manager
globalConfig.setMode(InstanceMode.MANAGER);
} else if (StringUtils.isNotEmpty(modeStr)) {
globalConfig.setMode(InstanceMode.valueOf(StringUtils.upperCase(modeStr)));
}
@@ -322,6 +340,10 @@ public class CanalController {
String managerAddress = getProperty(properties,
CanalConstants.getInstanceManagerAddressKey(CanalConstants.GLOBAL_NAME));
if (StringUtils.isNotEmpty(managerAddress)) {
if (StringUtils.equals(managerAddress, "${canal.admin.manager}")) {
managerAddress = adminManagerAddress;
}
globalConfig.setManagerAddress(managerAddress);
}
@@ -335,28 +357,18 @@ public class CanalController {
public CanalInstance generate(String destination) {
InstanceConfig config = instanceConfigs.get(destination);
if (config == null) {
throw new CanalServerException("can't find destination:{}");
throw new CanalServerException("can't find destination:" + destination);
}
if (config.getMode().isManager()) {
ManagerCanalInstanceGenerator instanceGenerator = new ManagerCanalInstanceGenerator();
PlainCanalInstanceGenerator instanceGenerator = new PlainCanalInstanceGenerator();
instanceGenerator.setCanalConfigClient(managerClients.get(config.getManagerAddress()));
instanceGenerator.setSpringXml(config.getSpringXml());
return instanceGenerator.generate(destination);
} else if (config.getMode().isSpring()) {
SpringCanalInstanceGenerator instanceGenerator = new SpringCanalInstanceGenerator();
synchronized (CanalEventParser.class) {
try {
// 设置当前正在加载的通道，加载spring查找文件时会用到该变量
System.setProperty(CanalConstants.CANAL_DESTINATION_PROPERTY, destination);
instanceGenerator.setBeanFactory(getBeanFactory(config.getSpringXml()));
return instanceGenerator.generate(destination);
} catch (Throwable e) {
logger.error("generator instance failed.", e);
throw new CanalException(e);
} finally {
System.setProperty(CanalConstants.CANAL_DESTINATION_PROPERTY, "");
}
}
instanceGenerator.setSpringXml(config.getSpringXml());
return instanceGenerator.generate(destination);
} else {
throw new UnsupportedOperationException("unknow mode :" + config.getMode());
}
@@ -368,13 +380,8 @@ public class CanalController {
return globalConfig;
}
private CanalConfigClient getManagerClient(String managerAddress) {
return new CanalConfigClient();
}
private BeanFactory getBeanFactory(String springXml) {
ApplicationContext applicationContext = new ClassPathXmlApplicationContext(springXml);
return applicationContext;
private PlainCanalConfigClient getManagerClient(String managerAddress) {
return new PlainCanalConfigClient(managerAddress, this.adminUser, this.adminPasswd);
}
private void initInstanceConfig(Properties properties) {
@@ -392,9 +399,13 @@ public class CanalController {
}
private InstanceConfig parseInstanceConfig(Properties properties, String destination) {
String adminManagerAddress = getProperty(properties, CanalConstants.CANAL_ADMIN_MANAGER);
InstanceConfig config = new InstanceConfig(globalInstanceConfig);
String modeStr = getProperty(properties, CanalConstants.getInstanceModeKey(destination));
if (!StringUtils.isEmpty(modeStr)) {
if (StringUtils.isNotEmpty(adminManagerAddress)) {
// 如果指定了manager地址,则强制适用manager
config.setMode(InstanceMode.MANAGER);
} else if (StringUtils.isNotEmpty(modeStr)) {
config.setMode(InstanceMode.valueOf(StringUtils.upperCase(modeStr)));
}
@@ -406,6 +417,9 @@ public class CanalController {
if (config.getMode().isManager()) {
String managerAddress = getProperty(properties, CanalConstants.getInstanceManagerAddressKey(destination));
if (StringUtils.isNotEmpty(managerAddress)) {
if (StringUtils.equals(managerAddress, "${canal.admin.manager}")) {
managerAddress = adminManagerAddress;
}
config.setManagerAddress(managerAddress);
}
} else if (config.getMode().isSpring()) {

79
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java
View File

@@ -3,14 +3,17 @@ package com.alibaba.otter.canal.deployer;
import java.io.FileInputStream;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.deployer.monitor.remote.RemoteCanalConfigMonitor;
import com.alibaba.otter.canal.deployer.monitor.remote.RemoteConfigLoader;
import com.alibaba.otter.canal.deployer.monitor.remote.RemoteConfigLoaderFactory;
import com.alibaba.otter.canal.common.utils.NamedThreadFactory;
import com.alibaba.otter.canal.instance.manager.plain.PlainCanal;
import com.alibaba.otter.canal.instance.manager.plain.PlainCanalConfigClient;
/**
* canal独立版本启动的入口类
@@ -20,9 +23,11 @@ import com.alibaba.otter.canal.deployer.monitor.remote.RemoteConfigLoaderFactory
*/
public class CanalLauncher {
private static final String CLASSPATH_URL_PREFIX = "classpath:";
private static final Logger logger = LoggerFactory.getLogger(CanalLauncher.class);
public static final CountDownLatch runningLatch = new CountDownLatch(1);
private static final String CLASSPATH_URL_PREFIX = "classpath:";
private static final Logger logger = LoggerFactory.getLogger(CanalLauncher.class);
public static final CountDownLatch runningLatch = new CountDownLatch(1);
private static ScheduledExecutorService executor = Executors.newScheduledThreadPool(1,
new NamedThreadFactory("canal-server-scan"));
public static void main(String[] args) {
try {
@@ -32,7 +37,6 @@ public class CanalLauncher {
logger.info("## load canal configurations");
String conf = System.getProperty("canal.conf", "classpath:canal.properties");
Properties properties = new Properties();
RemoteConfigLoader remoteConfigLoader = null;
if (conf.startsWith(CLASSPATH_URL_PREFIX)) {
conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX);
properties.load(CanalLauncher.class.getClassLoader().getResourceAsStream(conf));
@@ -40,43 +44,48 @@ public class CanalLauncher {
properties.load(new FileInputStream(conf));
}
remoteConfigLoader = RemoteConfigLoaderFactory.getRemoteConfigLoader(properties);
if (remoteConfigLoader != null) {
// 加载远程canal.properties
Properties remoteConfig = remoteConfigLoader.loadRemoteConfig();
// 加载remote instance配置
remoteConfigLoader.loadRemoteInstanceConfigs();
if (remoteConfig != null) {
properties = remoteConfig;
} else {
remoteConfigLoader = null;
}
}
final CanalStater canalStater = new CanalStater(properties);
canalStater.start();
String managerAddress = properties.getProperty(CanalConstants.CANAL_ADMIN_MANAGER);
if (StringUtils.isNotEmpty(managerAddress)) {
String user = properties.getProperty(CanalConstants.CANAL_ADMIN_USER);
String passwd = properties.getProperty(CanalConstants.CANAL_ADMIN_PASSWD);
final PlainCanalConfigClient configClient = new PlainCanalConfigClient(managerAddress, user, passwd);
PlainCanal canalConfig = configClient.findServer(null);
properties = canalConfig.getProperties();
int scanIntervalInSecond = Integer.valueOf(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL,
"5"));
executor.scheduleWithFixedDelay(new Runnable() {
if (remoteConfigLoader != null) {
remoteConfigLoader.startMonitor(new RemoteCanalConfigMonitor() {
private PlainCanal lastCanalConfig;
@Override
public void onChange(Properties properties) {
public void run() {
try {
// 远程配置canal.properties修改重新加载整个应用
canalStater.stop();
canalStater.setProperties(properties);
canalStater.start();
} catch (Throwable throwable) {
logger.error(throwable.getMessage(), throwable);
if (lastCanalConfig == null) {
lastCanalConfig = configClient.findServer(null);
} else {
PlainCanal newCanalConfig = configClient.findServer(lastCanalConfig.getMd5());
if (newCanalConfig != null) {
// 远程配置canal.properties修改重新加载整个应用
canalStater.stop();
canalStater.setProperties(newCanalConfig.getProperties());
canalStater.start();
lastCanalConfig = newCanalConfig;
}
}
} catch (Throwable e) {
logger.error("scan failed", e);
}
}
});
}, 0, scanIntervalInSecond, TimeUnit.SECONDS);
}
canalStater.setProperties(properties);
canalStater.start();
runningLatch.await();
if (remoteConfigLoader != null) {
remoteConfigLoader.destroy();
}
executor.shutdownNow();
} catch (Throwable e) {
logger.error("## Something goes wrong when starting up the canal Server:", e);
}

22
deployer/src/main/java/com/alibaba/otter/canal/deployer/admin/CanalAdminController.java
View File

@@ -19,6 +19,7 @@ import com.alibaba.otter.canal.deployer.CanalStater;
import com.alibaba.otter.canal.deployer.InstanceConfig;
import com.alibaba.otter.canal.deployer.monitor.InstanceAction;
import com.alibaba.otter.canal.deployer.monitor.InstanceConfigMonitor;
import com.alibaba.otter.canal.deployer.monitor.ManagerInstanceConfigMonitor;
import com.alibaba.otter.canal.deployer.monitor.SpringInstanceConfigMonitor;
import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.protocol.SecurityUtil;
@@ -204,9 +205,24 @@ public class CanalAdminController implements CanalAdmin {
private InstanceAction getInstanceAction(String destination) {
Map<InstanceConfig.InstanceMode, InstanceConfigMonitor> monitors = canalStater.getController()
.getInstanceConfigMonitors();
SpringInstanceConfigMonitor monitor = (SpringInstanceConfigMonitor) monitors.get(InstanceConfig.InstanceMode.SPRING);
Map<String, InstanceAction> instanceActions = monitor.getActions();
return instanceActions.get(destination);
InstanceAction instanceAction = null;
if (monitors.containsKey(InstanceConfig.InstanceMode.SPRING)) {
SpringInstanceConfigMonitor monitor = (SpringInstanceConfigMonitor) monitors.get(InstanceConfig.InstanceMode.SPRING);
Map<String, InstanceAction> instanceActions = monitor.getActions();
instanceAction = instanceActions.get(destination);
}
if (instanceAction != null) {
return instanceAction;
}
if (monitors.containsKey(InstanceConfig.InstanceMode.MANAGER)) {
ManagerInstanceConfigMonitor monitor = (ManagerInstanceConfigMonitor) monitors.get(InstanceConfig.InstanceMode.MANAGER);
Map<String, InstanceAction> instanceActions = monitor.getActions();
instanceAction = instanceActions.get(destination);
}
return instanceAction;
}
public void setUser(String user) {

171
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerInstanceConfigMonitor.java
View File

@@ -1,20 +1,185 @@
package com.alibaba.otter.canal.deployer.monitor;
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.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.common.CanalLifeCycle;
import com.alibaba.otter.canal.common.utils.NamedThreadFactory;
import com.alibaba.otter.canal.instance.manager.plain.PlainCanal;
import com.alibaba.otter.canal.instance.manager.plain.PlainCanalConfigClient;
import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.google.common.collect.MapMaker;
import com.google.common.collect.MigrateMap;
/**
* @author jianghang 2013-2-18 下午03:19:06
* @version 1.0.1
* 基于manager配置的实现
*
* @author agapple 2019年8月26日 下午10:00:20
* @since 1.1.4
*/
public class ManagerInstanceConfigMonitor extends AbstractCanalLifeCycle implements InstanceConfigMonitor, CanalLifeCycle {
public void register(String destination, InstanceAction action) {
private static final Logger logger = LoggerFactory.getLogger(ManagerInstanceConfigMonitor.class);
private long scanIntervalInSecond = 5;
private InstanceAction defaultAction = null;
private Map<String, InstanceAction> actions = new MapMaker().makeMap();
private Map<String, PlainCanal> configs = MigrateMap.makeComputingMap(new Function<String, PlainCanal>() {
public PlainCanal apply(String destination) {
return new PlainCanal();
}
});
private ScheduledExecutorService executor = Executors.newScheduledThreadPool(1,
new NamedThreadFactory("canal-instance-scan"));
private volatile boolean isFirst = true;
private PlainCanalConfigClient configClient;
private String ip;
private int port;
private String lastInstanceMD5;
public void start() {
super.start();
executor.scheduleWithFixedDelay(new Runnable() {
public void run() {
try {
scan();
if (isFirst) {
isFirst = false;
}
} catch (Throwable e) {
logger.error("scan failed", e);
}
}
}, 0, scanIntervalInSecond, TimeUnit.SECONDS);
}
public void stop() {
super.stop();
executor.shutdownNow();
actions.clear();
}
public void register(String destination, InstanceAction action) {
if (action != null) {
actions.put(destination, action);
} else {
actions.put(destination, defaultAction);
}
}
public void unregister(String destination) {
actions.remove(destination);
}
private void scan() {
String instances = configClient.findInstances(ip, String.valueOf(port), lastInstanceMD5);
final List<String> is = Lists.newArrayList(StringUtils.split(instances, ','));
List<String> start = Lists.newArrayList();
List<String> stop = Lists.newArrayList();
List<String> restart = Lists.newArrayList();
for (String instance : is) {
if (!configs.containsKey(instance)) {
PlainCanal newPlainCanal = configClient.findInstance(instance, null);
if (newPlainCanal != null) {
configs.put(instance, newPlainCanal);
start.add(instance);
}
} else {
PlainCanal plainCanal = configs.get(instance);
PlainCanal newPlainCanal = configClient.findInstance(instance, plainCanal.getMd5());
if (newPlainCanal != null) {
// 配置有变化
restart.add(instance);
configs.put(instance, newPlainCanal);
}
}
}
configs.forEach((instance, plainCanal) -> {
if (!is.contains(instance)) {
stop.add(instance);
}
});
stop.forEach(instance -> {
notifyStop(instance);
configs.remove(instance);
});
restart.forEach(instance -> {
notifyReload(instance);
});
start.forEach(instance -> {
notifyStart(instance);
});
}
private void notifyStart(String destination) {
try {
defaultAction.start(destination);
actions.put(destination, defaultAction);
// 启动成功后记录配置文件信息
} catch (Throwable e) {
logger.error(String.format("scan add found[%s] but start failed", destination), e);
}
}
private void notifyStop(String destination) {
InstanceAction action = actions.remove(destination);
try {
action.stop(destination);
} catch (Throwable e) {
logger.error(String.format("scan delete found[%s] but stop failed", destination), e);
actions.put(destination, action);// 再重新加回去，下一次scan时再执行删除
}
}
private void notifyReload(String destination) {
InstanceAction action = actions.get(destination);
if (action != null) {
try {
action.reload(destination);
} catch (Throwable e) {
logger.error(String.format("scan reload found[%s] but reload failed", destination), e);
}
}
}
public void setDefaultAction(InstanceAction defaultAction) {
this.defaultAction = defaultAction;
}
public void setScanIntervalInSecond(long scanIntervalInSecond) {
this.scanIntervalInSecond = scanIntervalInSecond;
}
public void setConfigClient(PlainCanalConfigClient configClient) {
this.configClient = configClient;
}
public void setIp(String ip) {
this.ip = ip;
}
public void setPort(int port) {
this.port = port;
}
public Map<String, InstanceAction> getActions() {
return actions;
}
}

4
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java
View File

@@ -186,8 +186,6 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
newFileInfo.add(new FileInfo(instanceConfig.getName(), instanceConfig.lastModified()));
}
lastFile.setInstanceFiles(newFileInfo);
logger.info("auto notify start {} successful.", destination);
} catch (Throwable e) {
logger.error(String.format("scan add found[%s] but start failed", destination), e);
}
@@ -198,7 +196,6 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
try {
action.stop(destination);
lastFiles.remove(destination);
logger.info("auto notify stop {} successful.", destination);
} catch (Throwable e) {
logger.error(String.format("scan delete found[%s] but stop failed", destination), e);
actions.put(destination, action);// 再重新加回去，下一次scan时再执行删除
@@ -210,7 +207,6 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
if (action != null) {
try {
action.reload(destination);
logger.info("auto notify reload {} successful.", destination);
} catch (Throwable e) {
logger.error(String.format("scan reload found[%s] but reload failed", destination), e);
}

47
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/ConfigItem.java
View File

@@ -1,47 +0,0 @@
package com.alibaba.otter.canal.deployer.monitor.remote;
/**
* 配置对应对象
*
* @author rewerma 2019-01-25 下午05:20:16
* @version 1.0.0
*/
public 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;
}
}

264
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/DbRemoteConfigLoader.java
View File

@@ -1,264 +0,0 @@
package com.alibaba.otter.canal.deployer.monitor.remote;
import java.io.ByteArrayInputStream;
import java.io.FileWriter;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.otter.canal.common.utils.CommonUtils;
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 2019-01-25 下午05:20:16
* @version 1.0.0
*/
public class DbRemoteConfigLoader implements RemoteConfigLoader {
private static final Logger logger = LoggerFactory.getLogger(DbRemoteConfigLoader.class);
private Map<String, ConfigItem> remoteInstanceConfigs = new MapMaker().makeMap();
private DruidDataSource dataSource;
private long currentConfigTimestamp = 0;
private long scanIntervalInSecond = 5;
private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2,
new NamedThreadFactory("remote-canal-config-scan"));
private RemoteInstanceMonitor remoteInstanceMonitor = new RemoteInstanceMonitorImpl();
public DbRemoteConfigLoader(String driverName, String jdbcUrl, String jdbcUsername, String jdbcPassword){
dataSource = new DruidDataSource();
if (StringUtils.isEmpty(driverName)) {
driverName = "com.mysql.jdbc.Driver";
}
dataSource.setDriverClassName(driverName);
dataSource.setUrl(jdbcUrl);
dataSource.setUsername(jdbcUsername);
dataSource.setPassword(jdbcPassword);
dataSource.setInitialSize(1);
dataSource.setMinIdle(1);
dataSource.setMaxActive(1);
dataSource.setMaxWait(60000);
dataSource.setTimeBetweenEvictionRunsMillis(60000);
dataSource.setMinEvictableIdleTimeMillis(300000);
try {
dataSource.init();
} catch (SQLException e) {
throw new RuntimeException(e.getMessage(), e);
}
}
/**
* 加载远程 canal.properties文件
*
* @return 远程配置的properties
*/
@Override
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
.parseInt(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL, "5"));
logger.info("## Loaded remote canal config: canal.properties ");
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
return properties;
}
/**
* 覆盖本地 canal.properties
*
* @param content 远程配置内容文本
*/
private void overrideLocalCanalConfig(String content) {
try (FileWriter writer = new FileWriter(CommonUtils.getConfPath() + "canal.properties")) {
writer.write(content);
writer.flush();
} 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 (Connection conn = dataSource.getConnection();
Statement stmt = conn.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;
}
/**
* 加载远程的instance配置
*/
@Override
public void loadRemoteInstanceConfigs() {
try {
// 加载远程instance配置
loadModifiedInstanceConfigs();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
/**
* 加载远程instance新增、修改、删除配置
*/
private void loadModifiedInstanceConfigs() {
Map<String, ConfigItem> remoteConfigStatus = new HashMap<>();
String sql = "select id, name, modified_time from canal_instance_config";
try (Connection conn = dataSource.getConnection();
Statement stmt = conn.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);
}
if (!remoteConfigStatus.isEmpty()) {
List<Long> 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()) {
String contentsSql = "select id, name, content, modified_time from canal_instance_config where id in ("
+ Joiner.on(",").join(changedIds) + ")";
try (Connection conn = dataSource.getConnection();
Statement stmt = conn.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);
remoteInstanceMonitor.onModify(configItemNew);
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
}
for (String name : remoteInstanceConfigs.keySet()) {
if (!remoteConfigStatus.containsKey(name)) {
// 删除
remoteInstanceConfigs.remove(name);
remoteInstanceMonitor.onDelete(name);
}
}
}
/**
* 监听 canal 主配置和 instance 配置变化
*
* @param remoteCanalConfigMonitor 监听回调方法
*/
public void startMonitor(final RemoteCanalConfigMonitor remoteCanalConfigMonitor) {
// 监听canal.properties变化
executor.scheduleWithFixedDelay(new Runnable() {
public void run() {
try {
Properties properties = loadRemoteConfig();
if (properties != null) {
remoteCanalConfigMonitor.onChange(properties);
}
} catch (Throwable e) {
logger.error("Scan remote canal config failed", e);
}
}
}, 10, scanIntervalInSecond, TimeUnit.SECONDS);
// 监听instance变化
executor.scheduleWithFixedDelay(new Runnable() {
public void run() {
try {
loadRemoteInstanceConfigs();
} catch (Throwable e) {
logger.error("Scan remote instance config failed", e);
}
}
}, 10, 3, TimeUnit.SECONDS);
}
/**
* 销毁
*/
public void destroy() {
executor.shutdownNow();
try {
dataSource.close();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
}

13
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteCanalConfigMonitor.java
View File

@@ -1,13 +0,0 @@
package com.alibaba.otter.canal.deployer.monitor.remote;
import java.util.Properties;
/**
* 远程canal.properties配置监听器接口
*
* @author rewerma 2019-01-25 下午05:20:16
* @version 1.0.0
*/
public interface RemoteCanalConfigMonitor {
void onChange(Properties properties);
}

36
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoader.java
View File

@@ -1,36 +0,0 @@
package com.alibaba.otter.canal.deployer.monitor.remote;
import java.util.Properties;
/**
* 远程配置装载器接口
*
* @author rewerma 2019-01-25 下午05:20:16
* @version 1.0.0
*/
public interface RemoteConfigLoader {
/**
* 加载远程 canal.properties文件
*
* @return 远程配置的properties
*/
Properties loadRemoteConfig();
/**
* 加载远程的instance配置
*/
void loadRemoteInstanceConfigs();
/**
* 启动监听 canal 主配置和 instance 配置变化
*
* @param remoteCanalConfigMonitor 监听回调方法
*/
void startMonitor(final RemoteCanalConfigMonitor remoteCanalConfigMonitor);
/**
* 销毁
*/
void destroy();
}

42
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoaderFactory.java
View File

@@ -1,42 +0,0 @@
package com.alibaba.otter.canal.deployer.monitor.remote;
import java.util.Properties;
import com.alibaba.otter.canal.deployer.monitor.remote.http.HttpRemoteConfigLoader;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* 远程配置装载器工厂类
*
* @author rewerma 2019-01-25 下午05:20:16
* @version 1.0.0
*/
public class RemoteConfigLoaderFactory {
private static final Logger logger = LoggerFactory.getLogger(RemoteConfigLoaderFactory.class);
public static RemoteConfigLoader getRemoteConfigLoader(Properties localProperties) {
String jdbcUrl = localProperties.getProperty("canal.manager.jdbc.url");
String httpUrl = localProperties.getProperty("canal.manager.http.url");
if (!StringUtils.isEmpty(jdbcUrl)) {
logger.info("## load remote db canal configurations");
// load remote config
String driverName = localProperties.getProperty("canal.manager.jdbc.driverName");
String jdbcUsername = localProperties.getProperty("canal.manager.jdbc.username");
String jdbcPassword = localProperties.getProperty("canal.manager.jdbc.password");
return new DbRemoteConfigLoader(driverName, jdbcUrl, jdbcUsername, jdbcPassword);
} else if (!StringUtils.isEmpty(httpUrl)) {
logger.info("## load remote http canal configurations");
String httpUsername = localProperties.getProperty("canal.manager.http.username");
String httpPassword = localProperties.getProperty("canal.manager.http.password");
return new HttpRemoteConfigLoader(httpUrl, httpUsername, httpPassword);
}
// 可扩展其它远程配置加载器
logger.info("## load local canal configurations");
return null;
}
}

31
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitor.java
View File

@@ -1,31 +0,0 @@
package com.alibaba.otter.canal.deployer.monitor.remote;
/**
* 远程xxx/instance.properties配置监听器接口
*
* @author rewerma 2019-01-25 下午05:20:16
* @version 1.0.0
*/
public interface RemoteInstanceMonitor {
/**
* 新增配置事件
*
* @param configItem 配置项
*/
void onAdd(ConfigItem configItem);
/**
* 修改配置事件
*
* @param configItem 配置项
*/
void onModify(ConfigItem configItem);
/**
* 删除配置事件
*
* @param instanceName 实例名
*/
void onDelete(String instanceName);
}

55
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitorImpl.java
View File

@@ -1,55 +0,0 @@
package com.alibaba.otter.canal.deployer.monitor.remote;
import java.io.File;
import java.io.FileWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.common.utils.CommonUtils;
/**
* 远程xxx/instance.properties配置监听器实现
*
* @author rewerma 2019-01-25 下午05:20:16
* @version 1.0.0
*/
public class RemoteInstanceMonitorImpl implements RemoteInstanceMonitor {
private static final Logger logger = LoggerFactory.getLogger(RemoteInstanceMonitorImpl.class);
@Override
public void onAdd(ConfigItem configItem) {
this.onModify(configItem);
}
@Override
public void onModify(ConfigItem configItem) {
String confDir = CommonUtils.getConfPath() + configItem.getName();
File instanceDir = new File(confDir);
if (!instanceDir.exists()) {
boolean mkDirs = instanceDir.mkdirs();
if (!mkDirs) {
logger.info("## Error to create instance config dir: {}", configItem.getName());
return;
}
}
try (FileWriter writer = new FileWriter(confDir + "/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);
}
}
@Override
public void onDelete(String instanceName) {
File file = new File(CommonUtils.getConfPath() + instanceName + "/");
if (file.exists()) {
CommonUtils.deleteDir(file);
logger.info("## Deleted and loaded remote instance config: {} ", instanceName);
}
}
}

165
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/HttpHelper.java
View File

@@ -1,165 +0,0 @@
package com.alibaba.otter.canal.deployer.monitor.remote.http;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.alibaba.otter.canal.deployer.monitor.remote.ConfigItem;
import org.apache.http.HttpStatus;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.protocol.HttpClientContext;
import org.apache.http.client.utils.URIBuilder;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.URI;
import java.util.Map;
import static org.apache.http.client.config.RequestConfig.custom;
/**
* http client 工具类
*
* @author rewerma 2019-08-26 上午09:40:36
* @version 1.0.0
*/
public class HttpHelper {
private final static Logger logger = LoggerFactory.getLogger(HttpRemoteConfigLoader.class);
public static final Integer REST_STATE_OK = 20000;
public static final Integer REST_STATE_TOKEN_INVALID = 50014;
public static final Integer REST_STATE_ERROR = 50000;
private CloseableHttpClient httpclient;
public HttpHelper(){
HttpClientBuilder builder = HttpClientBuilder.create();
builder.setMaxConnPerRoute(50);
builder.setMaxConnTotal(100);
httpclient = builder.build();
}
public String get(String url, Map<String, String> heads, int timeout) {
// 支持采用https协议，忽略证书
url = url.trim();
if (url.startsWith("https")) {
// FIXME
return "";
}
CloseableHttpResponse response = null;
HttpGet httpGet = null;
try {
URI uri = new URIBuilder(url).build();
RequestConfig config = custom().setConnectTimeout(timeout)
.setConnectionRequestTimeout(timeout)
.setSocketTimeout(timeout)
.build();
httpGet = new HttpGet(uri);
if (heads != null) {
for (Map.Entry<String, String> entry : heads.entrySet()) {
httpGet.setHeader(entry.getKey(), entry.getValue());
}
}
HttpClientContext context = HttpClientContext.create();
context.setRequestConfig(config);
response = httpclient.execute(httpGet, context);
int statusCode = response.getStatusLine().getStatusCode();
if (statusCode == HttpStatus.SC_OK) {
return EntityUtils.toString(response.getEntity());
} else {
String errorMsg = EntityUtils.toString(response.getEntity());
throw new RuntimeException("requestGet remote error, url=" + uri.toString() + ", code=" + statusCode
+ ", error msg=" + errorMsg);
}
} catch (Throwable t) {
throw new RuntimeException("requestGet remote error, request : " + url, t);
} finally {
if (response != null) {
try {
response.close();
} catch (IOException e) {
// ignore
}
}
if (httpGet != null) {
httpGet.releaseConnection();
}
}
}
public String post(String url, Map<String, String> heads, Object requestBody, int timeout) {
String json = JSON.toJSONString(requestBody);
return post0(url, heads, json, timeout);
}
public String post0(String url, Map<String, String> heads, String requestBody, int timeout) {
url = url.trim();
// 支持采用https协议，忽略证书
if (url.startsWith("https")) {
// FIXME
return "";
}
HttpPost httpPost = null;
CloseableHttpResponse response = null;
try {
URI uri = new URIBuilder(url).build();
RequestConfig config = custom().setConnectTimeout(timeout)
.setConnectionRequestTimeout(timeout)
.setSocketTimeout(timeout)
.build();
httpPost = new HttpPost(uri);
StringEntity entity = new StringEntity(requestBody, "UTF-8");
httpPost.setEntity(entity);
httpPost.setHeader("Content-Type", "application/json;charset=utf8");
if (heads != null) {
for (Map.Entry<String, String> entry : heads.entrySet()) {
httpPost.setHeader(entry.getKey(), entry.getValue());
}
}
HttpClientContext context = HttpClientContext.create();
context.setRequestConfig(config);
response = httpclient.execute(httpPost, context);
int statusCode = response.getStatusLine().getStatusCode();
if (statusCode == HttpStatus.SC_OK) {
return EntityUtils.toString(response.getEntity());
} else {
throw new RuntimeException("requestPost remote error, request : " + url + ", statusCode=" + statusCode
+ ";" + EntityUtils.toString(response.getEntity()));
}
} catch (Throwable t) {
throw new RuntimeException("requestPost remote error, request : " + url, t);
} finally {
if (response != null) {
try {
response.close();
} catch (IOException e) {
// ignore
}
}
if (httpPost != null) {
httpPost.releaseConnection();
}
}
}
public void close() {
if (httpclient != null) {
try {
httpclient.close();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
}
}

257
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/HttpRemoteConfigLoader.java
View File

@@ -1,257 +0,0 @@
package com.alibaba.otter.canal.deployer.monitor.remote.http;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.alibaba.otter.canal.common.utils.CommonUtils;
import com.alibaba.otter.canal.common.utils.NamedThreadFactory;
import com.alibaba.otter.canal.deployer.CanalConstants;
import com.alibaba.otter.canal.deployer.monitor.remote.*;
import com.google.common.collect.MapMaker;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.ByteArrayInputStream;
import java.io.FileWriter;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* 基于HTTP的远程配置装载器
*
* @author rewerma 2019-08-26 上午09:40:36
* @version 1.0.0
*/
public class HttpRemoteConfigLoader implements RemoteConfigLoader {
private final static Logger logger = LoggerFactory.getLogger(HttpRemoteConfigLoader.class);
private final static Integer REQUEST_TIMEOUT = 5000;
private String baseUrl;
private String username;
private String password;
private String token;
private HttpHelper httpHelper;
private long currentConfigTimestamp = 0;
private Map<String, ConfigItem> remoteInstanceConfigs = new MapMaker().makeMap();
private RemoteInstanceMonitor remoteInstanceMonitor = new RemoteInstanceMonitorImpl();
private long scanIntervalInSecond = 5;
private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2,
new NamedThreadFactory("remote-http-canal-config-scan"));
public HttpRemoteConfigLoader(String baseUrl, String username, String password){
this.baseUrl = baseUrl;
this.username = username;
this.password = password;
httpHelper = new HttpHelper();
}
private String login4Token(HttpHelper httpHelper) {
Map<String, Object> reqBody = new HashMap<>();
reqBody.put("username", username);
reqBody.put("password", password);
String response = httpHelper.post(baseUrl + "/api/v1/user/login", null, reqBody, REQUEST_TIMEOUT);
ResponseModel<JSONObject> resp = JSONObject.parseObject(response,
new TypeReference<ResponseModel<JSONObject>>() {
});
if (!HttpHelper.REST_STATE_OK.equals(resp.getCode())) {
throw new RuntimeException("requestPost for login error: " + resp.getMessage());
}
return (String) resp.getData().get("token");
}
@Override
public synchronized Properties loadRemoteConfig() {
Properties properties = null;
try {
ConfigItem configItem = getRemoteCanalConfig();
if (configItem != null) {
if (configItem.getModifiedTime() != currentConfigTimestamp) { // 修改时就不同说明配置有更新
Map<String, String> heads = new HashMap<>();
heads.put("X-Token", token);
String response = httpHelper.get(baseUrl + "/api/v1/canal/config", heads, REQUEST_TIMEOUT);
ResponseModel<ConfigItem> resp = JSONObject.parseObject(response,
new TypeReference<ResponseModel<ConfigItem>>() {
});
currentConfigTimestamp = configItem.getModifiedTime();
overrideLocalCanalConfig(resp.getData().getContent());
properties = new Properties();
properties
.load(new ByteArrayInputStream(resp.getData().getContent().getBytes(StandardCharsets.UTF_8)));
scanIntervalInSecond = Integer
.parseInt(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL, "5"));
logger.info("## Loaded http remote canal config: canal.properties ");
}
}
} catch (Throwable t) {
logger.error(t.getMessage(), t);
}
return properties;
}
/**
* 获取远程canal server配置摘要(无配置内容)
*
* @return 配置摘要
*/
private ConfigItem getRemoteCanalConfig() {
if (StringUtils.isEmpty(token)) {
token = login4Token(httpHelper);
}
Map<String, String> heads = new HashMap<>();
heads.put("X-Token", token);
String response = httpHelper.get(baseUrl + "/api/v1/canal/config/summary", heads, REQUEST_TIMEOUT);
ResponseModel<ConfigItem> resp = JSONObject.parseObject(response,
new TypeReference<ResponseModel<ConfigItem>>() {
});
if (HttpHelper.REST_STATE_TOKEN_INVALID.equals(resp.getCode())) {
// token 失效
token = login4Token(httpHelper);
heads.put("X-Token", token);
response = httpHelper.get(baseUrl + "/api/v1/canal/config/summary", heads, REQUEST_TIMEOUT);
resp = JSONObject.parseObject(response, new TypeReference<ResponseModel<ConfigItem>>() {
});
}
if (!HttpHelper.REST_STATE_OK.equals(resp.getCode())) {
throw new RuntimeException("requestGet for canal config error: " + resp.getMessage());
}
return resp.getData();
}
@Override
public synchronized void loadRemoteInstanceConfigs() {
try {
// 加载远程instance配置
loadModifiedInstanceConfigs();
} catch (Throwable t) {
logger.error(t.getMessage(), t);
}
}
/**
* 加载远程instance新增、修改、删除配置
*/
private void loadModifiedInstanceConfigs() {
if (StringUtils.isEmpty(token)) {
token = login4Token(httpHelper);
}
Map<String, ConfigItem> remoteConfigStatus = new HashMap<>();
Map<String, String> heads = new HashMap<>();
heads.put("X-Token", token);
String response = httpHelper.get(baseUrl + "/api/v1/canal/instances", heads, REQUEST_TIMEOUT);
ResponseModel<List<ConfigItem>> resp = JSONObject.parseObject(response,
new TypeReference<ResponseModel<List<ConfigItem>>>() {
});
if (HttpHelper.REST_STATE_TOKEN_INVALID.equals(resp.getCode())) {
// token 失效
token = login4Token(httpHelper);
heads.put("X-Token", token);
response = httpHelper.get(baseUrl + "/api/v1/canal/instances", heads, REQUEST_TIMEOUT);
resp = JSONObject.parseObject(response, new TypeReference<ResponseModel<List<ConfigItem>>>() {
});
}
if (!HttpHelper.REST_STATE_OK.equals(resp.getCode())) {
throw new RuntimeException("requestGet for canal instances error: " + resp.getMessage());
}
for (ConfigItem configItem : resp.getData()) {
remoteConfigStatus.put(configItem.getName(), configItem);
}
if (!remoteConfigStatus.isEmpty()) {
List<Long> 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()) {
for (Long changedId : changedIds) {
String response2 = httpHelper
.get(baseUrl + "/api/v1/canal/instance/" + changedId, heads, REQUEST_TIMEOUT);
ResponseModel<ConfigItem> resp2 = JSONObject.parseObject(response2,
new TypeReference<ResponseModel<ConfigItem>>() {
});
ConfigItem configItemNew = resp2.getData();
remoteInstanceConfigs.put(configItemNew.getName(), configItemNew);
remoteInstanceMonitor.onModify(configItemNew);
}
}
}
for (String name : remoteInstanceConfigs.keySet()) {
if (!remoteConfigStatus.containsKey(name)) {
// 删除
remoteInstanceConfigs.remove(name);
remoteInstanceMonitor.onDelete(name);
}
}
}
@Override
public synchronized void startMonitor(RemoteCanalConfigMonitor remoteCanalConfigMonitor) {
executor.scheduleWithFixedDelay(new Runnable() {
public void run() {
// 监听canal.properties变化
try {
Properties properties = loadRemoteConfig();
if (properties != null) {
remoteCanalConfigMonitor.onChange(properties);
}
} catch (Throwable e) {
logger.error("Scan remote canal config failed", e);
}
// 监听instance变化
try {
loadRemoteInstanceConfigs();
} catch (Throwable e) {
logger.error("Scan remote instance config failed", e);
}
}
}, 10, scanIntervalInSecond, TimeUnit.SECONDS);
}
/**
* 覆盖本地 canal.properties
*
* @param content 远程配置内容文本
*/
private void overrideLocalCanalConfig(String content) {
try (FileWriter writer = new FileWriter(CommonUtils.getConfPath() + "canal.properties")) {
writer.write(content);
writer.flush();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
@Override
public synchronized void destroy() {
if (httpHelper != null) {
httpHelper.close();
}
}
}

38
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/ResponseModel.java
View File

@@ -1,38 +0,0 @@
package com.alibaba.otter.canal.deployer.monitor.remote.http;
/**
* 响应类
*
* @author rewerma 2019-08-26 上午09:40:36
* @version 1.0.0
*/
public class ResponseModel<T> {
private Integer code;
private String message;
private T data;
public Integer getCode() {
return code;
}
public void setCode(Integer code) {
this.code = code;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
public T getData() {
return data;
}
public void setData(T data) {
this.data = data;
}
}

10
deployer/src/main/resources/canal.properties
View File

@@ -1,13 +1,6 @@
#################################################
######### 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.manager.http.url=http://127.0.0.1:8089
canal.manager.http.username=admin
canal.manager.http.password=121212
canal.id = 1
# tcp bind ip
canal.ip =
@@ -20,6 +13,7 @@ canal.metrics.pull.port = 11112
# canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458
# canal admin config
#canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
@@ -112,7 +106,7 @@ canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
canal.instance.global.mode = spring
canal.instance.global.lazy = false
#canal.instance.global.manager.address = 127.0.0.1:1099
canal.instance.global.manager.address = ${canal.admin.manager}
#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
canal.instance.global.spring.xml = classpath:spring/file-instance.xml
#canal.instance.global.spring.xml = classpath:spring/default-instance.xml