SpringBoot集成Flink CDC实现binlog监听

Flink CDC

CDC相关介绍

CDC是什么?

CDC是Change Data Capture(变更数据获取)的简称。焦颔首脑是，监测并捕获数据库的变更（包罗数据或数据表的插入、更新以及删除等），将这些变更按发生的顺序完备记载下来，写入到MQ以供其他服务进行订阅及消费
CDC分类

CDC主要分为基于查询和基于Binlog
基于查询基于Binlog开源产品Sqoop、DataXCanal、Maxwell、Debezium执行模式BatchStreaming是否可以捕获全部数据变化否是延迟性高延迟低延迟是否增长数据库压力是否 基于查询的都是Batch模式(即数据到达一定量后/一定时间才行会执行), 同时也由于这种模式, 那么延迟是必然高的, 而基于Streaming则是可以做到按条的粒度, 每条数据发生变化, 那么就会监听到
Flink CDC

Flink社区开发了flink-cdc-connectors组件，这是一个可以直接从MySQL、PostgreSQL 等数据库直接读取全量数据和增量变更数据的source组件。
目前也已开源，开源地址：https://github.com/ververica/flink-cdc-connectors
Java中集成Flink CDC

MySQL相关设置

执行初始化SQL数据

1. # 创建test数据库
2. create database test;
3. # 在test库中创建studnet, t1, t2, t3表, 插入数据
4. use test;
5. CREATE TABLE `student`  (
6.   `id` int(11) NOT NULL AUTO_INCREMENT,
7.   `name` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
8.   `age` int(11) NULL DEFAULT NULL,
9.   PRIMARY KEY (`id`) USING BTREE
10. ) ENGINE = InnoDB AUTO_INCREMENT = 4 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;
11. INSERT INTO `student` VALUES (1, 'joy', 18);
12. INSERT INTO `student` VALUES (2, 'tom', 123123);
14. CREATE TABLE t1(
15.     `id` VARCHAR(255) PRIMARY KEY,
16.     `name` VARCHAR(255)
17. );
18. CREATE TABLE t2(
19.     `id` VARCHAR(255) PRIMARY KEY,
20.     `name` VARCHAR(255)
21. );
22. CREATE TABLE t3(
23.     `id` VARCHAR(255) PRIMARY KEY,
24.     `name` VARCHAR(255)
25. );
27. # 创建test_route数据库
28. create database test_route;
29. # 在test_route库中创建t1, t2, t3表
30. use test_route;
31. CREATE TABLE t1(
32.     `id` VARCHAR(255) PRIMARY KEY,
33.     `name` VARCHAR(255)
34. );
35. CREATE TABLE t2(
36.     `id` VARCHAR(255) PRIMARY KEY,
37.     `name` VARCHAR(255)
38. );
39. CREATE TABLE t3(
40.     `id` VARCHAR(255) PRIMARY KEY,
41.     `name` VARCHAR(255)
42. );
45. # 在test_route数据库中的t1, t2, t3表插入数据
46. use test_route;
47. INSERT INTO t1 VALUES('1001','zhangsan');
48. INSERT INTO t1 VALUES('1002','lisi');
49. INSERT INTO t1 VALUES('1003','wangwu');
51. INSERT INTO t2 VALUES('1004','zhangsan');
52. INSERT INTO t2 VALUES('1005','lisi');
53. INSERT INTO t2 VALUES('1006','wangwu');
55. INSERT INTO t3 VALUES('1001','F');
56. INSERT INTO t3 VALUES('1002','F');
57. INSERT INTO t3 VALUES('1003','M');

复制代码

开启Binlog

通常来说默认安装MySQL的cnf都是存在/etc下的

1. sudo vim /etc/my.cnf

复制代码

1. # 添加如下配置信息,开启`test`以及`test_route`数据库的Binlog
2. # 数据库id
3. server-id = 1
4. # 时区, 如果不修改数据库时区, 那么Flink MySQL CDC无法启动
5. default-time-zone = '+8:00'
7. # 启动binlog，该参数的值会作为binlog的文件名
8. log-bin=mysql-bin
9. # binlog类型，maxwell要求为row类型
10. binlog_format=row
11. # 启用binlog的数据库，需根据实际情况作出修改
12. binlog-do-db=test
13. binlog-do-db=test_route

