Flink-03 Flink Java 3分钟上手 Stream 给 Flink-02 DataStreamSource Socket写一个测试的工具！

代码仓库

会同步代码到 GitHub
https://github.com/turbo-duck/flink-demo

在这里插入图片描述

当前章节

继续上一节的内容：https://blog.csdn.net/w776341482/article/details/139875037

上一节中，我们需要使用 nc 或者 telnet 等工具来模拟 Socket 流。这节我们写一个 ServerSocket 来模拟这些操作，让流自动的写入不用我们手动去操作了。

POM.xml

与上一节一致，不需要修改

编写代码

还是和上一节一样的 Socket 流，这里略去其他的代码

DataStreamSource<String> textStream = streamExecutionEnvironment.socketTextStream(ip, port, "\n");

FlinkServer

继承 Thread 启动一个线程来进行Flink的服务

package icu.wzk.demo03;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;public class FlinkServer extends Thread {@Overridepublic void run() {String ip = "0.0.0.0";int port = 9999;StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> textStream = streamExecutionEnvironment.socketTextStream(ip, port, "\n");SingleOutputStreamOperator<Tuple2<String, Long>> tuple2SingleOutputStreamOperator = textStream.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {@Overridepublic void flatMap(String s, Collector<Tuple2<String, Long>> collector) throws Exception {String[] splits = s.split("\\s");for (String word : splits) {collector.collect(Tuple2.of(word, 1L));}}});SingleOutputStreamOperator<Tuple2<String, Long>> word = tuple2SingleOutputStreamOperator.keyBy(new KeySelector<Tuple2<String, Long>, Object>() {@Overridepublic Object getKey(Tuple2<String, Long> stringLongTuple2) throws Exception {return stringLongTuple2.f0;}}).window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1))).sum(1);System.out.println("wait word print()");word.print();try {streamExecutionEnvironment.execute("stream!");} catch (Exception e) {throw new RuntimeException(e);}}
}

NumRandom

使用 ServerSocket 实现一个持续的流输出

package icu.wzk.demo03;import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Random;public class RandomNumClient extends Thread {@Overridepublic void run() {String ip = "0.0.0.0";int port = 9999;try {ServerSocket serverSocket = new ServerSocket();InetSocketAddress address = new InetSocketAddress(ip, port);serverSocket.bind(address);Socket socket = serverSocket.accept();OutputStream output = socket.getOutputStream();PrintWriter writer = new PrintWriter(output, true);Random random = new Random();for (int i = 0; i < 500; i ++) {int randomNumber = random.nextInt(10) + 1;writer.println(randomNumber);System.out.println("ServerSocket Send To Flink: " + randomNumber);Thread.sleep(200);}} catch (Exception e) {throw new RuntimeException(e);}}}

StartApp

将上述的两个类组装起来

请添加图片描述

package icu.wzk.demo03;public class StartApp {public static void main(String[] args) throws Exception {RandomNumClient randomNumClient = new RandomNumClient();FlinkServer flinkServer = new FlinkServer();flinkServer.start();randomNumClient.start();}}