Flink CDC (Mysql为例)

背景

业务中经常出现一些千万乃至亿级别的大表，此时可能考虑分库分表（Sharding-JDBC、MyCat等方案），也常同步数据进入ES中；同步数据这一业务场景中，Flink CDC是一个很不错的解决方案。

方案

如mysql、postgresql、sqlserver等，flink cdc通过读取binlog日志（注意：请先开启binlog日志），进行数据同步，实时性较好。

对数据的解析和消费进行了二次封装，使用者只需增加简单的配置，实现FlinkConsumerListener接口，关注编写业务代码即可。

代码

show coding

flink: flink cdc 暂时支持mysql

测试demo

创建一个springboot项目

依赖引入（引入上述工程打包后的依赖）

        
            com.kwin
            flink
            0.0.1-SNAPSHOT
        
1
2
3
4
5

配置文件

flink:
pipeline-name: flinkCDCTest
mysqlDataSource:
- port: 3306
hostname: 127.0.0.1
databaseList:
- flinktest
tableList:
- flinktest.student
username: root
password: 123456

如上，针对flinktest数据库的student表进行binlog监听。

flinktest.student的消费者

student实体

import lombok.Data;

/**
 * @author kwin
 * @Date 2022/7/25 18:27
 **/
@Data
public class Student {
    private Long id;

    private String name;

    private Integer age;

    private Integer maxInx;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

消费者

import com.kwin.demo.server.module.flink.test.entity.Student;
import com.kwin.flink.sink.FlinkConsumerListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * @author kwin
 * @Date 2022/7/25 18:29
 **/
@Slf4j
@Component
public class StudentConsumerListener implements FlinkConsumerListener {
    @Override
    public String getDBName() {
        return "flinktest";
    }

    @Override
    public String getTable() {
        return "student";
    }

    @Override
    public void insert(Student data) {
        System.out.println("insert: " + data);
    }

    @Override
    public void update(Student srcData, Student destData) {
        System.out.println("update: 
src:" + srcData + "
dest:" + destData);
    }

    @Override
    public void delete(Student data) {
        System.out.println("delete:"+data);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

启动项目

flinktest.student修改数据时：

flinktest.student插入数据时：

flinktest.student删除数据时：

如上，使用者只需实现FlinkConsumerListener接口，即可对指定表的数据进行消费和业务逻辑操作。

最后

深知大多数初中级Java工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则近万的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《Java开发全套学习资料》送给大家，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。

小编已加密：aHR0cHM6Ly9kb2NzLnFxLmNvbS9kb2MvRFVrVm9aSGxQZUVsTlkwUnc==出于安全原因，我们把网站通过base64编码了，大家可以通过base64解码把网址获取下来。