复制代码

修改数据库时区

永世修改, 那么就修改my.cnf配置(刚刚配置已经修改了, 记得重启即可)

1. default-time-zone = '+8:00'

复制代码

临时修改(重启会丢失)

1. # MySQL 8 执行这个
2. set persist time_zone='+8:00';
4. # MySQL 5.x版本执行这个
5. set time_zone='+8:00';

复制代码

重启MySQL

   留意了, 设置后必要重启MySQL!

1. service mysqld restart

复制代码

代码

相关依赖

1. <dependencies>
2.     <dependency>
3.         <groupId>org.springframework.boot</groupId>
4.         <artifactId>spring-boot-starter-web</artifactId>
5.     </dependency>
7.     <!-- Flink CDC依赖 start-->
8.     <!-- Flink核心依赖, 提供了Flink的核心API -->
9.     <dependency>
10.         <groupId>org.apache.flink</groupId>
11.         <artifactId>flink-java</artifactId>
12.         <version>${flink.version}</version>
13.     </dependency>
14.     <!--  Flink流处理Java API依赖 -->
15.     <dependency>
16.         <groupId>org.apache.flink</groupId>
17.         <artifactId>flink-streaming-java</artifactId>
18.         <version>${flink.version}</version>
19.     </dependency>
20.     <!-- Flink客户端工具依赖, 包含命令行界面和实用函数 -->
21.     <dependency>
22.         <groupId>org.apache.flink</groupId>
23.         <artifactId>flink-clients</artifactId>
24.         <version>${flink.version}</version>
25.     </dependency>
26.     <!-- Flink连接器基础包, 包含连接器公共功能 -->
27.     <dependency>
28.         <groupId>org.apache.flink</groupId>
29.         <artifactId>flink-connector-base</artifactId>
30.         <version>${flink.version}</version>
31.     </dependency>
32.     <!-- Flink Kafka连接器, 用于和Apache Kafka集成, 这里不需要集成, 所以注释掉, 代码可以使用其它的MQ代替 -->
33.     <!--<dependency>
34.             <groupId>org.apache.flink</groupId>
35.             <artifactId>flink-connector-kafka</artifactId>
36.             <version>3.2.0-1.19</version>
37.         </dependency>-->
38.     <!-- Flink Table Planner, 用于Table API和SQL的执行计划生成 -->
39.     <dependency>
40.         <groupId>org.apache.flink</groupId>
41.         <artifactId>flink-table-planner_2.12</artifactId>
42.         <version>${flink.version}</version>
43.     </dependency>
44.     <!-- Flink Table API桥接器, 连接DataStream API和Table API -->
45.     <dependency>
46.         <groupId>org.apache.flink</groupId>
47.         <artifactId>flink-table-api-java-bridge</artifactId>
48.         <version>${flink.version}</version>
49.     </dependency>
50.     <!-- Flink JSON格式化数据依赖 -->
51.     <dependency>
52.         <groupId>org.apache.flink</groupId>
53.         <artifactId>flink-json</artifactId>
54.         <version>${flink.version}</version>
55.     </dependency>
56.     <!-- 开启Web UI支持, 端口为8081, 默认为不开启-->
57.     <!--<dependency>
58.             <groupId>org.apache.flink</groupId>
59.             <artifactId>flink-runtime-web</artifactId>
60.             <version>1.19.1</version>
61.         </dependency>-->
63.     <!-- MySQL CDC依赖
64.         org.apache.flink的适用MySQL 8.0
66.          具体参照这篇博客 https://blog.csdn.net/kakaweb/article/details/129441408
67.          -->
68.     <dependency>
69.         <!--MySQL 8.0适用-->
70.         <!--<groupId>org.apache.flink</groupId>
71.             <artifactId>flink-sql-connector-mysql-cdc</artifactId>
72.             <version>3.1.0</version>-->
74.         <!-- MySQL 5.7适用 , 2.3.0, 3.0.1均可用-->
75.         <groupId>com.ververica</groupId>
76.         <artifactId>flink-sql-connector-mysql-cdc</artifactId>
77.         <!--<version>2.3.0</version>-->
78.         <version>3.0.1</version>
79.     </dependency>
81.     <!-- lombok -->
82.     <dependency>
83.         <groupId>org.projectlombok</groupId>
84.         <artifactId>lombok</artifactId>
85.     </dependency>
86.     <!-- gson工具类 -->
87.     <dependency>
88.         <groupId>com.google.code.gson</groupId>
89.         <artifactId>gson</artifactId>
90.         <version>2.11.0</version>
91.     </dependency>
93.     <!-- ognl表达式 -->
94.     <dependency>
95.         <groupId>ognl</groupId>
96.         <artifactId>ognl</artifactId>
97.         <version>3.1.1</version>
98.     </dependency>
100.     <!-- hutool工具类 -->
101.     <dependency>
102.         <groupId>cn.hutool</groupId>
103.         <artifactId>hutool-all</artifactId>
104.         <version>5.8.26</version>
105.     </dependency>
107.     <dependency>
108.         <groupId>com.alibaba</groupId>
109.         <artifactId>fastjson</artifactId>
110.         <version>2.0.31</version>
111.     </dependency>
112. </dependencies>
113. <name>cdc-test</name>
114. <description>cdc-test</description>
115. <properties>
116.     <java.version>11</java.version>
117.     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
118.     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
119.     <spring-boot.version>2.6.13</spring-boot.version>
120.     <flink.version>1.19.0</flink.version>
121. </properties>
122. <dependencyManagement>
123.     <dependencies>
124.         <dependency>
125.             <groupId>org.springframework.boot</groupId>
126.             <artifactId>spring-boot-dependencies</artifactId>
127.             <version>${spring-boot.version}</version>
128.             <type>pom</type>
129.             <scope>import</scope>
130.         </dependency>
131.     </dependencies>
132. </dependencyManagement>

