SpringBoot集成flink

Spring Boot 整合 Flink 主要涉及到以下几个步骤：

1. 添加 Flink 依赖到 Spring Boot 项目的 pom.xml 文件中。
2. 配置 Flink 环境相关的属性。
3. 创建 Flink 的 StreamExecutionEnvironment 和其他需要的组件，例如 StreamTableEnvironment。
4. 在 Spring 容器中初始化 Flink 组件。
5. 编写 Flink 作业并运行。

以下是一个简单的示例：

pom.xml 添加 Flink 依赖：

<dependencies>
    <!-- 添加 Flink 支持 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.11.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.11</artifactId>
        <version>1.11.0</version>
    </dependency>
</dependencies>

application.properties 配置 Flink 属性：

# Flink 相关配置
flink.job-manager-ip=localhost
flink.job-manager-port=8081
flink.parallelism=1

Flink 配置类：

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class FlinkConfig {
 
    @Bean
    public StreamExecutionEnvironment env(ParameterTool parameterTool) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().setGlobalJobParameters(parameterTool);
        env.setParallelism(parameterTool.getInt("flink.parallelism", 1));
        return env;
    }
}

Flink 作业类：

import org.apache.flink.api.common.functions.MapFunction;
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.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
 
@Component
public class FlinkStreamingJob {
 
    private final StreamExecutionEnvironment env;
 
    @Autowired
    public FlinkStreamingJob(StreamExecutionEnvironment env) {
        this.env = env;
    }
 
    public void run() throws Exception {
        DataStream<Tuple2<String, Integer>> dataStream = env
                .fromElements(Tuple2.of("hello", 1), Tuple2.of("world", 2))
                .map(new MapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map