本頁闡述了使用Flink的API來進行外部數(shù)據(jù)存儲的異步I/O，對于不熟悉異步或者事件驅動編程的用戶，一篇關于Future和事件驅動編程可能會很有用瑰抵。

注意:關于異步I/O的詳細設計和實現(xiàn)可以在異步I/O設計和實現(xiàn)這篇文章找到。

異步I/O操作的需要

當與外部系統(tǒng)進行交互(例如使用存儲在數(shù)據(jù)庫中的數(shù)據(jù)豐富流事件)時, 需要注意的是, 與外部系統(tǒng)的通信延遲并不決定流應用程序的總體工作。

原始的訪問外部系統(tǒng)中的數(shù)據(jù)局荚，例如通過一個MapFunction來訪問，通常意味著同步交互:將一個請求發(fā)送到數(shù)據(jù)庫，MapFunction等待直到接收到響應為止耀态。很多情況下轮傍，這種等待會占用很大一部分函數(shù)的時間。

與外部數(shù)據(jù)庫系統(tǒng)進行異步交互意味著一個并行函數(shù)實例可以并發(fā)地處理多個請求和并發(fā)地接收多個響應首装。那樣的話创夜，等待時間就可以被其他的請求或者響應所覆蓋。至少仙逻，等待時間可以被多個請求攤銷驰吓，這在很多情況下會導致更高的流吞吐量。

前提

如上節(jié)所述瞬内，實現(xiàn)一個連接數(shù)據(jù)庫(或者key/value存儲系統(tǒng))的正確異步I/O需要一個客戶端迷雪，數(shù)據(jù)庫支持通過該客戶端來進行異步請求。許多流行的數(shù)據(jù)庫都支持這種客戶端虫蝶。

對于沒有這種客戶端的情況下章咧，用戶可以將異步客戶端換成一個可以通過創(chuàng)建多個客戶端并使用線程池處理同步調用來嘗試將同步客戶端轉換為有限的并發(fā)客戶端。然而能真，這個方法通常比純粹的異步客戶端性能要低一些赁严。

異步I/O API

Flink的Async I/O允許用戶在數(shù)據(jù)流中使用異步的請求客戶端，這個API會處理與數(shù)據(jù)流的交互粉铐，同時還處理順序疼约、事件時間、容錯等蝙泼。

假設已經(jīng)目標數(shù)據(jù)庫已經(jīng)有了異步客戶端程剥，要實現(xiàn)一個通過異步I/O來操作數(shù)據(jù)庫還需要三個步驟:
　　1、實現(xiàn)用來分發(fā)請求的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時就完成了溪胶，所有后續(xù)的collect調用都會被忽略搂擦。

下面的兩個參數(shù)控制了異步操作:
　　****Timeout****:timeout定義了異步操作過了多長時間后會被丟棄，這個參數(shù)是防止了死的或者失敗的請求
　　****Capacity****:這個參數(shù)定義了可以同時處理多少個異步請求载荔，雖然異步I/O方法會帶來更好的吞吐量盾饮，但是算子任然會成為流應用的瓶頸。限制并發(fā)請求的數(shù)量確保了算子不會積累不斷增加的積壓的待處理請求，但一旦容量耗盡丘损，它將觸發(fā)背壓普办。

結果順序

由AsyncFunction發(fā)出的并發(fā)請求經(jīng)常是以無序的形式完成，取決于哪個請求先完成徘钥。為了控制發(fā)出請求結果的順序衔蹲，F(xiàn)link提供了兩種模式:
　　****Unordered****:結果記錄在異步請求完成后就發(fā)出，流中的記錄的順序通過異步I/O操作后會與先前的不一致呈础。當使用處理時間作為時間特性時這種模式具有低延遲舆驶、低消耗特點。通過AsyncDataStream.unorderedWait(...)來使用這種模式而钞。
　　****Ordered****:在這種情況下沙廉，流的順序是保留的，結果記錄發(fā)出的順利與異步請求觸發(fā)的順序(算子輸入記錄的順序)一致臼节。為了實現(xiàn)這一點撬陵，算子會將結果記錄緩存起來直到所有的處理記錄都被發(fā)出(或者超時)為止。這常常會導致一定程度的延遲和checkpoint消耗网缝，因為跟非排序模式相比巨税，記錄或者結果會被長時間保存在checkpoint State中。通過AsyncDataStream.orderedWait(...)來使用這種模式粉臊。

事件時間

當使用流程序使用事件時間時草添，異步I/O操作將正確處理水印，這具體說明了如下兩種模式:
　　****Unordered****:水印不會超過記錄反之亦然扼仲，這也就意味著水印建立起了一個秩序邊界远寸。記錄在兩個水印間無序地發(fā)出。在一個水印后產生的記錄只能在這個水印發(fā)出之后才能發(fā)出屠凶，同樣水印也只能在所有水印之前的記錄都發(fā)出之后才能發(fā)出而晒。
　　****Ordered****:保存水印的順序，就如保存記錄之間的順序一樣阅畴。與處理時間相比，開銷沒有顯著變化迅耘。
請記住贱枣，攝入時間是一個特殊的事件時間，會基于源處理時間的自動產生水印颤专。

容錯性保證

異步I/O操作提供了exactly-once容錯性保證纽哥，它將異步請求的記錄存儲在checkpoint中，并在從故障中恢復時恢復/重新觸發(fā)請求栖秕。

實施提示

警告