SpringBoot整合Apache Flink的详细指南

<dependencies>
    <!-- Spring Boot Starter -->
    <dependency>
     www.chinasem.cn   <grouwww.chinasem.cnpId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>

    <!-- Flink 核心依赖 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.17.2</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>1.17.2</version>
        <scope>provided</scope>
    </dependency>

​​​​​​​   <!-- 本地执行时需添加 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-runtime&androidlt;/artifactId>
        <version>1.17.2</version>
        <scope>test</scope>
    </dependency>
</dependencies>

// src/main/java/com/example/demo/flink/WordCountJob.java
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class WordCountJob {
    public static void execute() throws Exception {
        // 1. 获取 Flink 执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 定义输入数据
        DataStream<String> text = env.fromElements(
            "Spring Boot整合Flink",
            "Flink实时流处理",
            "Spring生态集成"
        );

        // 3. 处理数据流
        DataStream<Tuple2<String, Integer>> counts = text
            .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
                    for (String word : value.split("\\s")) {
                        out.collect(new Tuple2<>(word, 1));
                    }
                }
            })
            .keyBy(value -> value.f0) // 按单词分组
            .sum(1); // 对计数求和

        // 4. 打印结果
        counts.print();

        // 5. 启动作业
        env.execute("WordCountJob");
    }
}

// src/main/java/com/example/demo/DemoApplication.java
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class DemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
        System.out.println("Spring Boot Application Started...");

        try {
            // 触发 Flink 作业
            WordCountJob.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}