Flink 异步io

流程

异步io就是将一条条的记录同步与外部系统交互，变成并发的访问外部io。并不会将整个拓扑的次序打乱。

重要提示

ResultFuture在第一次通话时完成ResultFuture.complete。所有后续complete调用都将被忽略。

参数

超时：超时定义异步请求在被视为失败之前可能需要多长时间。此参数可防止死/失败的请求。
容量：此参数定义可能同时有多少异步请求正在进行中。尽管异步I / O方法通常会带来更好的吞吐量，但算子仍然可能成为流应用程序的瓶颈。限制并发请求的数量可确保算子不会累积不断增加的待处理请求积压，但一旦容量耗尽就会触发反压。

超时处理

当异步I / O请求超时时，默认情况下会引发异常并重新启动作业。如果要处理超时，可以覆盖该AsyncFunction#timeout方法。

结果顺序

AsyncDataStream 有两个静态方法，orderedWait 和 unorderedWait，对应了两种输出模式：有序和无序。

有序：消息的发送顺序与接受到的顺序相同（包括 watermark ），也就是先进先出。
无序：
在 ProcessingTime 的情况下，完全无序，先返回的结果先发送。
在 EventTime 的情况下，watermark 不能超越消息，消息也不能超越 watermark，也就是说 watermark 定义的顺序的边界。在两个 watermark 之间的消息的发送是无序的，但是在watermark之后的消息不能先于该watermark之前的消息发送。

public class AsyncDatabaseRequest extends RichAsyncFunction<PlayerCountEvent, PlayerCountEvent> {

  
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
        super.close();
        executorService.shutdown();
    }

    @Override
    public void asyncInvoke(PlayerCountEvent input, ResultFuture<PlayerCountEvent> resultFuture) throws Exception {
        
        CompletableFuture.supplyAsync(new Supplier<PlayerCountEvent>() {
            @Override
            public PlayerCountEvent get() {
                try {
                   // do something in the thread 
                   return input;
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
                return null;
            }

        }).thenAccept(new Consumer<PlayerCountEvent>() {
            @Override
            public void accept(PlayerCountEvent playerCountEvent) {
                resultFuture.complete(Collections.singleton(playerCountEvent));
            }
        });


    }
}