SpringBoot 整合 Flink CDC 实时追踪 MySQL 数据变动

3. 添加依赖

在 pom.xml 中添加 Apache Flink 和 Flink CDC 的依赖。以下是必要的依赖：

<dependencies>
    <!-- Flink dependency -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.14.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>1.14.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-mysql-cdc</artifactId>
        <version>2.0.0</version>
    </dependency>
    <!-- Spring Boot dependencies -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
</dependencies>

4. 配置 Flink 和 MySQL CDC

在 Spring Boot 的 application.yml 或 application.properties 文件中配置 Flink 和 MySQL 数据库连接：

flink:
  checkpoint:
    interval: 10000
  parallelism: 1
spring:
  datasource:
    url: jdbc:mysql://localhost:3306/your_database
    username: your_username
    password: your_password

5. 实现数据实时追踪

创建一个服务类来实现数据的实时追踪：

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.springframework.stereotype.Service;

@Service
public class FlinkCdcService {
    public void startDataStreaming() {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 使用 Flink CDC 连接 MySQL
        String name = "inventory";
        tableEnv.executeSql("CREATE TABLE " + name + " ("
                + " id INT,"
                + " name STRING,"
                + " description STRING,"
                + " weight DECIMAL(10, 3)"
                + ") WITH ("
                + " 'connector' = 'mysql-cdc',"
                + " 'hostname' = 'localhost',"
                + " 'port' = '3306',"
                + " 'username' = 'your_username',"
                + " 'password' = 'your_password',"
                + " 'database-name' = 'your_database',"
                + " 'table-name' = 'your_table'"
                + ")");
        // 查询并打印结果
        DataStream<String> dataStream = tableEnv.sqlQuery("SELECT * FROM " + name).execute().print();
        try {
            env.execute("Flink CDC Demo");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

6. 启动 Spring Boot 应用

在你的 Spring Boot 应用的启动类中调用 FlinkCdcService 的 startDataStreaming 方法来启动数据追踪：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class FlinkCdcApplication implements CommandLineRunner {
    @Autowired
    private FlinkCdcService flinkCdcService;

    public static void main(String[] args) {
        SpringApplication.run(FlinkCdcApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        flinkCdcService.startDataStreaming();
    }
}

7. 运行并测试

运行 Spring Boot 应用，并在 MySQL 数据库中做出一些数据变动。你应该能在控制台看到实时打印的数据变动。