1、组件版本

canal.deployer-1.1.7

kafka_2.13-2.6.0

jdk8

apache-zookeeper-3.6.0-bin

mysql5.7

2、解压canal.deployer-1.1.7

2.1、修改canal.properties配置文件

##改为kafka
canal.serverMode=kafka## 改为instance
canal.destinations = instance## 改为true，如果新增库或者改库相关的ddl语句，如果binlog都推送到一个topic，canal监听就不会有问题，如果区分表会导致新产生的binlog无法推送到kafka
canal.instance.filter.query.ddl = true## canal.instance.tsdb,新增库canal_tsdb
canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb?useUnicode=true&&characterEncoding=UTF-8&connectTimeout=3000&autoReconnect=true&socketTimeout=300000&useSSL=false
canal.instance.tsdb.dbUsername=root
canal.instance.tsdb.dbPassword=123456## canal默认使用h2，这里需要改为mysql
canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml

2.2、修改example/instance.properties配置文件

## 配置mysql master数据库信息
canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=canal_test
canal.instance.dbPassword=123456## 注释当前配置，就默认走按表推送数据至kafka，否则所有binlog都推送到同一个topic
#canal.mq.topic=canal_test
## 按表推送数据至kafka
canal.mq.dynamicTopic=mytest,.*,mytest.user,mytest\\..*,.*\\..*

2.3、给canal_test用户授权

检查binlog是否开启

show variables like '%log_bin%';

检查binlog日志格式是否为ROW

show variables like '%binlog_format%';

检查canal_test用户是否授予权限，如果没有，需要进行授权

show master status;

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

2.5、启动canal

终端执行命令：sh bin/startup.sh，是否启动成功，查看日志canal.deployer-1.1.7/logs/canal/canal.log、canal.deployer-1.1.7/logs/instance/instance.log