复制代码

yaml

1. # Flink CDC相关配置
2. flink-cdc:
3.   mysql:
4.     hostname: 192.168.132.101
5.     port: 3306
6.     username: root
7.     password: 12345678
8.     databaseList: test
9.     tableList: test.student, test.t1
10.     includeSchemaChanges: false
11.     parallelism: 1
12.     enableCheckpointing: 5000

复制代码

FlinkCDCConfig

1. import lombok.Data;
2. import org.springframework.boot.context.properties.ConfigurationProperties;
3. import org.springframework.context.annotation.Configuration;
5. /**
6. * @author whiteBrocade
7. * @version 1.0
8. * @description: Flink CDC配置
9. */
10. @Data
11. @Configuration
12. @ConfigurationProperties("flink-cdc.mysql")
13. public class FlinkCDCConfig {
14.     /**
15.      * 数据库地址
16.      */
17.     private String hostname;
19.     /**
20.      * 数据库端口
21.      */
22.     private Integer port;
24.     /**
25.      * 数据库用户名
26.      */
27.     private String username;
29.     /**
30.      * 数据库密码
31.      */
32.     private String password;
34.     /**
35.      * 数据库名
36.      */
37.     private String[] databaseList;
39.     /**
40.      * 表名
41.      */
42.     private String[] tableList;
44.     /**
45.      * 是否包含schema变更
46.      */
47.     private Boolean includeSchemaChanges;
49.     /**
50.      * 并行度
51.      */
52.     private Integer parallelism;
54.     /**
55.      * 检查点间隔, 单位毫秒
56.      */
57.     private Integer enableCheckpointing;
58. }

复制代码

相关摆列

OperatorTypeEnum

1. import lombok.AllArgsConstructor;
2. import lombok.Getter;
4. /**
5. * @author whiteBrocade
6. * @version 1.0
7. * @description 操作类型枚举
8. */
9. @Getter
10. @AllArgsConstructor
11. public enum OperatorTypeEnum {
12.     /**
13.      * 新增
14.      */
15.     INSERT(1),
17.     /**
18.      * 修改
19.      */
20.     UPDATE(2),
22.     /**
23.      * 删除
24.      */
25.     DELETE(3),
26.     ;
28.     private final int type;
29. }

复制代码

StrategyEnum

