一、canal的程序入口
在官网给的GitHub里面的deployer模块里面,官网:https://github.com/alibaba/canal/tree/master/deployer
其中的main方法在:com.alibaba.otter.canal.deployer.CanalLauncher类里面
二、加载canal的全局配置文件
追踪源码发现canal加载配置文件有两种方法:一种加载的是本地的./conf/canal.properties;另一种是加载一个指定数据库的某个字段的值。第一种:称为本地加载,第二种是远程加载。
1、本地加载(默认加载方式)
官网的简易教程就是本地加载的方式:https://github.com/alibaba/canal
2、远程加载
远程加载的配置内容是在一个canal_manager的数据库里面,里面有canal_config、canal_instance_config表
(1)、配置远程加载的数据库的信息,修改./conf/canal.properties文件里面的前三行(加载了远程配置文件的就会覆盖本地的)
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
(2)、创建远程数据库和表
create database if not EXISTS canal_manager;
create table if not EXISTS canal_manager.canal_config(
id BIGINT,
name VARCHAR(20),
content LONGTEXT,
modified_time TIMESTAMP
);
create table if not EXISTS canal_manager.canal_instance_config(
id BIGINT,
name VARCHAR(20),
content LONGTEXT,
modified_time TIMESTAMP
);
其中canal_config里面的content值就是canal.properties的内容,并且这个表的数据只能有一行,id必须是1
其中canal_instance_config里面的content的值是instance.properties的内容,instance的个数就是数据的行数,也就是每一行对应一个instance。这样在数据库里面修改content的值就可以实现动态的远程的加载配置文件。(配置的这个数据库服务和同步的数据库服务不是一个,ip不一样)
三、官网给的配置方式
1、canal配置方式有两种:
- ManagerCanalInstanceGenerator: 基于manager管理的配置方式,目前alibaba内部配置使用这种方式。大家可以实现CanalConfigClient,连接各自的管理系统,即可完成接入。
- SpringCanalInstanceGenerator:基于本地spring xml的配置方式,目前开源版本已经自带该功能所有代码,建议使用
针对于以上两种方式:网上更多的是第二种,目前没有看到任何关于第一种manager的配置,以下是个人对于manager的配置。
2、manager配置
这个过程需要修改源码,准备工作需要先把官网的源码下载到本地,源码clone地址:https://github.com/alibaba/canal.git
(1)、修改canal.properties文件
canal.instance.global.mode = manager
canal.instance.global.lazy = false
(2)、重写com.alibaba.otter.canal.instance.manager.CanalConfigClient类(在instance里面)
public class CanalConfigClient {
/**
* 根据对应的destinantion查询Canal信息
*/
public Canal findCanal(String destination) {
// TODO 根据自己的业务实现
//获取Apollo的配置
System.setProperty("app.id","canal");
System.setProperty("apollo.cluster","default");
System.setProperty("apollo.meta","http://*.*.*.*:8080,http://*.*.*.*:8080");
Config config = ConfigService.getConfig(destination+".properties"); //config instance is singleton for each namespace and is never null
Long slaveId = Long.parseLong(config.getProperty("canal.instance.mysql.slaveId", "123456"));
String USERNAME = config.getProperty("canal.instance.dbUsername", "canal");
String PASSWORD = config.getProperty("canal.instance.dbPassword", "canal");
String MYSQL_ADDRESS = config.getProperty("canal.instance.master.address", "*.*.*.*:3306").split(":")[0];
int port = Integer.parseInt(config.getProperty("canal.instance.master.address", "*.*.*.*:3306").split(":")[1]);
Canal canal = new Canal();
canal.setId(1L);
canal.setName(destination);
canal.setDesc("test");
CanalParameter parameter = new CanalParameter();
parameter.setZkClusters(Arrays.asList("127.0.0.1:2188"));
parameter.setMetaMode(CanalParameter.MetaMode.MEMORY);
parameter.setHaMode(CanalParameter.HAMode.HEARTBEAT);
parameter.setIndexMode(CanalParameter.IndexMode.MEMORY);
parameter.setStorageMode(CanalParameter.StorageMode.MEMORY);
parameter.setMemoryStorageBufferSize(32 * 1024);
parameter.setSourcingType(CanalParameter.SourcingType.MYSQL);
parameter.setDbAddresses(Arrays.asList(new InetSocketAddress(MYSQL_ADDRESS, 3306),
new InetSocketAddress(MYSQL_ADDRESS, 3306)));
parameter.setDbUsername(USERNAME);
parameter.setDbPassword(PASSWORD);
//parameter.setPositions(Arrays.asList("{\"journalName\":\"mysql-bin.000001\",\"position\":6163L,\"timestamp\":1322803601000L}",
// "{\"journalName\":\"mysql-bin.000001\",\"position\":6163L,\"timestamp\":1322803601000L}"));
System.out.println(USERNAME+PASSWORD);
parameter.setSlaveId(1234L);
parameter.setDefaultConnectionTimeoutInSeconds(30);
parameter.setConnectionCharset("UTF-8");
parameter.setConnectionCharsetNumber((byte) 33);
parameter.setReceiveBufferSize(8 * 1024);
parameter.setSendBufferSize(8 * 1024);
parameter.setDetectingEnable(false);
parameter.setDetectingIntervalInSeconds(10);
parameter.setDetectingRetryTimes(3);
//parameter.setDetectingSQL(DETECTING_SQL);
canal.setCanalParameter(parameter);
/* System.out.println(username+password+host+port);
CanalParameter canalParameter = new CanalParameter();
canalParameter.setSlaveId(slaveId);
canalParameter.setDbUsername(username);
canalParameter.setDbPassword(password);
canalParameter.setIndexMode(CanalParameter.IndexMode.MEMORY);
List<InetSocketAddress> dbAddresses = new ArrayList<>();
dbAddresses.add(new InetSocketAddress(host,port));
canalParameter.setDbAddresses(dbAddresses);
canal.setCanalParameter(canalParameter);
canal.setName(destination);*/
return canal;
}
/**
* 根据对应的destinantion查询filter信息
*/
public String findFilter(String destination) {
// TODO 根据自己的业务实现
Config config = ConfigService.getConfig(destination+".properties");
return config.getProperty("canal.instance.filter.regex",".*\\..*");
}
}
这两个方法的实现比较简单,但是在运行不成功,可能官网给的源码的关于manager的不是太全,毕竟阿里自己用的是manager配置文件的方式。官网也推荐使用spring的方式,自己修改源码,用manager的方式总是会有意想不到的问题(也可能自己解读源码不全)。
四、总结
对于canal来说配置文件有两部分,一个是:全局的canal.properties(也可以说是canal-server配置),另一个是:instance.properties(也可以说是canal-client配置,也即instance配置)
总的来说有四种加载配置文件的方式:
1、本地加载+spring:默认加载../conf下的所有配置文件,canal.instance.global.mode = spring
2、本地加载+manager:默认加载../conf下的所有配置文件,canal.instance.global.mode = manager,需要重写CanalConfigClient类
3、远程加载+spring:需要在数据库配置,canal_instance_config表里面canal.instance.global.mode = spring
4、远程加载+manager:需要在数据库配置,canal_instance_config表里面canal.instance.global.mode = manager,需要重写CanalConfigClient类