canal集群部署

下载canal组件： Releases · alibaba/canal · GitHub

canal组件：

canal-admin：canal控制台，可以统一管理canal服务
canal-deployer：也是canal-server：canal的一个节点服务
canal-instance： canal-server中的一个处理实例，可以处理不同的业务逻辑。

安装canal-admin

1、创建canal_manager库

根据canal-admin conf文件夹下canal_manager.sql文件创建数据库

2、修改application.yml配置文件

修改 address、database、username、password 四个参数

server:port: 8089
spring:jackson:date-format: yyyy-MM-dd HH:mm:sstime-zone: GMT+8spring.datasource:address: 127.0.0.1:63306database: canal_managerusername: rootpassword: 123456driver-class-name: com.mysql.jdbc.Driverurl: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=falsehikari:maximum-pool-size: 30minimum-idle: 1canal:adminUser: adminadminPasswd: admin

3、部署项目并启动

项目放到服务器上，进入bin目录

使用sudo chmod 777 ./startup.sh sudo chmod 777 ./stop.sh 命令添加权限

./startup.sh启动项目

127.0.0.1:8089访问项目

4、添加集群

需要集群信息和zookeeper地址

如果要多个canal集群共用zookeeper

多个canal-server集群共用一套zookeeper解决方案 - 墨天轮

xshell链接zookeeper，进入bin目录，执行./zkCli.sh

ls -w / 查看根目录节点名称

create /canal-master-order-dev "canal-master-order-dev" 创建目录

配置集群地址： 10.1.8.100:2181/canal-master-order-dev

5、修改默认配置（通用的canal.properties）

载入模板配置，和默认的canal.properties一致

主要修改以下配置：

canal.zkServers zk的ip:2181 如果有路径，也要加上路径

配置zookeeper集群地址：canal.instance.global.spring.xml 改为classpath:spring/default-instance.xml

配置canalserver类型：canal.serverMode = rocketMQ # tcp, kafka, rocketMQ, rabbitMQ

配置instance名称：canal.destinations

配置mq密码：canal.aliyun.accessKey（如果没有账号密码，可以不填）

配置mq账号：canal.aliyun.secretKey

配置mq信息

rocketmq.producer.group = sync_passenger_order

rocketmq.enable.message.trace = false

rocketmq.customized.trace.topic =

rocketmq.namespace = 选填比如默认namespace是default就可以填这个

rocketmq.namesrv.addr = mq地址，必须填

rocketmq.retry.times.when.send.failed = 0

rocketmq.vip.channel.enabled = false

rocketmq.tag = tag_sync_passenger_order

同时注释掉

#canal.port

#canal.metrics.pull.port

#canal.admin.port = 11110

#canal.admin.user = admin

#canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441

#################################################
######### 		common argument		#############
#################################################
# tcp bind ip
canal.ip =
# register ip to zookeeper
canal.register.ip =
#canal.port = 22111
#canal.metrics.pull.port = 22112
# canal instance user/passwd
# canal.user = canal
# canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458# canal admin config
#canal.admin.manager = 127.0.0.1:8089
#canal.admin.port = 22110
#canal.admin.user = admin
#canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
#canal.admin.register.auto = true
#canal.admin.register.cluster =
#canal.admin.register.name =canal.zkServers = zk的ip:2181 如果有路径，也要加上路径
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, rocketMQ, rabbitMQ
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 = true
canal.instance.filter.query.dml = true
canal.instance.filter.query.ddl = true
canal.instance.filter.table.error = false
canal.instance.filter.rows = false
canal.instance.filter.transaction.entry = false
canal.instance.filter.dml.insert = false
canal.instance.filter.dml.update = false
canal.instance.filter.dml.delete = true# 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 = false
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#################################################
######### 		destinations		#############
#################################################
canal.destinations = sync_passenger_order,sync_passenger_order_ext,sync_passenger_order_after
# 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
# set this value to 'true' means that when binlog pos not found, skip to latest.
# WARN: pls keep 'false' in production env, or if you know what you want.
canal.auto.reset.latest.pos.mode = falsecanal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xmlcanal.instance.global.mode = spring
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 Properties      #############
##################################################
# aliyun ak/sk , support rds/mq
canal.aliyun.accessKey = mq的账号
canal.aliyun.secretKey = mq的密码
canal.aliyun.uid=canal.mq.flatMessage = true
canal.mq.canalBatchSize = 50
canal.mq.canalGetTimeout = 100
# Set this value to "cloud", if you want open message trace feature in aliyun.
canal.mq.accessChannel = localcanal.mq.database.hash = true
canal.mq.send.thread.size = 30
canal.mq.build.thread.size = 8##################################################
######### 		     Kafka 		     #############
##################################################
kafka.bootstrap.servers = 127.0.0.1:9092
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
kafka.linger.ms = 1
kafka.max.request.size = 1048576
kafka.buffer.memory = 33554432
kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0kafka.kerberos.enable = false
kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf"
kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf"##################################################
######### 		    RocketMQ	     #############
##################################################
rocketmq.producer.group = sync_passenger_order
rocketmq.enable.message.trace = false
rocketmq.customized.trace.topic =
rocketmq.namespace = mq的namespace
rocketmq.namesrv.addr = mq的地址
rocketmq.retry.times.when.send.failed = 0
rocketmq.vip.channel.enabled = false
rocketmq.tag = tag_sync_passenger_order##################################################
######### 		    RabbitMQ	     #############
##################################################
rabbitmq.host =
rabbitmq.virtual.host =
rabbitmq.exchange =
rabbitmq.username =
rabbitmq.password =
rabbitmq.deliveryMode =