1. import cn.hutool.core.bean.BeanUtil;
2. import cn.hutool.core.util.ObjUtil;
3. import cn.hutool.json.JSONObject;
4. import cn.hutool.json.JSONUtil;
5. import lombok.AllArgsConstructor;
6. import lombok.Getter;
8. import java.beans.Introspector;
10. /**
11. * @author whiteBrocade
12. * @version 1.0
13. * @description 表处理策略
14. * todo 后续在这里新增相关枚举即可
15. */
16. @Getter
17. @AllArgsConstructor
18. public enum StrategyEnum {
19.     /**
20.      * Student处理策略
21.      */
22.     STUDENT("student", Student.class, Introspector.decapitalize(StudentLogHandler.class.getSimpleName())),
23.     ;
25.     /**
26.      * 表名
27.      */
28.     private final String tableName;
30.     /**
31.      * class对象
32.      */
33.     private final Class<?> varClass;
35.     /**
36.      * 处理器名
37.      */
38.     private final String handlerName;
40.     /**
41.      * 策略选择器, 根据传入的 DataChangeInfo 对象中的 tableName 属性, 从一系列预定义的策略 (StrategyEnum) 中选择一个合适的处理策略, 并封装进 StrategyHandleSelector 对象中返回
42.      *
43.      * @param dataChangeInfo 数据变更对象
44.      * @return StrategyHandlerSelector
45.      */
46.     public static StrategyHandleSelector getSelector(DataChangeInfo dataChangeInfo) {
47.         if (ObjUtil.isNull(dataChangeInfo)) {
48.             return null;
49.         }
50.         String tableName = dataChangeInfo.getTableName();
51.         StrategyHandleSelector selector = new StrategyHandleSelector();
52.         // 遍历所有的策略枚举(StrategyEnum), 寻找与当前表名相匹配的策略
53.         for (StrategyEnum strategyEnum : values()) {
54.             // 如果找到匹配的策略, 创建并配置 StrategyHandleSelector
55.             if (strategyEnum.getTableName().equals(tableName)) {
56.                 selector.setHandlerName(strategyEnum.handlerName);
57.                 selector.setOperatorTime(dataChangeInfo.getOperatorTime());
58.                 selector.setOperatorType(dataChangeInfo.getOperatorType());
59.                 JSONObject jsonObject = JSONUtil.parseObj(dataChangeInfo.getData());
60.                 selector.setData(BeanUtil.copyProperties(jsonObject, strategyEnum.varClass));
61.                 return selector;
62.             }
63.         }
64.         return null;
65.     }
66. }

复制代码

pojo

Student

1. import lombok.Data;
3. /**
4. * @author whiteBrocade
5. * @version 1.0
6. * @description 学生类, 用于演示
7. */
8. @Data
9. public class Student {
10.     /**
11.      * id
12.      */
13.     private Integer id;
15.     /**
16.      * 姓名
17.      */
18.     private String name;
20.     /**
21.      * 年龄
22.      */
23.     private Integer age;
24. }

复制代码

DataChangeInfo

1. import lombok.Data;
3. /**
4. * @author whiteBrocade
5. * @version 1.0
6. * @description 数据变更对象
7. */
8. @Data
9. public class DataChangeInfo {
10.     /**
11.      * 变更前数据
12.      */
13.     private String beforeData;
15.     /**
16.      * 变更后数据
17.      */
18.     private String afterData;
20.     /**
21.      * 操作的数据
22.      */
23.     private String data;
25.     /**
26.      * 变更类型 1->新增 2->修改 3->删除
27.      */
28.     private Integer operatorType;
30.     /**
31.      * binlog文件名
32.      */
33.     private String fileName;
35.     /**
36.      * binlog当前读取点位
37.      */
38.     private Integer filePos;
39.     /**
40.      * 数据库名
41.      */
42.     private String database;
44.     /**
45.      * 表名
46.      */
47.     private String tableName;
49.     /**
50.      * 变更时间
51.      */
52.     private Long operatorTime;
53. }

复制代码

自界说Sink

DataChangeSink

