canal+mysql+zookeeper+kafka|canal+mysql+zookeeper+kafka 实时同步 canal+mysql+zookeeper+kafka实时同步

搭建环境
操作系统： 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%';

文章图片
image.png
执行下面代码创建新的用户名 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!@';

canal+mysql+zookeeper+kafka|canal+mysql+zookeeper+kafka 实时同步

文章图片
image.png

#查看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 zkmkdir datacd ../ 下载安装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 错误

文章图片
image.png
解压zk

tar -zxvf apache-zookeeper-3.6.2-bin.tar.gzcd 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文件夹

文章图片
image.png 然后启动Zookeeper：
进入zk bin 目录
./bin/zkServer.sh start

文章图片
image.png

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 kafkacd kafkamkdir datawget [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`服务：

文章图片
image.png
【canal+mysql+zookeeper+kafka|canal+mysql+zookeeper+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

文章图片
image.png 这样启动一旦退出控制台就会结束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 就会有日志输出