深入浅出系列之 -- MySQL的数据增量同步到kafka

￡神魔★判官ぃ 2021-10-19 15:34 588阅读 0赞

## **一、架构介绍** ##

生产中由于历史原因 web 后端，mysql 集群，kafka 集群（或者其它消息队列）会存在以下三种结构。

### **1 .数据先入mysql 集群，再入kafka** ###

数据入的MySQL 集群是不可更改的，如何再高效的将数据写入kafka呢？

> A），在表中存在自增ID 的字段，然后根据ID ，定期扫描表，然后将数据入kafka 。
> 
> B），有时间字段的，可以按照时间字段定期扫描入kafka 集群。
> 
> C），直接解析binlog 日志，然后解析后的数据写入kafka 。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70][]

### **2 .web 后端同时将数据写入kafka 和mysql 集群** ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70 1][]

### **3 .web 后端将数据先入kafka ，再入mysql 集群** ###

这个方式，有很多优点，比如可以用kafka解耦，然后将数据按照离线存储和计算，实时计算两个模块构建很好的大数据架构。抗高峰，便于扩展等等。

![20190813144530287.png][]

## **二、实现步骤** ##

### **1 .mysql 安装准备** ###

> A），假如你单机测试请配置好SERVER\_ID
> 
> B），开启binlog ，只需配置log-bin

\[root@localhost ~\]\# cat /etc/my.cnf

\[mysqld\]

**server\_id=1**

datadir=/var/lib/mysql

socket=/var/lib/mysql/mysql.sock

user=mysql

\# Disabling symbolic-links is recommended to prevent assorted security risks

symbolic-links=0

**log-bin=/var/lib/mysql/mysql-binlog**

\[mysqld\_safe\]

log-error=/var/log/mysqld.log

pid-file=/var/run/mysqld/mysqld.pid

创建测试库和表

create database school character set utf8 collate utf8\_general\_ci;

create table student(

name varchar(20) not null comment '姓名',

sid int(10) not null primary key comment '学员',

majora varchar(50) not null default '' comment '专业',

tel varchar(11) not null unique key comment '手机号',

birthday date not null comment '出生日期'

);

### **2 .binlog 日志解析** ###

两种方式：

> 一是扫面二进制日志文件
> 
> 二是通过复制同步的方式

暂实现了第二种方式，样例代码如下：

MysqlBinlogParse mysqlBinlogParse = new MysqlBinlogParse(args[0],Integer.valueOf(args[1]),args[2],args[3]){
      @Override
      public void processDelete(String queryType, String database, String sql) {
        try {
          String jsonString = SqlParse.parseDeleteSql(sql);
          JSONObject jsonObject = JSONObject.fromObject(jsonString);
          jsonObject.accumulate("database", database);
          jsonObject.accumulate("queryType", queryType);
          System.out.println(sql);
          System.out.println(" ");
          System.out.println(" ");
          System.out.println(jsonObject.toString());
        } catch (Exception e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
        }
      }
    
      @Override
      public void processInsert(String queryType, String database, String sql) {
        try {
          String jsonString = SqlParse.parseInsertSql(sql);
          JSONObject jsonObject = JSONObject.fromObject(jsonString);
          jsonObject.accumulate("database", database);
          jsonObject.accumulate("queryType", queryType);
          System.out.println(sql);
          System.out.println(" ");
          System.out.println(" ");
          System.out.println(jsonObject.toString());
        } catch (Exception e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
        }
      }
    
      @Override
      public void processUpdate(String queryType, String database, String sql) {
        String jsonString;
        try {
          jsonString = SqlParse.parseUpdateSql(sql);
          JSONObject jsonObject = JSONObject.fromObject(jsonString);
          jsonObject.accumulate("database", database);
          jsonObject.accumulate("queryType", queryType);
          System.out.println(sql);
          System.out.println(" ");
          System.out.println(" ");
          System.out.println(jsonObject.toString());
        } catch (Exception e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
        }
      }
    
    };
    mysqlBinlogParse.setServerId(3);
    mysqlBinlogParse.start();

### **3.sql语法解析** ###

从原始的mysql 的binlog event中，我们能解析到的信息，主要的也就是mysql的database，query类型(INSERT,DELETE,UPDATE)，具体执行的sql。我这里封装了三个重要的方法。只暴露了这三个接口，那么我们要明白的事情是，我们入kafka，然后流式处理的时候希望的到的是跟插入mysql后一样格式的数据。这个时候我们就要自己做sql的解析，将query的sql解析成字段形式的数据，供流式处理。最终将解析后的数据封装成了json，然后我们自己写kafka producer将消息发送到kafka，后端就可以处理了。

## **三、总结** ##

最后建议web后端数据最好先入消息队列，如kafka，然后分离线和实时将数据进行解耦分流，用于实时处理和离线处理。 消息队列的订阅者可以根据需要随时扩展，可以很好的扩展数据的使用者。消息队列的横向扩展，增加吞吐量，做起来还是很简单的。这个用传统数据库，分库分表还是很麻烦的。 由于消息队列的存在，也可以帮助我们抗高峰，避免高峰时期后端处理压力过大导致整个业务处理宕机。

\-----------------------------------------------------------------------------------------------------------------------------------------------------------------------

考虑一千次，不如去做一次；犹豫一万次，不如实践一次；华丽的跌倒，胜过无谓的彷徨，将来的你，一定会感谢现在奋斗的你。欢迎大家加入大数据交流群：725967421     一起交流，一起进步！！

\------------------------------------------------------------------------------------------------------------------------------------------------------------------------

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70]: /images/20211019/9a6498fc75bc4067b65de87c73e9d529.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70 1]: /images/20211019/1c6c32bfdab74d809f6312b8a39b5a5f.png
[20190813144530287.png]: /images/20211019/191a9c72c3584b4481be21da598d4b10.png