1. import cn.hutool.core.util.ObjUtil;
2. import lombok.AllArgsConstructor;
3. import lombok.extern.slf4j.Slf4j;
4. import org.apache.flink.api.common.functions.OpenContext;
5. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
6. import org.springframework.stereotype.Component;
8. import java.io.Serializable;
9. import java.util.Map;
11. /**
12. * @author whiteBrocade
13. * @version 1.0
14. * @description
15. */
16. @Slf4j
17. @Component
18. @AllArgsConstructor
19. public class DataChangeSink extends RichSinkFunction<DataChangeInfo> implements Serializable {
20.     /**
21.      * BaseLogHandler相关的缓存
22.      * Spring自动将相关BaseLogHandler的Bean注入注入到本地缓存Map中
23.      */
24.     private final Map<String, BaseLogHandler> strategyHandlerMap;
26.     @Override
27.     public void invoke(DataChangeInfo value, Context context) {
28.         log.info("收到变更原始数据:{}", value);
29.         // 选择策略
30.         StrategyHandleSelector selector = StrategyEnum.getSelector(value);
31.         if (ObjUtil.isNull(selector)) {
32.             return;
33.         }
34.         BaseLogHandler<Object> handler = strategyHandlerMap.get(selector.getHandlerName());
36.         // insert操作
37.         if (selector.getOperatorType().equals(OperatorTypeEnum.INSERT.getType())) {
38.             handler.handleInsertLog(selector.getData(), selector.getOperatorTime());
39.             return;
40.         }
41.         // update操作
42.         if (selector.getOperatorType().equals(OperatorTypeEnum.UPDATE.getType())) {
43.             handler.handleUpdateLog(selector.getData(), selector.getOperatorTime());
44.             return;
45.         }
46.         // delete操作
47.         if (selector.getOperatorType().equals(OperatorTypeEnum.DELETE.getType())) {
48.             handler.handleDeleteLog(selector.getData(), selector.getOperatorTime());
49.         }
50.     }
52.     /**
53.      * 前置操作
54.      */
55.     @Override
56.     public void open(OpenContext openContext) throws Exception {
57.         super.open(openContext);
58.     }
60.     /**
61.      * 后置操作
62.      */
63.     @Override
64.     public void close() throws Exception {
65.         super.close();
66.     }
67. }

复制代码

CustomSink

1. import lombok.extern.slf4j.Slf4j;
2. import org.apache.flink.api.common.functions.OpenContext;
3. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
4. import org.springframework.stereotype.Component;
6. /**
7. * @author whiteBrocade
8. * @version 1.0
9. * @description 自定义Sink处理器, 这个是根据ognl表达式区分ddl语句类型, 搭配
10. */
11. @Slf4j
12. @Component
13. public class CustomSink extends RichSinkFunction<String> {
14.     @Override
15.     public void invoke(String json, Context context) throws Exception {
16.         // op字段:  该字段也有4种取值，分别是C(create)、U(Update)、D(Delete)、Read
17.         // 对于U操作，其数据部分同时包含了Before和After。
18.         log.info("监听到数据: {}", json);
19.         String op = JSONUtil.getValue(json, "op", String.class);
20.         // 语句的id
21.         Integer id = null;
22.         // 如果是update语句
23.         if ("u".equals(op)) {
24.             id = JSONUtil.getValue(json, "after.id", Integer.class);
25.             log.info("执行update语句");
26.             // 执行update语句
27.         }
29.         // 如果是delete语句
30.         if ("d".equals(op)) {
31.             id = JSONUtil.getValue(json, "before.id", Integer.class);
32.             log.info("执行delete语句");
33.             // 执行删除语句
34.         }
35.         // 如果是新增
36.         if ("c".equals(op)) {
37.             log.info("执行insert语句");
38.         }
39.     }
41.     // 前置操作
42.     @Override
43.     public void open(OpenContext openContext) throws Exception {
44.         super.open(openContext);
45.     }
47.     // 后置操作
48.     @Override
49.     public void close() throws Exception {
50.         super.close();
51.     }
52. }

复制代码

自界说反序列化器 MySQLDeserialization

