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