3.canal解析binlog送入kafka

Posted 2022-09-20 PacosonSWJTU

【README】

1.本文介绍了 canal把日志解析后送入到kafka的配置；

2.canal抽取增量binlog日志的架构图，参见
2.canal服务器配置及java客户端_PacosonSWJTU的博客-CSDN博客mysql master节点可以开启biglog日志记录功能，开启后每次向mysql服务端发送写操作命令，会把命令记录在一种特殊的文件中，这个特殊的文件称为biglog日志。若服务器异常退出，借用binlog可以恢复数据！二进制日志索引文件（文件名后缀为.index）用于记录所有的二进制文件；二进制日志文件（文件名后缀为.00000*）记录数据库所有的 DDL 和 DML(除了数据查询语句)语句事件；4）查看日志文件及日志索引文件// 内容如下；https://blog.csdn.net/PacosonSWJTU/article/details/1269033803.原理如下：

• 即 canal服务器做了kafka的配置，即canal服务器会启动一个线程，这个线程创建一个kafka生产者；
• canal在做增量binlog解析时，把解析结果送入 kafka（生产者）即可； 

---

【2】canal解析结果送入kafka配置

【2.1】修改canal配置

1）服务器配置修改如下。

// canal.properties 修改如下。
// 服务器模式为kafka
canal.serverMode = kafka  
// kafka集群节点列表
kafka.bootstrap.servers = centos201:9092,centos202:9092,centos203:9092 

2）example 实例的 instance.properties 修改如下。

// instance.properties 修改如下。
// slave机器编号，与mysql的master机器编号区别开 
canal.instance.mysql.slaveId=201

// mysql服务器master节点的ip及端口
canal.instance.master.address=192.168.163.201:3306

// 连接到master节点的用户名与密码
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal

// kafka topic
canal.mq.topic=canal_topic

3）修改完成；

重新启动 canal 服务器。

bin/startup.sh

---

 【3】kafka配置 

1）创建topic， canal_topic

kafka-topics.sh --bootstrap-server centos201:9092 --create --topic canal_topic --partitions 3 --replication-factor 2

2）命令行开启 canal_topic 的消费者：

[root@centos201 conf]# kafka-console-consumer.sh --topic canal_topic --bootstrap-server centos201:9092 

3）插入数据，并查看kafka消费者消费效果：

3.1）插入数据：

INSERT INTO trcanal.user_inf_tbl (id, name, sex) VALUES 
('20220917_0028', 'zhangsan0027', 'male27')
;

3.2）kafka消费者打印消息：

[root@centos201 conf]# kafka-console-consumer.sh --topic canal_topic --bootstrap-server centos201:9092         
"data":["id":"20220917_0028","name":"zhangsan0027","sex":"male27"],"database":"trcanal","es":1663408824000,"id":3,"isDdl":false,"mysqlType":"id":"varchar(20)","name":"varchar(255)","sex":"varchar(255)","old":null,"pkNames":["id"],"sql":"","sqlType":"id":12,"name":12,"sex":12,"table":"user_inf_tbl","ts":1663408825413,"type":"INSERT"
"data":["id":"20220917_0028","name":"zhangsan0027","sex":"male27"],"database":"trcanal","es":1663408824000,"id":3,"isDdl":false,"mysqlType":"id":"varchar(20)","name":"varchar(255)","sex":"varchar(255)","old":null,"pkNames":["id"],"sql":"","sqlType":"id":12,"name":12,"sex":12,"table":"user_inf_tbl","ts":1663408825452,"type":"INSERT"

canal解析增量binlog日志后，成功送入kafka；

4）数据结构：

---

 【补充】

显然 canal 解析增量binlog日志，在重放日志后，可能产生重复数据（这是我们需要注意的）；