一、MySQL配置

1.1 安装 MySQL

略

1.2 修改 my.ini 配置文件

1[mysqld]
2log-bin=mysql-bin # 开启binlog
3binlog-format=ROW # 选择row模式
4server_id=1 # serverid唯一

1.3 授权

1GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal';
2FLUSH PRIVILEGES;

1.4 重启数据库

1service mysqld restart

二、部署 canal

2.1 安装 java

1yum -y install java

验证安装

1java -version
2openjdk version "1.8.0_312"
3OpenJDK Runtime Environment (build 1.8.0_312-b07)
4OpenJDK 64-Bit Server VM (build 25.312-b07, mixed mode)

2.2 下载 canal

1wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz
2mkdir -p /opt/canal/deploy
3tar xf canal.deployer-1.1.5.tar.gz -C /opt/canal/deploy

2.3 配置 canal

 1cd /opt/canal/deploy
 2vim conf/example/instance.properties
 3#################################################
 4## mysql serverId , v1.0.26+ will autoGen server-id 自动生成
 5# canal.instance.mysql.slaveId=0
 6
 7# enable gtid use true/false 是否启用gtid
 8canal.instance.gtidon=false
 9
10# position info 位置信息
11canal.instance.master.address=127.0.0.1:3306
12canal.instance.master.journal.name=
13canal.instance.master.position=
14canal.instance.master.timestamp=
15canal.instance.master.gtid=
16
17# rds oss binlog rds oss信息
18canal.instance.rds.accesskey=
19canal.instance.rds.secretkey=
20canal.instance.rds.instanceId=
21
22# table meta tsdb info 时序数据库相关信息
23canal.instance.tsdb.enable=true # 启用时序数据库，默认使用h2数据库保存
24#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
25#canal.instance.tsdb.dbUsername=canal
26#canal.instance.tsdb.dbPassword=canal
27
28# canal 备机信息
29#canal.instance.standby.address =
30#canal.instance.standby.journal.name =
31#canal.instance.standby.position =
32#canal.instance.standby.timestamp =
33#canal.instance.standby.gtid=
34
35# username/password 账号密码
36canal.instance.dbUsername=canal
37canal.instance.dbPassword=canal # 修改用户密码
38canal.instance.connectionCharset = UTF-8
39canal.instance.defaultDatabaseName = test # 需要同步的数据库
40# enable druid Decrypt database password
41canal.instance.enableDruid=false
42#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==
43
44# table regex
45canal.instance.filter.regex=.*\\..*
46# table black regex
47canal.instance.filter.black.regex=mysql\\.slave_.*
48# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
49#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
50# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
51#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch
52
53# mq config
54canal.mq.topic=example
55# dynamic topic route by schema or table regex
56#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
57canal.mq.partition=0
58# hash partition config
59#canal.mq.partitionsNum=3
60#canal.mq.partitionHash=test.table:id^name,.*\\..*
61#canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6
62#################################################

2.3 运行

1./bin/startup.sh

运行之后会监听11110、11111、11112端口

检查

1netstat -tnlp | grep java

三、部署 canal-apapter

从库的配置通过 adapter 进行管理

3.1 下载 canal-adapter

1wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.adapter-1.1.5.tar.gz
2mkdir /opt/canal/adapter
3tar xf canal.adapter-1.1.5.tar.gz -C /opt/canal/adapter

3.2 配置 canal-adapter

 1cd /opt/canal/adapter
 2vim conf/application.yml
 3server:
 4  port: 8081
 5spring:
 6  jackson:
 7    date-format: yyyy-MM-dd HH:mm:ss
 8    time-zone: GMT+8
 9    default-property-inclusion: non_null
10
11canal.conf:
12  mode: tcp #tcp kafka rocketMQ rabbitMQ
13  flatMessage: true
14  zookeeperHosts:
15  syncBatchSize: 1000
16  retries: 0
17  timeout:
18  accessKey:
19  secretKey:
20  consumerProperties:
21    # canal tcp consumer
22    canal.tcp.server.host: 127.0.0.1:11111
23    canal.tcp.zookeeper.hosts:
24    canal.tcp.batch.size: 500
25    canal.tcp.username:
26    canal.tcp.password:
27...
28  canalAdapters:
29  - instance: example # canal instance Name or mq topic name
30    groups:
31    - groupId: g1
32      outerAdapters:
33      - name: logger
34      - name: rdb
35        key: mysql1
36        properties:
37          jdbc.driverClassName: com.mysql.jdbc.Driver
38          jdbc.url: jdbc:mysql://127.0.0.1:3307/mytest2?useUnicode=true
39          jdbc.username: root
40          jdbc.password: 'Admin12#$'

记得给用户授权，请确保上面的账号有写入权限

这里同步整个 test 数据库为例

 1vim conf/rdb/mytest_user.yml
 2dataSourceKey: defaultDS
 3destination: example
 4groupId: g1
 5outerAdapterKey: mysql1 # 需要跟上面保持一致
 6concurrent: true
 7dbMapping:
 8  mirrorDb: true
 9  # 配置数据库，需要提前创建
