Flink實作異步IO實戰
基本概念
首先通過官網的一個圖檔了解一下Asynchronous I/O
Flink source收到一條資料就會進行處理,如果需要通過這條資料關聯外部資料源,例如mysql,在發出查詢請求後,同步IO的方式是會等待查詢結果再處理下一條資料的查詢,也就是每一條資料都要等待上一個查詢結束。而異步IO是指資料來了以後發出查詢請求,先不等查詢結果,直接繼續發送下一條的查詢請求,對于查詢結果是異步傳回的,傳回結果之後再進入下一個算子的計算。這兩種方式性能差距請看下的樣例。
案例
生成6條資料,從0開始遞增的6個數字。模拟異步查詢之後,加上時間戳輸出。
public class AsyncIODemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
final int maxCount = 6;
final int taskNum = 1;
final long timeout = 40000;
DataStream<Integer> inputStream = env.addSource(new SimpleSource(maxCount));
AsyncFunction<Integer, String> function = new SampleAsyncFunction();
DataStream<String> result = AsyncDataStream.unorderedWait(
inputStream,
function,
timeout,
TimeUnit.MILLISECONDS,
10).setParallelism(taskNum);
result.map(new MapFunction<String, String>() {
@Override
public String map(String value) throws Exception {
return value + "," + System.currentTimeMillis();
}
}).print();
env.execute("Async IO Demo");
}
private static class SimpleSource implements SourceFunction<Integer> {
private volatile boolean isRunning = true;
private int counter = 0;
private int start = 0;
public SimpleSource(int maxNum) {
this.counter = maxNum;
}
@Override
public void run(SourceContext<Integer> ctx) throws Exception {
while ((start < counter || counter == -1) && isRunning) {
synchronized (ctx.getCheckpointLock()) {
System.out.println("send data:" + start);
ctx.collect(start);
++start;
}
Thread.sleep(10L);
}
}
@Override
public void cancel() {
isRunning = false;
}
}
}
異步方法
代碼如下(示例):
public class SampleAsyncFunction extends RichAsyncFunction<Integer, String> {
private long[] sleep = {100L, 1000L, 5000L, 2000L, 6000L, 100L};
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
}
@Override
public void close() throws Exception {
super.close();
}
@Override
public void asyncInvoke(final Integer input, final ResultFuture<String> resultFuture) {
System.out.println(System.currentTimeMillis() + "-input:" + input + " will sleep " + sleep[input] + " ms");
query(input, resultFuture);
}
private void query(final Integer input, final ResultFuture<String> 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<String> resultFuture) {
CompletableFuture.supplyAsync(new Supplier<Integer>() {
@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(),其中Thread.sleep(sleep[input]);用來模拟查詢需要等待的時間,每條資料等待的時間分别為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毫秒,執行的順序與source中資料的順序一緻,并且是串行的。
運作asyncQuery()的結果為:
```xml
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毫秒,而且輸出結果的順序并不是source中的順序,而是按照查詢時間遞增的順序輸出,并且查詢請求幾乎是同一時間發出的。
通過上面的例子可以看出,flink所謂的異步IO,并不是隻要實作了asyncInvoke方法就是異步了,這個方法并不是異步的,而是要依靠這個方法裡面所寫的查詢是異步的才可以。否則像是上面query()方法那樣,同樣會阻塞查詢相當于同步IO。在實作flink異步IO的時候一定要注意。官方文檔也給出了相關的說明。
2021年2月4日創作
軍鷹