异步I/O操作的需要

当与外部系统进行交互(例如使用存储在数据库中的数据丰富流事件)时, 需要注意的是, 与外部系统的通信延迟并不决定流应用程序的总体工作。

原始的访问外部系统中的数据，例如通过一个MapFunction来访问，通常意味着同步交互:将一个请求发送到数据库，MapFunction等待直到接收到响应为止。很多情况下，这种等待会占用很大一部分函数的时间。

与外部数据库系统进行异步交互意味着一个并行函数实例可以并发地处理多个请求和并发地接收多个响应。那样的话，等待时间就可以被其他的请求或者响应所覆盖。至少，等待时间可以被多个请求摊销，这在很多情况下会导致更高的流吞吐量。

注意:通过扩展MapFunction到一个很高的并发度来提高吞吐量在一定程度上是可行的，但是常常会导致很高的资源消耗:有很多的并行MapFunction实例意味着更多的任务、线程、Flink内部网络连接、与数据库之间的网络连接、缓存以及通常的内部开销。

前提

如上节所述，实现一个连接数据库(或者key/value存储系统)的正确异步I/O需要一个客户端，数据库支持通过该客户端来进行异步请求。许多流行的数据库都支持这种客户端。

对于没有这种客户端的情况下，用户可以将异步客户端换成一个可以通过创建多个客户端并使用线程池处理同步调用来尝试将同步客户端转换为有限的并发客户端。然而，这个方法通常比纯粹的异步客户端性能要低一些。

异步I/O API

Flink的Async I/O允许用户在数据流中使用异步的请求客户端，这个API会处理与数据流的交互，同时还处理顺序、事件时间、容错等。

假设已经目标数据库已经有了异步客户端，要实现一个通过异步I/O来操作数据库还需要三个步骤:
　　1、实现用来分发请求的AsyncFunction
　　2、获取操作结果的callback，并将它提交到AsyncCollector中
　　3、将异步I/O操作作为转换操作应用于DataStream中

下面代码展示了基本的模式:

// This example implements the asynchronous request and callback with Futures that have the
// interface of Java 8's futures (which is the same one followed by Flink's Future)

/**
 * An implementation of the 'AsyncFunction' that sends requests and sets the callback.
 */
class AsyncDatabaseRequest extends RichAsyncFunction<String, Tuple2<String, String>> {

    /** The database specific client that can issue concurrent requests with callbacks */
    private transient DatabaseClient client;

    @Override
    public void open(Configuration parameters) throws Exception {
        client = new DatabaseClient(host, post, credentials);
    }

    @Override
    public void close() throws Exception {
        client.close();
    }

    @Override
    public void asyncInvoke(final String str, final AsyncCollector<Tuple2<String, String>> asyncCollector) throws Exception {

        // issue the asynchronous request, receive a future for result
        Future<String> resultFuture = client.query(str);

        // set the callback to be executed once the request by the client is complete
        // the callback simply forwards the result to the collector
        resultFuture.thenAccept( (String result) -> {

            asyncCollector.collect(Collections.singleton(new Tuple2<>(str, result)));
         
        });
    }
}

// create the original stream
DataStream<String> stream = ...;

// apply the async I/O transformation
DataStream<Tuple2<String, String>> resultStream =
    AsyncDataStream.unorderedWait(stream, new AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100);
/**
 * An implementation of the 'AsyncFunction' that sends requests and sets the callback.
 */
class AsyncDatabaseRequest extends AsyncFunction[String, (String, String)] {

    /** The database specific client that can issue concurrent requests with callbacks */
    lazy val client: DatabaseClient = new DatabaseClient(host, post, credentials)

    /** The context used for the future callbacks */
    implicit lazy val executor: ExecutionContext = ExecutionContext.fromExecutor(Executors.directExecutor())


    override def asyncInvoke(str: String, asyncCollector: AsyncCollector[(String, String)]): Unit = {

        // issue the asynchronous request, receive a future for the result
        val resultFuture: Future[String] = client.query(str)

        // set the callback to be executed once the request by the client is complete
        // the callback simply forwards the result to the collector
        resultFuture.onSuccess {
            case result: String => asyncCollector.collect(Iterable((str, result)));
        }
    }
}

// create the original stream
val stream: DataStream[String] = ...

// apply the async I/O transformation
val resultStream: DataStream[(String, String)] =
    AsyncDataStream.unorderedWait(stream, new AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100)

重要提醒:AsyncCollector在第一次调用AsyncCollector.collect时就完成了，所有后续的collect调用都会被忽略。

下面的两个参数控制了异步操作:
　　****Timeout****:timeout定义了异步操作过了多长时间后会被丢弃，这个参数是防止了死的或者失败的请求
　　****Capacity****:这个参数定义了可以同时处理多少个异步请求，虽然异步I/O方法会带来更好的吞吐量，但是算子任然会成为流应用的瓶颈。限制并发请求的数量确保了算子不会积累不断增加的积压的待处理请求，但一旦容量耗尽，它将触发背压。

结果顺序

由AsyncFunction发出的并发请求经常是以无序的形式完成，取决于哪个请求先完成。为了控制发出请求结果的顺序，Flink提供了两种模式:
　　****Unordered****:结果记录在异步请求完成后就发出，流中的记录的顺序通过异步I/O操作后会与先前的不一致。当使用处理时间作为时间特性时这种模式具有低延迟、低消耗特点。通过AsyncDataStream.unorderedWait(...)来使用这种模式。
　　****Ordered****:在这种情况下，流的顺序是保留的，结果记录发出的顺利与异步请求触发的顺序(算子输入记录的顺序)一致。为了实现这一点，算子会将结果记录缓存起来直到所有的处理记录都被发出(或者超时)为止。这常常会导致一定程度的延迟和checkpoint消耗，因为跟非排序模式相比，记录或者结果会被长时间保存在checkpoint State中。通过AsyncDataStream.orderedWait(...)来使用这种模式。

事件时间

当使用流程序使用事件时间时，异步I/O操作将正确处理水印，这具体说明了如下两种模式:
　　****Unordered****:水印不会超过记录反之亦然，这也就意味着水印建立起了一个秩序边界。记录在两个水印间无序地发出。在一个水印后产生的记录只能在这个水印发出之后才能发出，同样水印也只能在所有水印之前的记录都发出之后才能发出。
　　****Ordered****:保存水印的顺序，就如保存记录之间的顺序一样。与处理时间相比，开销没有显著变化。
请记住，摄入时间是一个特殊的事件时间，会基于源处理时间的自动产生水印。