10  database: test
11...

3.3 运行 canal-adapter

1./bin/startup.sh

程序运行起来之后监听 8081 端口。

3.4 测试数据同步

1CREATE TABLE t1(id INT auto_increment PRIMARY KEY, name VARCHAR(32));
2INSERT INTO t1(name) VALUES("tom"),("jerry"),("jane");

观察数据同步情况，这边看数据已经实时同步过去。

四、部署 canal-admin

canal-admin 是一款 web 管理工具（可选）

4.1 下载 canal-admin

最好是跟 canal 版本保持一致

1wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.admin-1.1.5.tar.gz
2mkdir /opt/canal/admin
3tar xf canal.admin-1.1.5.tar.gz -C /opt/canal/admin

4.2 配置 canal-admin

 1cd /opt/canal/admin
 2vim conf/application.yml
 3server:
 4  port: 8089
 5spring:
 6  jackson:
 7    date-format: yyyy-MM-dd HH:mm:ss
 8    time-zone: GMT+8
 9
10spring.datasource:
11  address: 127.0.0.1:3306
12  database: canal_manager
13  username: canal
14  password: IeT6Cohc.
15  driver-class-name: com.mysql.jdbc.Driver
16  url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false
17  hikari:
18    maximum-pool-size: 30
19    minimum-idle: 1
20
21canal:
22  adminUser: admin
23  adminPasswd: admin

4.3 初始化 canal-admin

从上面的配置文件我们可以看到需要一个 canal_manager 的数据库，对应的 sql 文件在 conf 目录下

1source conf/canal_manager.sql
2GRANT ALL ON canal_manager.* TO canal@'%';
3FLUSH PRIVILEGES;

该 sql 会创建数据库和初始化库表结构并初始化一条记录到表中，用于登录

4.4 运行

1./bin/startup.sh

默认监听 8089 端口

4.5 访问

http://<server-ip>:8089

默认账号密码admin/123456

4.6 web 配置

Server 管理 -> 新建 Server

输入服务器IP信息和端口，添加本机的Server，添加完后看到状态为启动状态。

Instance 管理->新建 Instance

输入Instance 名称，选择刚新建的Server,点击载入模板，填写相关信息，最后点击保存。

五、canal 整合 kafka

5.1 部署 kafka

部署 Zookeeper 和 Kafka 略

5.2 配置 canal

修改 canal.properties 配置文件

 1# tcp, kafka, rocketMQ, rabbitMQ
 2canal.serverMode = kafka # 默认为tcp
 3...
 4##################################################
 5#########                    Kafka                   #############
 6##################################################
 7kafka.bootstrap.servers = 127.0.0.1:9092 # kafka服务器地址，多个用逗号分隔
 8kafka.acks = all
 9kafka.compression.type = none
10kafka.batch.size = 16384
11kafka.linger.ms = 1
12kafka.max.request.size = 1048576
13kafka.buffer.memory = 33554432
14kafka.max.in.flight.requests.per.connection = 1
15kafka.retries = 0
16
17kafka.kerberos.enable = false
18kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf"
19kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf"
20...

5.3 重启 canal 服务

1./bin/restart.sh

5.4 配置 adapter

 1vim conf/application.yml
 2server:
 3  port: 8081
 4spring:
 5  jackson:
 6    date-format: yyyy-MM-dd HH:mm:ss
 7    time-zone: GMT+8
 8    default-property-inclusion: non_null
 9
10canal.conf:
11  mode: kafka #tcp kafka rocketMQ rabbitMQ
12...
13  consumerProperties:
14    # canal tcp consumer
15    canal.tcp.server.host: 127.0.0.1:11111
16    canal.tcp.zookeeper.hosts:
17    canal.tcp.batch.size: 500
18    canal.tcp.username:
19    canal.tcp.password:
20    # kafka consumer
21    kafka.bootstrap.servers: 127.0.0.1:9092 # 修改kafka服务器地址信息，多个服务器用逗号分隔
22    kafka.enable.auto.commit: false
23    kafka.auto.commit.interval.ms: 1000
24    kafka.auto.offset.reset: latest
25    kafka.request.timeout.ms: 40000
26    kafka.session.timeout.ms: 30000
27    kafka.isolation.level: read_committed
28    kafka.max.poll.records: 1000
29...

5.5 重启 canal-adapter

1./bin/restart.sh

六、全量数据同步

6.1 修改 instance.properties 配置文件

 1vim conf/example/instance.properties
 2# position info
 3canal.instance.master.address=127.0.0.1:3306
 4# mysql binlog文件
 5canal.instance.master.journal.name=mysql-bin.0001
 6# binlog起始位置
 7canal.instance.master.position=0
 8# 获取binlog的起始时间戳
 9canal.instance.master.timestamp=1638005221
10canal.instance.master.gtid=

6.2 删除记录文件 meta.dat

因为这个文件记录 canal 已经消费的位置等信息。

6.3 重启服务

1./bin/restart.sh