当前位置： 首页 > news >正文

canal集群部署

news 来源：原创 2025/5/18 16:57:00
下载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:ss
    time-zone: GMT+8

spring.datasource:
  address: 127.0.0.1:63306
  database: canal_manager
  username: root
  password: 123456
  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

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 = false

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 = 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 = local

canal.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 = 0

kafka.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 如何实现运维告警？-阿里云开发者社区

相关文章：

  • 求出e的值（信息学奥赛一本通-1092）
  • ctfshow做题笔记—栈溢出—pwn69~pwn72
  • HybridCLR Generate All 报错UnityLinker.exe
  • Ubuntu-配置apt国内源
  • SpringBoot 实现接口数据脱敏
  • 【自学笔记】MoonBit语言基础知识点总览-持续更新
  • GOF设计模式在 Spring 框架中的核心应用分析
  • golang算法快慢指针
  • 19个判定学术写作内容有AI生成痕迹的例子
  • （Lauterbach调试器学习笔记）一、首次连接TriCore开发板调试
  • AutoGen学习笔记系列（十三）Advanced - Logging
  • 第75期 Doxygen是干嘛的，Windows版本，如何安装，学习
  • 函数题 01-复杂度3 二分查找【PAT】
  • 市盈率研究
  • Spring Boot集成EasyExcel
  • Python使用入门(二)
  • 侯捷 C++ 课程学习笔记：C++ 新标准11/14
  • 力扣练习之确定两个字符串是否接近
  • 【net2】mii，mdio，ncsi，bond，vlan，dns，ipv6
  • FPGA学习（三）——LED流水灯
  • 意德首脑会谈，梅洛尼警告欧盟绿色政策面临“工业荒漠化”
  • 新疆多地市民拍到不明飞行物：几秒内加速消失，气象部门回应
  • 美国考虑让移民上真人秀竞逐公民权，制片人称非现实版《饥饿游戏》
  • 专访｜《内沙》导演杨弋枢：挽留终将失去的美好
  • 以军称已开始在加沙的新一轮大规模攻势
  • 幼儿园教师拍打孩子额头，新疆库尔勒教育局：涉事教师已被辞退