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.xml2.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