6、添加server实例（即canal-deployer）

配置项：

所属集群，可以选择为单机或者集群。一般单机Server的模式主要用于一次性的任务或者测试任务

Server名称，唯一即可，方便自己记忆

Server Ip，机器ip

admin端口，canal 1.1.4版本新增的能力，会在canal-server上提供远程管理操作，默认值11110

tcp端口，canal提供netty数据订阅服务的端口

metric端口， promethues的exporter监控数据端口 (未来会对接监控)

多台Server关联同一个集群即可形成主备HA架构

7、创建Instance

先填写实例名

选择刚刚创建的集群

载入模板配置

主要修改以下配置：

「canal.instance.master.address」配置要同步的数据库地址

「canal.instance.dbUsername」数据库用户名（需同步权限）

「canal.instance.dbPassword」数据库密码

「canal.instance.filter.regex」 mysql 数据解析关注的表，Perl正则表达式.多个正则之间以逗号(,)分隔，转义符需要双斜杠()

「canal.mq.topic」mq的topic

#################################################
## mysql serverId , v1.0.26+ will autoGen
# canal.instance.mysql.slaveId=# enable gtid use true/false
canal.instance.gtidon=false# position info
canal.instance.master.address=1数据库ip:63306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=# rds oss binlog
canal.instance.rds.accesskey=
canal.instance.rds.secretkey=
canal.instance.rds.instanceId=# table meta tsdb info
canal.instance.tsdb.enable=true
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
#canal.instance.tsdb.dbUsername=canal
#canal.instance.tsdb.dbPassword=canal#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=# username/password
canal.instance.dbUsername=数据库账号
canal.instance.dbPassword=数据库密码
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==# table regex
canal.instance.filter.regex=gac_order.order_info
# table black regex
canal.instance.filter.black.regex=mysql\\.slave_.*
# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch# mq config
canal.mq.topic=topic_sync_passenger_order
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
canal.mq.partition=0
# hash partition config
#canal.mq.partitionsNum=3
#canal.mq.partitionHash=test.table:id^name,.*\\..*
#canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6
#################################################

部署canal-deployer

作用：

伪装成 MySQL 的从库，同步主库的binlog日志。

解析并结构化 binary log 对象。

1、修改配置canal_local.properties

主要修改

canal.admin.manager

canal.admin.port

canal.port

canal.metrics.pull.port

如果admin的账号密码改了也要修改，尽量用默认的防止出错

# 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 = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster =
canal.admin.register.name = #server暴露端口 每个server的port都不一样
canal.port = 25111
canal.metrics.pull.port = 25112

2、启动server

进入server的bin目录

使用sudo chmod 777 ./startup.sh sudo chmod 777 ./stop.sh 命令添加权限

./startup.sh local

监控

canal 默认已通过 11112 端口暴露同步相关的 metrics 信息，只需通过集成 prometheus 与 grafana 即可实现实时监控同步情况，效果图如下：

canal-admin的部署与使用，及相关监控_canal admin-CSDN博客

Canal的配置管理、监控与性能测试 - 墨天轮

指标	简述
Basic	Canal instance 基本信息。
Network bandwith	网络带宽。包含inbound(canal server读取binlog的网络带宽)和outbound(canal server返回给canal client的网络带宽)。
Delay	Canal server与master延时；store 的put, get, ack操作对应的延时。
Blocking	sink线程blocking占比；dump线程blocking占比(仅parallel mode)。
TPS(events)	Canal instance消费所有binlog事件的TPS，以MySQL binlog events为单位计算。
TPS(transaction)	Canal instance 处理binlog的TPS，以MySQL transaction为单位计算。
TPS(tableRows)	分别对应store的put, get, ack操作针对数据表变更行的TPS。
Client requests	Canal client请求server的请求数统计，结果按请求类型分类(比如get/ack/sub/rollback等)。
Client QPS	client发送请求的QPS，按GET与CLIENTACK分类统计。
Empty packets	Canal client请求server返回空结果的统计。
Response time	Canal client请求server的响应时间统计。
Store remain events	Canal instance ringbuffer中堆积的events数量。
Store remain mem	Canal instance ringbuffer中堆积的events内存使用量。

安装prometheus

https://prometheus.io

 - job_name: 'canal'static_configs:- targets: ['localhost:11112'] //端口配置即为canal.properties中的canal.metrics.pull.port

安装grafana

Grafana: The open and composable observability platform | Grafana Labs

十二、Linxu下安装grafana（性能） - 陈橙橙橙子 - 博客园

添加数据源 Grafana安装和配置Prometheus数据源教程_grafana配置prometheus-CSDN博客

import模板

下载文件
wget https://dl.grafana.com/enterprise/release/grafana-enterprise-9.0.7.linux-amd64.tar.gz然后上传到服务器进行解压tar -zxvf grafana-enterprise-9.0.7.linux-amd64.tar.gz三、启动grafana
1、首先进入安装目录的conf文件夹，如果有需要修改的配置，可修改defaults.ini配置文件2、然后进入bin目录，启动grafana
nohup ./grafana-server &
nohup ./grafana-server > /dev/null 2&>1 &四、配置grafana
访问http://localhost:3000初始用户名和密码是admin，登录后会强制修改密码，然后在datasouce里配置Prometheus数据源五、导入grafana模板
1、点击Dashboards--Browse,点击import按钮 2、然后点击upload json file，选择之前导出的json文件上传

配置告警信息

(二) prometheus报警-----自定义 / alertmanager监控，报警设置_prometheus 自定义监控及告警配置-CSDN博客

彻底搞懂监控系统，使用Prometheus和Grafana 如何实现运维告警？-阿里云开发者社区