1. import com.alibaba.fastjson.JSONObject;
2. import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.Field;
3. import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.Schema;
4. import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.Struct;
5. import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.source.SourceRecord;
6. import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
7. import io.debezium.data.Envelope;
8. import lombok.extern.slf4j.Slf4j;
9. import org.apache.flink.api.common.typeinfo.TypeInformation;
10. import org.apache.flink.util.Collector;
11. import org.springframework.stereotype.Service;
13. import java.util.List;
14. import java.util.Optional;
16. /**
17. * @author whiteBrocade
18. * @version 1.0
19. * @description MySQL消息读取 自定义反序列化器
20. */
21. @Slf4j
22. @Service
23. public class MySQLDeserialization implements DebeziumDeserializationSchema<DataChangeInfo> {
25.     public static final String TS_MS = "ts_ms";
26.     public static final String BIN_FILE = "file";
27.     public static final String POS = "pos";
28.     public static final String CREATE = "CREATE";
29.     public static final String BEFORE = "before";
30.     public static final String AFTER = "after";
31.     public static final String SOURCE = "source";
32.     public static final String UPDATE = "UPDATE";
34.     /**
35.      * 反序列化数据, 转为变更JSON对象
36.      *
37.      * @param sourceRecord SourceRecord
38.      * @param collector Collector<DataChangeInfo>
39.      */
40.     @Override
41.     public void deserialize(SourceRecord sourceRecord, Collector<DataChangeInfo> collector) {
42.         try {
43.             // 根据主题的格式，获取数据库名(database)和表名(tableName)
44.             String topic = sourceRecord.topic();
45.             String[] fields = topic.split("\\.");
46.             String database = fields[1];
47.             String tableName = fields[2];
49.             Struct struct = (Struct) sourceRecord.value();
50.             final Struct source = struct.getStruct(SOURCE);
51.             DataChangeInfo dataChangeInfo = new DataChangeInfo();
52.             // 获取操作类型  CREATE UPDATE DELETE
53.             Envelope.Operation operation = Envelope.operationFor(sourceRecord);
54.             String type = operation.toString().toUpperCase();
55.             int eventType = type.equals(CREATE) ? OperatorTypeEnum.INSERT.getType() : UPDATE.equals(type) ?
56.                     OperatorTypeEnum.UPDATE.getType() : OperatorTypeEnum.DELETE.getType();
58.             // 一般情况是无需关心其之前之后数据的, 直接获取最新的日志数据即可, 但这里为了演示, 都进行输出
59.             // 获取变更前和变更后的数据，并将其设置到DataChangeInfo对象中
60.             dataChangeInfo.setBeforeData(this.getJsonObject(struct, BEFORE).toJSONString());
61.             dataChangeInfo.setAfterData(this.getJsonObject(struct, AFTER).toJSONString());
62.             if (eventType == OperatorTypeEnum.DELETE.getType()) {
63.                 dataChangeInfo.setData(this.getJsonObject(struct, BEFORE).toJSONString());
64.             } else {
65.                 dataChangeInfo.setData(this.getJsonObject(struct, AFTER).toJSONString());
66.             }
67.             dataChangeInfo.setOperatorType(eventType);
68.             dataChangeInfo.setFileName(Optional.ofNullable(source.get(BIN_FILE))
69.                     .map(Object::toString)
70.                     .orElse(""));
71.             dataChangeInfo.setFilePos(Optional.ofNullable(source.get(POS))
72.                     .map(x -> Integer.parseInt(x.toString()))
73.                     .orElse(0));
74.             dataChangeInfo.setDatabase(database);
75.             dataChangeInfo.setTableName(tableName);
76.             dataChangeInfo.setOperatorTime(Optional.ofNullable(struct.get(TS_MS))
77.                     .map(x -> Long.parseLong(x.toString()))
78.                     .orElseGet(System::currentTimeMillis));
79.             // 输出数据
80.             collector.collect(dataChangeInfo);
81.         } catch (Exception e) {
82.             log.error("反序列binlog失败", e);
83.         }
84.     }
86.     /**
87.      * 从源数据获取出变更之前或之后的数据
88.      *
89.      * @param value Struct
90.      * @param fieldElement 字段
91.      * @return JSONObject
92.      */
93.     private JSONObject getJsonObject(Struct value, String fieldElement) {
94.         Struct element = value.getStruct(fieldElement);
95.         JSONObject jsonObject = new JSONObject();
96.         if (element != null) {
97.             Schema afterSchema = element.schema();
98.             List<Field> fieldList = afterSchema.fields();
99.             for (Field field : fieldList) {
100.                 Object afterValue = element.get(field);
101.                 jsonObject.put(field.name(), afterValue);
102.             }
103.         }
104.         return jsonObject;
105.     }
108.     @Override
109.     public TypeInformation<DataChangeInfo> getProducedType() {
110.         return TypeInformation.of(DataChangeInfo.class);
111.     }
112. }

