搭建环境
操作系统： CentOS release 6.5 (Final)
MySQL版本： 8.0
JDK版本：11
canal版本： 1.1.4
zookeeper 3.6.2
kafka
mysql 配置
进入MySQL 查看是否开启binlog
注意一点是MySQL8.x默认开启binlog
SHOW VARIABLES LIKE '%bin%';
执行下面代码 创建新的用户名 canal 密码 QWqw12!@ 赋予
REPLICATION SLAVE和 REPLICATION CLIENT权限
CREATE USER canal IDENTIFIED BY 'QWqw12!@';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;
ALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'QWqw12!@';
#查看mysql binlog 是否开启 ON 开启 OFF 未开启
show variables like 'log_bin';
#查看binlog日志文件列表
show binary logs;
#查看binglog 模式 是否为ROW
show variables like 'binlog_format'
#查看当前正在写入的binlog文件：
show master status
安装zookeeper
Canal和Kafka集群都依赖于Zookeeper做服务协调，为了方便管理，一般会独立部署Zookeeper服务或者Zookeeper集群
cd usr/local/ 进入要安装软件目录
mkdir zk
cd zk
mkdir data
cd ../
下载安装zk
wget http://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.6.2/apache-zookeeper-3.6.2-bin.tar.gz
下载前打开 http://mirror.bit.edu.cn/apache/zookeeper 查看版本号多少
避免404 错误
解压zk
tar -zxvf apache-zookeeper-3.6.2-bin.tar.gz
cd apache-zookeeper-3.6.2-bin/conf
复值zoo_sample.cfg 到zoo.cfg 并打开zoo.cfk
cp zoo_sample.cfg zoo.cfg && vim zoo.cfg
修改dataDir 为刚才建的data文件夹
然后启动Zookeeper：
进入zk bin 目录
./bin/zkServer.sh start
tail -100f /usr/local/zk/apache-zookeeper-3.6.2-bin/logs/zookeeper-root-server-asp.out 查看打印日志
tail -fn 10 test.log -f 循环查看 -fn 循环查看实时10行数据
安装kafka
cd usr/local/ 进入要安装软件目录
在目录下创建kafka 文件夹以及在kafka 下创建 data
mkdir kafka
cd kafka
mkdir data
wget [http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.6.0/kafka_2.13-2.6.0.tgz](http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.6.0/kafka_2.13-2.6.0.tgz)
tar -zxvf kafka_2.13-2.6.0.tgz
由于解压后 kafka_2.13-2.4.0/config/server.properties配置中对应的zookeeper.connect=localhost:2181已经符合需要，不必修改，
需要修改日志文件的目录log.dirs为/data/kafka/data。然后启动Kafka`服务：
修改文件下log.dirs路径为刚刚创建的
修改server.properties 配置文件下的
listeners为
listeners=PLAINTEXT://localhost:9092
data 目录
注意目录在kafka_2.13-2.6.0 下
执行 ./bin/kafka-server-start.sh /usr/local/kafka/kafka_2.13-2.6.0/config/server.properties
这样启动一旦退出控制台就会结束Kafka进程，可以添加-daemon参数用于控制Kafka进程后台不挂断运行
./bin/kafka-server-start.sh -daemon /usr/local/kafka/kafka_2.13-2.6.0/config/server.properties
canal 安装
cd usr/local/ 进入要安装软件目录
mkdir canal
cd canal
wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gz
建议复制链接到浏览器手动下载，在上传服务器，翻墙不翻墙都特慢
- bin # 运维脚本
- conf # 配置文件
canal_local.properties # canal本地配置，一般不需要动
canal.properties # canal服务配置
logback.xml # logback日志配置
metrics # 度量统计配置
spring # spring-实例配置，主要和binlog位置计算、一些策略配置相关，可以在canal.properties选用其中的任意一个配置文件
example # 实例配置文件夹，一般认为单个数据库对应一个独立的实例配置文件夹
instance.properties # 实例配置，一般指单个数据库的配置
- lib # 服务依赖包
- logs # 日志文件输出目录
canal 配置
在开发和测试环境建议把logback.xml的日志级别修改为DEBUG方便定位问题。这里需要关注canal.properties和instance.properties两个配置文件。canal.properties文件中，需要修改：
- 去掉
canal.instance.parser.parallelThreadSize = 16这个配置项的注释，也就是启用此配置项，和实例解析器的线程数相关，不配置会表现为阻塞或者不进行解析。 -
canal.serverMode配置项指定为kafka，可选值有tcp、kafka和rocketmq（master分支或者最新的的v1.1.5-alpha-1版本，可以选用rabbitmq），默认是kafka。 -
canal.mq.servers配置需要指定为Kafka服务或者集群Broker的地址，这里配置为127.0.0.1:9092。
instance.properties一般指一个数据库实例的配置，Canal架构支持一个Canal服务实例，处理多个数据库实例的binlog异步解析。instance.properties需要修改的配置项主要包括：
canal.instance.mysql.slaveId需要配置一个和Master节点的服务ID完全不同的值，这里笔者配置为654321。-
配置数据源实例，包括地址、用户、密码和目标数据库：
-
canal.instance.master.address，这里指定为127.0.0.1:3306。 -
canal.instance.dbUsername，这里指定为canal。 -
canal.instance.dbPassword，这里指定为QWqw12!@。 - 新增
canal.instance.defaultDatabaseName，这里指定为test（需要在MySQL中建立一个test数据库，见前面的流程）。
Kafka 相关配置，这里暂时使用静态 topic和单个partition
-
canal.mq.topic，这里指定为test，也就是解析完的binlog结构化数据会发送到Kafka的命名为test的topic中。canal.mq.partition，这里指定为0。
#进入bin目录下运行
./startup.sh
# 查看服务日志
tail -100f ../logs/canal/canal
# 查看实例日志
-- 一般情况下，关注实例日志即可
tail -100f ../logs/example/example.log
canal 启动报错的话可以看看这个
https://www.jianshu.com/p/a0788a8f3832
对需要同步的数据库进行crud操作后
进入kafka bin 目录 运行
./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --from-beginning --topic asp_be
对数据进行增加和修改后
kafka 就会有日志输出
