查阅无数资料，爬了无数个坑!!!
整体思路:把SpringBoot当成一个任务放进Flink集群中运行，并且该任务会一直运行，当其他任务需要执行时只需要调用SpringBoot的接口来动态生成任务，可以把每一个接口都当成一个任务，调用接口时Flink会根据当前环境动态创建任务并执行
注意事项: 使用 ./flink run 命令以后台运行的方式去运行打好的jar包
一、引入以下依赖
8 8 1.8 1.13.0 2.12 1.7.30 5.1.47 2.0.3.RELEASE org.springframework.boot spring-boot-starter-parent2.1.1.RELEASE org.springframework.boot spring-boot-starter-weborg.springframework.boot spring-boot-starter-logging* *org.apache.flink flink-java${flink.version} org.apache.flink flink-streaming-java_2.11${flink.version} org.apache.flink flink-clients_2.11${flink.version} org.apache.flink flink-runtime-web_2.11${flink.version} org.springframework.boot spring-boot-starter${spring.boot.version} org.apache.flink flink-connector-jdbc_${scala.binary.version}${flink.version} org.projectlombok lombok1.18.8 provided mysql mysql-connector-java${mysql.version} flink org.apache.maven.plugins maven-shade-plugin3.2.4 package shade false com.google.code.findbugs:jsr305 org.slf4j:* log4j:* *:* module-info.class META-INF/*.SF META-INF/*.DSA META-INF/*.RSA META-INF/spring.handlers reference.conf META-INF/spring.factories META-INF/spring.schemas cn.sdata.FlinkBootApplication
二、yml配置
server: port: 10001
三、SpringBoot启动类
@SpringBootApplication
public class FlinkBootApplication {
public static void main(String[] args) {
SpringApplication.run(FlinkBootApplication.class,args);
//加一个死循环来保证main方法不会停止
while (true) {
Thread.sleep(30000);
}
}
}
四、Controller层业务测试代码(一个流处理、一个批处理)
@RequestMapping("test")
@RestController
@AllArgsConstructor
public class TestController {
@GetMapping("test1")
public void test(Integer parallelism) throws Exception {
//创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//设置并行度
env.setParallelism(parallelism);
//读取文本流
DataStreamSource source = env.socketTextStream("192.168.1.200", 8888);
SingleOutputStreamOperator> operator = source.flatMap(new FlatMapFunction>() {
@Override
public void flatMap(String s, Collector> collector) throws Exception {
String[] splits = s.split(" ");
for (String split : splits) {
collector.collect(new Tuple2<>(split, 1));
}
}
}).filter(data -> StringUtils.isNotEmpty(data.f0)).keyBy(data -> data.f0).sum(1);
//打印
operator.print();
//执行
env.execute();
}
@GetMapping("test2")
public void test2() throws Exception {
//执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//设置并行度
env.setParallelism(1);
//将一列元素作为数据源
DataStreamSource integerDataStream = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12);
//控制台打印
integerDataStream.print("int");
//执行任务
env.execute();
}
}
五、在你的服务器上开启一个nc端口
六、打包上传至你的JobManager所在的服务器
七、进入Flink中的bin目录
八、后台方式运行jar包
九、通过接口访问流处理方法并行度设置为4
可以看到流处理方法所占用了4个插槽
十、通过接口访问批处理方法
可以看到该方法占用1个插槽
稍等一会再次刷新页面
发现该任务以执行完毕，插槽不再占用
总结:完成SpringBoot和Flink的高度整合，通过调用接口的方式来进行Flink任务的创建，SpringBoot在处理一些数据量比较大且计算量也比较大的业务场景时就可以通过Flink进行数据的处理入库，并且支持Flink集群节点可无限制扩充