复制代码

LogHandler

BaseLogHandler

1. import java.io.Serializable;
3. /**
4. * @author whiteBrocade
5. * @version 1.0
6. * @description 日志处理器
7. * todo 新建一个类实现该BaseLogHandler类, 添加相应的处理逻辑即可, 可参考StudentLogHandler实现
8. */
9. public interface BaseLogHandler<T> extends Serializable {
10.     /**
11.      * 日志处理
12.      *
13.      * @param data 数据转换后模型
14.      * @param operatorTime 操作时间
15.      */
16.     void handleInsertLog(T data, Long operatorTime);
18.     /**
19.      * 日志处理
20.      *
21.      * @param data 数据转换后模型
22.      * @param operatorTime 操作时间
23.      */
24.     void handleUpdateLog(T data, Long operatorTime);
26.     /**
27.      * 日志处理
28.      *
29.      * @param data 数据转换后模型
30.      * @param operatorTime 操作时间
31.      */
32.     void handleDeleteLog(T data, Long operatorTime);
33. }

复制代码

StrategyHandleSelector

1. import lombok.Data;
3. /**
4. * @author whiteBrocade
5. * @version 1.0
6. * @description 策略处理选择器
7. */
8. @Data
9. public class StrategyHandleSelector {
10.     /**
11.      * 策略处理器名称
12.      */
13.     private String handlerName;
15.     /**
16.      * 数据源
17.      */
18.     private Object data;
20.     /**
21.      * 操作时间
22.      */
23.     private Long operatorTime;
25.     /**
26.      * 操作类型
27.      */
28.     private Integer operatorType;
29. }

复制代码

StudentLogHandler

1. import lombok.extern.slf4j.Slf4j;
2. import org.springframework.stereotype.Service;
4. /**
5. * @author whiteBrocade
6. * @version 1.0
7. * @description Student对应处理器
8. */
9. @Slf4j
10. @Service
11. public class StudentLogHandler implements BaseLogHandler<Student> {
12.     @Override
13.     public void handleInsertLog(Student data, Long operatorTime) {
14.         log.info("处理Student表的新增日志: {}", data);
15.     }
17.     @Override
18.     public void handleUpdateLog(Student data, Long operatorTime) {
19.         log.info("处理Student表的修改日志: {}", data);
20.     }
22.     @Override
23.     public void handleDeleteLog(Student data, Long operatorTime) {
24.         log.info("处理Student表的删除日志: {}", data);
25.     }
26. }

复制代码

JSONUtil

1. import com.google.gson.Gson;
2. import com.google.gson.reflect.TypeToken;
3. import ognl.Ognl;
4. import ognl.OgnlContext;
6. import java.util.Map;
8. /**
9. * @author whiteBrocade
10. * @version 1.0
11. * @description: JSON工具类
12. */
13. public class JSONUtil {
14.     /**
15.      * 将指定JSON转为Map对象, Key类型为String,对应JSON的key
16.      * Value分情况:
17.      * 1. Value是字符串, 自动转为字符串, 例如:{"a","b"}
18.      * 2. Value是其他JSON对象, 自动转为Map,例如::{"a":{"b":"2"}}
19.      * 3. Value是数组, 自动转为list<Map>,例如::{"a":[:{"b":"2"},"c":"3"]}
20.      *
21.      * @param json 输入的的JSON对象
22.      * @return 动态Map集合
23.      */
24.     public static Map<String, Object> transferToMap(String json) {
25.         Gson gson = new Gson();
26.         Map<String, Object> map = gson.fromJson(json, new TypeToken<Map<String, Object>>() {}.getType());
27.         return map;
28.     }
30.     /**
31.      * 获取指定JSON的指定路径的值
32.      *
33.      * @param json  原始JSON数据
34.      * @param path  OGNL原则表达式
35.      * @param clazz Value对应的目标类
36.      * @return clazz对应的数据
37.      */
38.     public static <T> T getValue(String json, String path, Class<T> clazz) {
39.         try {
40.             Map<String, Object> map = JSONUtil.transferToMap(json);
41.             OgnlContext ognlContext = new OgnlContext();
42.             ognlContext.setRoot(map);
43.             T value = (T) Ognl.getValue(path, ognlContext, ognlContext.getRoot(), clazz);
44.             return value;
45.         } catch (Exception e) {
46.             throw new RuntimeException(e);
47.         }
48.     }
49. }

