在南海建设工程交易中心网站宁波优化

前言

引言

数据流API

基于POJO的数据流

一般来说flink中的源数据我们都会以简单java对象即pojo(Plain Ordinary Java Object )的形式进行传输或游走，只要满足以下条件，flink就会识别这些数据类型：

1. 类中所有非静态、非transient修饰的字段，要么以public且非final修饰或者对外提供get和set方法
2. 该类不存在非静态的内部类
3. 提供无参构造函数

对应的我们给出日常比较常用的POJO 示例，即私有成员但是提供get、set符合上述的要求：

public class Person {private String name;private Integer age;//提供无参构造函数public Person() {}public Person(String name, Integer age) {this.name = name;this.age = age;}//......//get set方法
}

以上述POJO作为源数据，可以看到笔者通过StreamExecutionEnvironment 构建流的执行环境，并通过fromData进行关联：

	final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//使用 fromData 关联源数据DataStreamSource<Person> source = env.fromData(new Person("Alice", 18),new Person("Bob", 28),new Person("Charlie", 32));

基于上述的源数据利用DataStream api尝试过滤出18岁以上的person数据并将过滤结果打印输出：

//基于  filter过滤出大于18岁的personSingleOutputStreamOperator<Person> filterRes = source.filter(person -> person.getAge() > 18);//输出打印filterRes.print();

flink中的流操作和lambda类似需要有一个终端操作才能启动运行，所以我们再完成上述的执行环境设置之后，需确保通过 env.execute();将当前job提交到JobManager ，JobManager 切割为无数个子并行任务分发到指定的Task Managers 的slot槽中等待运行：

  //执行execute后，上述任务提交到JobManager中的taskmanager某个slot中等待执行，若没提交这个则不会execute执行，这一点和java lambda的终端流操作思想一致env.execute();

需要补充的是，flink的fromData方法提供了多种的重载，上面的示例我们也可以通过List的方式将源数据传入：

List<Person> list = Arrays.asList(new Person("Alice", 18),new Person("Bob", 28),new Person("Charlie", 32));//使用 fromData 关联源数据DataStreamSource<Person> source = env.fromData(list);

基本源流配置示例

上述的示例我们基于DataStream的fromData方法来构建一些简单源流，实际上flink支持在配置直接直接指明文件流或者socket流，因为socket流相对于物理文件流更常用，所以我们给出一个采集本地9999端口的socket流示例：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<String> dataStream = env.socketTextStream("localhost", 9999);

因为我们本案例发送的数据格式为hello,序列化的person对象的json字符串，所以收到数据流之后需要对数据进行提取转换，所以我们还是通过map和filter完成映射转换和过滤：

dataStream.map(s -> {String jsonStr = s.substring(s.indexOf(",")+1);Person person = JSONUtil.toBean(jsonStr, Person.class);return person;}).filter(p -> p.getAge() > 18).print();env.execute();

为了方便测试，笔者这里给出个人服务端socket代码使用示例，当然读者也可以在自己的系统上使用nc示例完成：

public static void main(String[] args) {try {// 1. 创建ServerSocket，监听9999端口ServerSocket serverSocket = new ServerSocket(9999);System.out.println("服务器启动，等待客户端连接...");// 2. 接受客户端连接Socket clientSocket = serverSocket.accept();System.out.println("客户端已连接: " + clientSocket.getInetAddress());// 3. 获取输出流，用于向客户端发送数据PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), true);// 4. 每隔1秒发送"hello"while (true) {Person person = new Person(RandomUtil.randomString(3), RandomUtil.randomInt(35));out.println("hello," + JSONUtil.toJsonStr(person));System.out.println("服务器发送: hello " + JSONUtil.toJsonStr(person));Thread.sleep(1000); // 暂停1秒}// 注意：这里为了简化代码，没有关闭资源，实际应用应该添加try-catch-finally} catch (Exception e) {e.printStackTrace();}}

可以看到转换和实际收到的数据流结果如下：

2> {"name":"Qex","age":27}
7> {"name":"nH7","age":25}
14> {"name":"zmN","age":34}

在实际的应用中这种配置方式常用于那些高吞吐、低延迟的数据源，例如Kafka这样的消息中间件，这一点flink也提供和上述一样方便的操作API。

基本流接收器

上文过滤出成年person的例子中我们在完成filter过滤后调用print方法进行打印输出，实际上其原理本质上就是为这个源流添加一个以打印输出的sink，这一点我们可以查看DataStream的print方法源码知晓：

@PublicEvolvingpublic DataStreamSink<T> print() {PrintSinkFunction<T> printFunction = new PrintSinkFunction<>();return addSink(printFunction).name("Print to Std. Out");}

同时我们也需要说明在输出结果前面类似于14>、7>代表当前输出是由哪个并行流线程(子任务)执行。

当然关于接收器我们也可以基于源数据类型进行自定义，例如下面这段代码，笔者指明源数据为person希望按照我们预期的方式打印，可通过创建一个SinkFunction指明person泛型重写invoke实现自定义输出逻辑：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();List<Person> list = Arrays.asList(new Person("Alice", 18),new Person("Bob", 28),new Person("Charlie", 32));//使用 fromData 关联源数据DataStreamSource<Person> source = env.fromData(list);//添加一个person的s