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%';

canal+mysql+zookeeper+kafka|canal+mysql+zookeeper+kafka 实时同步
文章图片
image.png
执行下面代码 创建新的用户名 canal 密码 QWqw12!@ 赋予 REPLICATION SLAVEREPLICATION 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 CanalKafka集群都依赖于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 错误

canal+mysql+zookeeper+kafka|canal+mysql+zookeeper+kafka 实时同步
文章图片
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文件夹

canal+mysql+zookeeper+kafka|canal+mysql+zookeeper+kafka 实时同步
文章图片
image.png 然后启动Zookeeper
进入zk bin 目录
./bin/zkServer.sh start
canal+mysql+zookeeper+kafka|canal+mysql+zookeeper+kafka 实时同步
文章图片
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`服务:
canal+mysql+zookeeper+kafka|canal+mysql+zookeeper+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
canal+mysql+zookeeper+kafka|canal+mysql+zookeeper+kafka 实时同步
文章图片
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.propertiesinstance.properties两个配置文件。canal.properties文件中,需要修改:
  • 去掉canal.instance.parser.parallelThreadSize = 16这个配置项的注释,也就是启用此配置项,和实例解析器的线程数相关,不配置会表现为阻塞或者不进行解析。
  • canal.serverMode配置项指定为kafka,可选值有tcpkafkarocketmqmaster分支或者最新的的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的命名为testtopic中。
  • 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 就会有日志输出

    推荐阅读