准备

对于自建 MySQL , 需要先开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，my.cnf 中配置如下

[mysqld]

log-bin=mysql-bin # 开启 binlog

binlog-format=ROW # 选择 ROW 模式

server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

注意：针对阿里云 RDS for MySQL , 默认打开了 binlog , 并且账号默认具有 binlog dump 权限 , 不需要任何权限或者 binlog 设置,可以直接跳过这一步

授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant

CREATE USER canal IDENTIFIED BY 'canal';

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';

-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;

FLUSH PRIVILEGES;

安装canal.server

1 下载压缩包

到官网地址(release)下载最新压缩包,请下载 canal.deployer-latest.tar.gz

2 将canal.deployer 复制到固定目录并解压

mkdir -p /usr/local/canal

cp canal.deployer-1.1.1.tar.gz /usr/local/canal

tar -zxvf canal.deployer-1.1.1.tar.gz

3 配置修改参数

a. 修改instance 配置文件，修改单个数据源的配置 vi conf/example/instance.properties

# 按需修改成自己的数据库信息

#################################################

...

#需要同步的mysql地址，端口

canal.instance.master.address=192.168.1.20:3306

# username/password,数据库的用户名和密码

...

canal.instance.dbUsername = root

canal.instance.dbPassword = 123456

#需要同步的具体表，也可以使用正则表达式监听多张表或者库

canal.instance.filter.regex=realtime.tpp_vehicle_through_points_history

...

# mq config

# topic的名称

canal.mq.topic=topic_start

# 针对库名或者表名发送动态topic

#canal.mq.dynamicTopic=mytest,.*,mytest.user,mytest\\..*,.*\\..*

canal.mq.partition=0

# hash partition config

#canal.mq.partitionsNum=3

# 库名.表名: 唯一主键，多个表之间用逗号分隔

#canal.mq.partitionHash=mytest.person:id,mytest.role:id

#################################################

b. 修改canal 配置文件，修改 Canal 全局设置  vi ./conf/canal.properties

# ...

# 可选项: tcp(默认), kafka, RocketMQ

# 更改模式，直接把数据扔进 Kafka

canal.serverMode = kafka

# ...

# kafka/rocketmq 集群配置: 192.168.1.117:9092,192.168.1.118:9092,192.168.1.119:9092

## Kafka 的地址

canal.mq.servers = 10.255.xx.xx:9092,10.255.xx.xx:9092,10.255.xx.xx:9092

canal.mq.retries = 0

# flagMessage模式下可以调大该值, 但不要超过MQ消息体大小上限

canal.mq.batchSize = 16384

canal.mq.maxRequestSize = 1048576

# flatMessage模式下请将该值改大, 建议50-200

canal.mq.lingerMs = 1

canal.mq.bufferMemory = 33554432

# Canal的batch size, 默认50K, 由于kafka最大消息体限制请勿超过1M(900K以下)

canal.mq.canalBatchSize = 50

# Canal get数据的超时时间, 单位: 毫秒, 空为不限超时

canal.mq.canalGetTimeout = 100

# 是否为flat json格式对象

#canal.mq.flatMessage = false

canal.mq.flatMessage = true# 使用文本格式(JSON)进行传输，否则 Kafka 里扔进去的是二进制数据，虽然不影响，但是看起来不方便

canal.mq.compressionType = none

canal.mq.acks = all

# kafka消息投递是否使用事务

canal.mq.transaction = false

开启 Canal bin/startup.sh

查看日志是否有异常 vim logs/canal/canal.log vim logs/test/test.log

测试，连上数据库尝试执行更改 SQL

启动kafka的消费监控：kafka-console-consumer --bootstrap-server app02:9092 --from-beginning --topic topic_start

看得到同步的表名称和同步的数据类型：UPDATE，INSERT，DELETE

如果出现一些意外导致数据错误，数据不能采集，删除/opt/zp/canal/conf/example/h2.mv.db和meta.dat。重启canal