复制代码

MysqlEventListener

1. import com.ververica.cdc.connectors.mysql.source.MySqlSource;
2. import com.ververica.cdc.connectors.mysql.source.MySqlSourceBuilder;
3. import com.ververica.cdc.connectors.mysql.table.StartupOptions;
4. import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
5. import lombok.AllArgsConstructor;
6. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
8. import org.springframework.boot.ApplicationArguments;
9. import org.springframework.boot.ApplicationRunner;
10. import org.springframework.stereotype.Component;
12. /**
13. * @author whiteBrocade
14. * @version 1.0
15. * @description MySQL变更监听
16. */
17. @Component
18. @AllArgsConstructor
19. public class MysqlEventListener implements ApplicationRunner {
21.     /**
22.      * Flink CDC相关配置
23.      */
24.     private final FlinkCDCConfig flinkCDCConfig;
26.     /**
27.      * 自定义Sink
28.      * customSink: 通过ognl解析ddl语句类型
29.      * dataChangeSink: 通过struct解析ddl语句类型
30.      * 通常两个选择一个就行
31.      */
32.     private final CustomSink customSink;
33.     private final DataChangeSink dataChangeSink;
35.     /**
36.      * 自定义反序列化处理器
37.      */
38.     private final MySQLDeserialization mySQLDeserialization;
40.     @Override
41.     public void run(ApplicationArguments args) throws Exception {
42.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
43.         // 设置整个Flink程序的默认并行度
44.         env.setParallelism(flinkCDCConfig.getParallelism());
45.         // 设置checkpoint 间隔
46.         env.enableCheckpointing(flinkCDCConfig.getEnableCheckpointing());
47.         // 设置任务关闭的时候保留最后一次 CK 数据
48.         env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
50.         // todo 下列的两个MySqlSource选择一个
51.         // 自定义的反序列化器
52.         // MySqlSource<DataChangeInfo> mySqlSource = this.buildBaseMySqlSource(DataChangeInfo.class)
53.         //         .deserializer(mySQLDeserialization)
54.         //         .build();
56.         // Flink CDC自带的反序列化器
57.         MySqlSource<String> mySqlSource = this.buildBaseMySqlSource(String.class)
58.             .deserializer(new JsonDebeziumDeserializationSchema())
59.             .build();
62.         env.fromSource(mySqlSource,
63.                        WatermarkStrategy.noWatermarks(),
64.                        "mysql-source")
65.             // 设置该数据源的并行度
66.             .setParallelism(flinkCDCConfig.getParallelism())
67.             // todo 根据上述的选择，选择对应的Sink
68.             // .addSink(dataChangeSink); // 添加Sink, 这里配合mySQLDeserialization+dataChangeSink
69.             .addSink(customSink);
71.         env.execute("mysql-stream-cdc");
72.     }
74.     /**
75.      * 构建基本的MySqlSourceBuilder
76.      *
77.      * @param clazz 返回的数据类型Class对象
78.      * @param <T>   源数据中存储的类型
79.      * @return MySqlSourceBuilder
80.      */
81.     private <T> MySqlSourceBuilder<T> buildBaseMySqlSource(Class<T> clazz) {
82.         return MySqlSource.<T>builder()
83.             .hostname(flinkCDCConfig.getHostname())
84.             .port(flinkCDCConfig.getPort())
85.             .username(flinkCDCConfig.getUsername())
86.             .password(flinkCDCConfig.getPassword())
87.             .databaseList(flinkCDCConfig.getDatabaseList())
88.             .tableList(flinkCDCConfig.getTableList())
89.             /* initial: 初始化快照,即全量导入后增量导入(检测更新数据写入)
90.                  * latest: 只进行增量导入(不读取历史变化)
91.                  * timestamp: 指定时间戳进行数据导入(大于等于指定时间错读取数据)
92.                  */
93.             .startupOptions(StartupOptions.latest())
94.             .includeSchemaChanges(flinkCDCConfig.getIncludeSchemaChanges()) // 包括schema的改变
95.             .serverTimeZone("GMT+8"); // 时区
96.     }
97. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。