canal [kə'næl],译意为水道/管道/沟渠,主要用途是基于 MySQL 数据库增量日志解析,提供增量数据 订阅 和 消费。应该是阿里云DTS(Data Transfer Service)的开源版本,开源地址:
https://github.com/alibaba/canal。mysql
canal从1.1.4版本开始引入了admin控制台,有了不少不同的配置方式。在搭建过程当中若是仅仅按照wiki的用户手册,仍是容易踩不少坑的。所以,将笔者在搭建过程当中的步骤记录下来,做为官方wiki的 补充,但愿能有所帮助。git
根据本文内容与搭建顺序 ,并搭配对应的官网文档连接,应该就能快速搭建完成了,enjoy~github
官方文档地址:
https://github.com/alibaba/canal/wiki/Canal-Admin-QuickStartspring
主要配置application.yml文件sql
server: port: 8089 spring: jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 spring.datasource: address: 127.0.0.1:3306 database: canal_manager username: xxxx assword: xxxxx driver-class-name: com.mysql.jdbc.Driver url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false hikari: maximum-pool-size: 30 minimum-idle: 1 canal: adminUser: admin adminPasswd: admin
这里须要注意,canal的adminPasswd并非登录admin的密码,登录admin的密码是设置在对应的数据库中的,默认为123456。数据库
另外,由于 Canal Admin 是一个管理系统,须要使用数据库存放配置信息,只用在 MySQL 中执行 Canal Admin 提供的数据库初始化文件便可,该文件在“conf/canal_manager.sql”路径下面。json
上面的 Canal Admin 配置好了以后直接根据“/bin/startup.sh”启动 Canal Admin 便可,在浏览器上面输入 hostip:8089 便可进入到管理页面,若是使用的默认的配置信息,用户名入”admin”,密码输入”123456”便可访问首页。浏览器
进入到首页点击集群的菜单栏,而后选择新建集群。bash
在里面输入集群的名称以及 Zookeeper便可,这里的集群目前尚未任务节点,后续经过配置 Canal Server 的自动注册功能,即可以查看该集群下面拥有的节点。app
建立集群后,须要先配置集群 主配置,载入模板便可。
若是没有载入这个模版,那么在canal-server执行 sh bin/startup.sh local 命令,读取canal_local.propeties配置启动时,会报错
1 Caused by: com.alibaba.otter.canal.common.CanalException: requestGet for canal config error: canal.properties config is empty
注意,在主配置中,仍是须要记得加入zk的地址配置,跟上面配置集群名字的时候输入的zk无关 (那个zk地址不知道有啥用):
canal.zkServers = xxx.xx.xx.xx:2181,xxx.xx.xx.xx:2181,xxx.xx.xx.xx:2181
若是但愿使用canal的集群模式(推荐使用),记得更改配置使用default-instance.xml,不要使用file-instance.xml配置,以下:
#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
注意,建议先部署好admin,再来部署canal-server,省事很多。
官方文档地址:
https://github.com/alibaba/canal/wiki/QuickStart
注意,不一样于单机版的properties配置,使用admin后,各类配置经过admin全局管理,因此只须要配置canal_local.properties便可。canal_local.properties内容以下:
# register ip
canal.register.ip =
# canal admin config
canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = xxxxxxxxxxxxxx
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster =
对各个参数说明一下:
启动server,切记切记带上参数local,这样才会读取canal_local.properties的配置
sh bin/startup.sh local
启动完成后,能够在admin界面看到server的链接信息
同时,登录zk,查看是否已经注册成功。
采用admin的集群模式后,集群内的canal-server经过zk作HA,所以,canal-server的配置也只能经过集群作全局配置。
就是前面的集群配置-主配置进入。
另外,canal做为一个增量数据抓取模块,抓到变动信息后须要投递。
经过canal.sererMode配置
这里也限制了,一个集群内的canal只能支持一种投递模式。
咱们目前暂时以投递RocketMQ为例进行配置。
官方文档地址:
https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart
最终配置文件的配置以下:
#################################################
######### common argument #############
#################################################
# tcp bind ip
canal.ip =
# register ip to zookeeper
canal.register.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
# canal instance user/passwd
canal.user = xxxx
canal.passwd = xxxxxxxxxxxxxxxxxxx
# canal admin config
canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = xxxx
canal.admin.passwd = xxxxxxxxxxxxxx
canal.zkServers = xx.xx.xx.xx:xxxx,xx.xx.xx.xxx:xxxx,xx.xx.xx.xx:xxxx
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, RocketMQ
canal.serverMode = RocketMQ
# flush meta cursor/parse position to file
canal.file.data.dir = ${canal.conf.dir}
canal.file.flush.period = 1000
## memory store RingBuffer size, should be Math.pow(2,n)
canal.instance.memory.buffer.size = 16384
## memory store RingBuffer used memory unit size , default 1kb
canal.instance.memory.buffer.memunit = 1024
## meory store gets mode used MEMSIZE or ITEMSIZE
canal.instance.memory.batch.mode = MEMSIZE
canal.instance.memory.rawEntry = true
## detecing config
canal.instance.detecting.enable = false
#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
canal.instance.detecting.sql = select 1
canal.instance.detecting.interval.time = 3
canal.instance.detecting.retry.threshold = 3
canal.instance.detecting.heartbeatHaEnable = false
# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
canal.instance.transaction.size = 1024
# mysql fallback connected to new master should fallback times
canal.instance.fallbackIntervalInSeconds = 60
# network config
canal.instance.network.receiveBufferSize = 16384
canal.instance.network.sendBufferSize = 16384
canal.instance.network.soTimeout = 30
# binlog filter config
# 这里能够对订阅的消息作过滤
canal.instance.filter.druid.ddl = true
canal.instance.filter.query.dcl = false
canal.instance.filter.query.dml = false
canal.instance.filter.query.ddl = false
canal.instance.filter.table.error = false
canal.instance.filter.rows = false
canal.instance.filter.transaction.entry = false
# binlog format/image check
canal.instance.binlog.format = ROW,STATEMENT,MIXED
canal.instance.binlog.image = FULL,MINIMAL,NOBLOB
# binlog ddl isolation
canal.instance.get.ddl.isolation = false
# parallel parser config
canal.instance.parser.parallel = true
## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
#canal.instance.parser.parallelThreadSize = 16
## disruptor ringbuffer size, must be power of 2
canal.instance.parser.parallelBufferSize = 256
# table meta tsdb info
canal.instance.tsdb.enable = true
canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
canal.instance.tsdb.dbUsername = canal
canal.instance.tsdb.dbPassword = canal
# dump snapshot interval, default 24 hour
canal.instance.tsdb.snapshot.interval = 24
# purge snapshot expire , default 360 hour(15 days)
canal.instance.tsdb.snapshot.expire = 360
# aliyun ak/sk , support rds/mq
canal.aliyun.accessKey =
canal.aliyun.secretKey =
#################################################
######### destinations #############
#################################################
canal.destinations =
# conf root dir
canal.conf.dir = ../conf
# auto scan instance dir add/remove and start/stop instance
canal.auto.scan = true
canal.auto.scan.interval = 5
canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
canal.instance.global.mode = manager
canal.instance.global.lazy = false
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
##################################################
######### MQ #############
##################################################
canal.mq.servers = xx.xx.xx.xx:xxxx,xx.xx.xx.xxx:xxxx,xx.xx.xx.xx:xxxx
canal.mq.retries = 0
canal.mq.canalBatchSize = 50
canal.mq.canalGetTimeout = 100
canal.mq.flatMessage = true
canal.mq.compressionType = none
canal.mq.acks = all
#canal.mq.properties. =
canal.mq.producerGroup = test_canal_cluster
# Set this value to "cloud", if you want open message trace feature in aliyun.
canal.mq.accessChannel = local
# aliyun mq namespace
#canal.mq.namespace =
##################################################
######### Kafka Kerberos Info #############
##################################################
canal.mq.kafka.kerberos.enable = false
canal.mq.kafka.kerberos.krb5FilePath = "../conf/kerberos/krb5.conf"
canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf"
Canal Admin 提供了 Canal Instance 的管理功能。
咱们尝试经过 UI 界面添加须要监听的数据库,让该 Instance 消费 binlog 并将事件发送到 MQ。
主要修改如下配置:
保存,启动,观察日志,没有报错便可。
同时能够从 canal-server的 操做-详情 进入,查看能够看到正在运行的instance
这里须要注意,若是没有正确注册到zk,那么每一个canal-server都会有一个正在运行的instance,会致使变动投递屡次。
若是正确注册了zk,同一个集群下,每一个instance应该只在某一个cannal-server中运行。
能够查看zk路径
/otter/canal/destinations/{xxxx}/running
xxxx就是instance的 名称,running节点表示它运行在哪一个server上。
另外,目前对binlog的订阅支持gtid模式和position模式,经过如下参数设置
# enable gtid use true/false
canal.instance.gtidon=false
在demo过程当中,发现高可用模式下,gtid没法正常更新到zk,而position模式使用正常。
gtid模式在canal高可用模式下可能存在bug(参考issue:
https://github.com/alibaba/canal/issues/2616),暂时不建议使用。
在rds执行如下sql
UPDATE `album` SET `attribute`='5' WHERE `albumid`='1';
在MQ中的消息格式以下:
{
"data":[
{
"albumid":"1",
"picid":"1234",
"attribute":"5",
"lastmodified":"2020-05-15 18:13:35",
"created":"2019-09-04 18:18:51"
}
],
"database":"dbvtest",
"es":1589537615000,
"id":75,
"isDdl":false,
"mysqlType":{
"albumid":"int(10) unsigned",
"picid":"int(10) unsigned",
"attribute":"varchar(200)",
"lastmodified":"timestamp",
"created":"timestamp"
},
"old":[
{
"attribute":"2",
"lastmodified":"2020-05-15 17:42:57"
}
],
"pkNames":[
"albumid"
],
"sql":"",
"sqlType":{
"albumid":4,
"picid":4,
"attribute":12,
"lastmodified":93,
"created":93
},
"table":"album",
"ts":1589537615877,
"type":"UPDATE"
}
官方文档地址:
https://github.com/alibaba/canal/wiki/Prometheus-QuickStart
若是你已经有一套prometheus的监控体系,咱们只须要导入模板(
canal/conf/metrics/Canal_instances_tmpl.json)便可,完美~
都看到最后了,原创不易,点个关注,点个赞吧~
知识碎片从新梳理,构建Java知识图谱: github.com/saigu/JavaK…(历史文章查阅很是方便)