flink异步iomysql缓存_Flink异步IO实战

基本概念

首先通过官网的一个图片了解一下Asynchronous I/O Operation

Flink source收到一条数据就会进行处理&＃xff0c;如果需要通过这条数据关联外部数据源&＃xff0c;例如mysql&＃xff0c;在发出查询请求后&＃xff0c;同步IO的方式是会等待查询结果再处理下一条数据的查询&＃xff0c;也就是每一条数据都要等待上一个查询结束。而异步IO是指数据来了以后发出查询请求&＃xff0c;先不等查询结果&＃xff0c;直接继续发送下一条的查询请求&＃xff0c;对于查询结果是异步返回的&＃xff0c;返回结果之后再进入下一个算子的计算。这两种方式性能差距请看下的样例。

样例

代码传送门

生成6条数据&＃xff0c;从0开始递增的6个数字。模拟异步查询之后&＃xff0c;加上时间戳输出

public class AsyncIODemo {

public static void main(String[] args) throws Exception {

StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment();

env.setParallelism(1);

final int maxCount &＃61; 6;

final int taskNum &＃61; 1;

final long timeout &＃61; 40000;

DataStream inputStream &＃61; env.addSource(new SimpleSource(maxCount));

AsyncFunction function &＃61; new SampleAsyncFunction();

DataStream result &＃61; AsyncDataStream.unorderedWait(

inputStream,

function,

timeout,

TimeUnit.MILLISECONDS,

10).setParallelism(taskNum);

result.map(new MapFunction() {

&＃64;Override

public String map(String value) throws Exception {

return value &＃43; "," &＃43; System.currentTimeMillis();

}

}).print();

env.execute("Async IO Demo");

}

private static class SimpleSource implements SourceFunction {

private volatile boolean isRunning &＃61; true;

private int counter &＃61; 0;

private int start &＃61; 0;

public SimpleSource(int maxNum) {

this.counter &＃61; maxNum;

}

&＃64;Override

public void run(SourceContext ctx) throws Exception {

while ((start

synchronized (ctx.getCheckpointLock()) {

System.out.println("send data:" &＃43; start);

ctx.collect(start);

&＃43;&＃43;start;

}

Thread.sleep(10L);

}

}

&＃64;Override

public void cancel() {

isRunning &＃61; false;

}

}

}

异步方法

public class SampleAsyncFunction extends RichAsyncFunction {

private long[] sleep &＃61; {100L, 1000L, 5000L, 2000L, 6000L, 100L};

&＃64;Override

public void open(Configuration parameters) throws Exception {

super.open(parameters);

}

&＃64;Override

public void close() throws Exception {

super.close();

}

&＃64;Override

public void asyncInvoke(final Integer input, final ResultFuture resultFuture) {

System.out.println(System.currentTimeMillis() &＃43; "-input:" &＃43; input &＃43; " will sleep " &＃43; sleep[input] &＃43; " ms");

query(input, resultFuture);

}

private void query(final Integer input, final ResultFuture resultFuture) {

try {

Thread.sleep(sleep[input]);

resultFuture.complete(Collections.singletonList(String.valueOf(input)));

} catch (InterruptedException e) {

resultFuture.complete(new ArrayList<>(0));

}

}

private void asyncQuery(final Integer input, final ResultFuture resultFuture) {

CompletableFuture.supplyAsync(new Supplier() {

&＃64;Override

public Integer get() {

try {

Thread.sleep(sleep[input]);

return input;

} catch (Exception e) {

return null;

}

}

}).thenAccept((Integer dbResult) -> {

resultFuture.complete(Collections.singleton(String.valueOf(dbResult)));

});

}

}

上面的代码中有两个方法query()和asyncQuery()&＃xff0c;其中Thread.sleep(sleep[input]);用来模拟查询需要等待的时间&＃xff0c;每条数据等待的时间分别为100L, 1000L, 5000L, 2000L, 6000L, 100L毫秒。

结果分析

运行query()的结果为

send data:0

send data:1

send data:2

send data:3

send data:4

send data:5

1577801193230-input:0 will sleep 100 ms

1577801193331-input:1 will sleep 1000 ms

0,1577801194336

1,1577801194336

1577801194336-input:2 will sleep 5000 ms

1577801199339-input:3 will sleep 2000 ms

2,1577801201341

1577801201342-input:4 will sleep 6000 ms

3,1577801207345

4,1577801207345

1577801207346-input:5 will sleep 100 ms

5,1577801207451

可以看到第一条数据进入到map算子的时间与最后一条相差了13115毫秒&＃xff0c;执行的顺序与source中数据的顺序一致&＃xff0c;并且是串行的。

运行asyncQuery()的结果为

send data:0

send data:1

send data:2

send data:3

1577802161755-input:0 will sleep 100 ms

1577802161756-input:1 will sleep 1000 ms

1577802161757-input:2 will sleep 5000 ms

send data:4

send data:5

1577802161783-input:3 will sleep 2000 ms

1577802161784-input:4 will sleep 6000 ms

1577802161785-input:5 will sleep 100 ms

0,1577802161859

1,1577802162759

3,1577802163862

5,1577802163962

2,1577802166760

4,1577802168762

同样第一条数据进入map算子的时间与最后一条仅相差了6903毫秒&＃xff0c;而且输出结果的顺序并不是source中的顺序&＃xff0c;而是按照查询时间递增的顺序输出&＃xff0c;并且查询请求几乎是同一时间发出的。

通过上面的例子可以看出&＃xff0c;flink所谓的异步IO&＃xff0c;并不是只要实现了asyncInvoke方法就是异步了&＃xff0c;这个方法并不是异步的&＃xff0c;而是要依靠这个方法里面所写的查询是异步的才可以。否则像是上面query()方法那样&＃xff0c;同样会阻塞查询相当于同步IO。在实现flink异步IO的时候一定要注意。官方文档也给出了相关的说明。

For example, the following patterns result in a blocking asyncInvoke(...) functions and thus void the asynchronous behavior:Using a database client whose lookup/query method call blocks until the result has been received back

总结

本文基于flink 1.9.0。通过样例介绍了如何实现flink异步IO&＃xff0c;读者可以修改本文样例体验异步IO其他的特性&＃xff0c;例如Order of Results或者Event Time。