springboot 使用flinkCDC 基于bin_log 监听mysql 数据 到 mysql （多表分片处理）_flink cdc 同步mysql数据 springboot版本

前提： 被监听数据库需要开启bin_log , 账号需要有可查看bin_log日志权限

1. 添加依赖

    <flink-version>1.12.0</flink-version>
    <flink-mysql-version>2.0.0</flink-mysql-version>



    <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink-version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>${flink-version}</version>
        </dependency>


        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.11</artifactId>
            <version>${flink-version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_2.11</artifactId>
            <version>1.12.0</version>
        </dependency>


        <dependency>
            <groupId>com.ververica</groupId>
            <artifactId>flink-connector-mysql-cdc</artifactId>
            <version>${flink-mysql-version}</version>

        </dependency>

        <!--flinkcdc 结束-->

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
40
41

2. 编写封装自定义监听返回结果集 json

/**
 * @author: xrp
 * @date: 2022/05/05/10:25
 * @description
 */
public class LinkConfig  implements DebeziumDeserializationSchema<String> {

    /**
     *
     * @param sourceRecord
     * @param collector
     */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) {
        JSONObject result = new JSONObject();
        String topic = sourceRecord.topic();
        String[] fields = topic.split("\\.");
        result.put("db", fields[1]);
        result.put("tableName", fields[2]);

        //获取before数据
        Struct value = (Struct) sourceRecord.value();
        Struct before = value.getStruct("before");
        JSONObject beforeJson = new JSONObject();
        if (before != null) {
            beforeJson = getJson(before);
        }
        result.put("before", beforeJson);

        //获取after数据
        Struct after = value.getStruct("after");
        JSONObject afterJson = new JSONObject();
        if (after != null) {
            //获取列信息
            afterJson = getJson(after);
        }
        result.put("after", afterJson);

        //获取操作类型
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        result.put("op", operation);
        collector.collect(result.toJSONString());
    }

    private JSONObject getJson(Struct struct) {
        JSONObject jsonObject = new JSONObject();
        //获取列信息
        Schema schema = struct.schema();
        List<Field> fieldList = schema.fields();

        for (Field field : fieldList) {
            jsonObject.put(field.name(), struct.get(field));
        }
        return jsonObject;
    }




    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}

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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

3. 编写主程序，实现CommandLineRunner ，并且使用@component (目的为了自动执行改方法，进行实时数据监听)

/**
 * @author: xrp
 * @date: 2022/05/05/14:49
 * @description
 */
@Component
public class StartUp implements CommandLineRunner {
    @Override
    public void run(String... args) throws Exception {
        

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DebeziumSourceFunction<String> sourceFunction = MySqlSource.<String>builder()
                .hostname("要监听的数据库地址")
                .port(端口号)
                .username("数据库账户")
                .password("数据库密码")
                .databaseList("数据库")
                .tableList("数据库.数据库表1", "数据库。数据库表2") //多个表逗号分隔
                .deserializer(new LinkConfig()) //自定义返回结果集
                .startupOptions(StartupOptions.initial())
                .serverTimeZone("UTC")
                .build();

        DataStreamSource<String> streamSource = env.addSource(sourceFunction);
        // 多表进行分片处理
        OutputTag<String> orderTag = new OutputTag<>("表1", Types.STRING);

        OutputTag<String> userTag = new OutputTag<>("表2",Types.STRING);


        SingleOutputStreamOperator<String> process = streamSource.map((MapFunction<String, JSONObject>) JSON::parseObject).process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context context, Collector<String> collector) {
                if ("表1".equals(value.getString("tableName"))) {
                    context.output(orderTag, value.toJSONString());
                } else if ("表2".equals(value.getString("tableName"))) {
                    context.output(userTag, value.toJSONString());
                }
            }
        });
        DataStream<String> orderStream = process.getSideOutput(orderTag);
        DataStream<String> userStream = process.getSideOutput(userTag);
        orderStream.print();
        userStream.print();
        //自定义sink
        streamSource.addSink(new ListenerOrderSink());
        userStream.addSink(new  ListenerUserSink());
        env.executeAsync("fLinkCDC");
    }
}

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
40
41
42
43
44
45
46
47
48
49
50
51
52
53

4. 编写sink （一张表一个对应一个sink）

/**
 * @author: xrp
 * @date: 2022/05/24/17:27
 * @description  订单sink
 */
public class ListenerOrderSink extends RichSinkFunction<String> {

    private static final Logger LOGGER = LoggerFactory.getLogger(ListenerOrderSink.class);


    private PreparedStatement ps = null;

    private Connection connection = null;
    String driver = "com.mysql.cj.jdbc.Driver";
    String url = "jdbc:mysql://要将监听到的数据同步到哪，另一个数据库地址:端口号/数据库名字?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC";
    String username = "数据库账号";
    String password = "数据库密码";

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = getConn();
        ps = connection.prepareStatement("insert into 数据库名.表名values (?,?,?,?,?,?,?,?,?,?,?)");
    }
    private Connection getConn() {
        try {
            Class.forName(driver);
            connection = DriverManager.getConnection(url, username, password);
            LOGGER.error("数据库连接成功");
        } catch (Exception e) {
            LOGGER.error("数据库连接失败");
        }
        return connection;
    }


    @Override
    public void invoke(String p, Context context) throws Exception {
    //TranslateJson  将自定义的返回结果集转为具体实体
        TranslateJson translateJson = JSON.parseObject(p, TranslateJson.class);
        if (ConstantValue.CREATE_INFO.equals(translateJson.getOp())) {
            ErpOrder erpOrder = JSON.parseObject(translateJson.getAfter(), ErpOrder.class);
            // 注意： 字段个数需要与表字段个数对应上
            ps.setString(1,erpOrder.getId());
            ps.setString(2,erpOrder.getCode());
            ps.setString(3,erpOrder.getCustomerName());
            ps.setBigDecimal(4,erpOrder.getOrderAmount());
            ps.setString(5,erpOrder.getUserId());
            ps.setString(6,erpOrder.getUserName());
            ps.setString(7,erpOrder.getCreateName());
            ps.setString(8,erpOrder.getCreateDate());
            ps.setInt(9,erpOrder.getStatus());
            ps.setInt(10,erpOrder.getOrderId());
            ps.setString(11,erpOrder.getOrderCode());
            ps.executeUpdate();
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        if(connection != null){
            connection.close();
        }
        if (ps != null){
            ps.close();
        }
    }


}

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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

5. 将自定义统一监听结果集转为实体

/**
 * @author: xrp
 * @date: 2022/05/05/10:50
 * @description
 */
@Data
public class TranslateJson {
    private static final long serialVersionUID = -74375380912179188L;
    private String op;
    private String before;
    private String after;
    private String db;
    private String tableName;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15