Flink異步I/O

1 概述

流計算系統中常常須要與外部系統進行交互，咱們一般的作法如向數據庫發送用戶a的查詢請求，而後等待結果返回，在這以前，咱們的程序沒法發送用戶b的查詢請求。這是一種同步訪問方式，以下圖所示。

image

圖中棕色的長條表示等待時間，能夠發現網絡等待時間極大地阻礙了吞吐和延遲。爲了解決同步訪問的問題，異步模式能夠併發地處理多個請求和回覆。也就是說，你能夠連續地向數據庫發送用戶a、b、c等的請求，與此同時，哪一個請求的回覆先返回了就處理哪一個回覆，從而連續的請求之間不須要阻塞等待，如上圖右邊所示。這也正是 Async I/O 的實現原理。

2 Async I/O原理實現

以下官方示例代碼：

image

AsyncDataStream.(un)orderedWait 的主要工做就是建立了一個 AsyncWaitOperator。AsyncWaitOperator 是支持異步 IO 訪問的算子實現，該算子會運行 AsyncFunction 並處理異步返回的結果，其內部原理以下圖所示。

image

如圖所示，AsyncWaitOperator 主要由兩部分組成：StreamElementQueue 和 Emitter。StreamElementQueue 是一個 Promise 隊列，所謂 Promise 是一種異步抽象表示未來會有一個值，這個隊列是未完成的 Promise 隊列，也就是進行中的請求隊列。Emitter 是一個單獨的線程，負責發送消息（收到的異步回覆）給下游。

圖中E5表示進入該算子的第五個元素（」Element-5」），在執行過程當中首先會將其包裝成一個 「Promise」 P5，而後將P5放入隊列。最後調用 AsyncFunction 的 ayncInvoke 方法，該方法會向外部服務發起一個異步的請求，並註冊回調。該回調會在異步請求成功返回時調用 AsyncCollector.collect 方法將返回的結果交給框架處理。實際上 AsyncCollector 是一個 Promise ，也就是 P5，在調用 collect 的時候會標記 Promise 爲完成狀態，並通知 Emitter 線程有完成的消息能夠發送了。Emitter 就會從隊列中拉取完成的 Promise ，並從 Promise 中取出消息發送給下游。

3 消息順序性

3.1 有序

有序比較簡單，使用一個隊列就能實現。全部新進入該算子的元素（包括 watermark），都會包裝成 Promise 並按到達順序放入該隊列。以下圖所示，儘管P4的結果先返回，但並不會發送，只有 P1 （隊首）的結果返回了纔會觸發 Emitter 拉取隊首元素進行發送。

image

3.2 無序

使用兩個隊列就能實現，一個 uncompletedQueue 一個 completedQueue。全部新進入該算子的元素，一樣的包裝成 Promise 並放入 uncompletedQueue 隊列，當uncompletedQueue隊列中任意的Promise返回了數據，則將該 Promise 移到 completedQueue 隊列中，並通知 Emitter 消費。以下圖所示：

image

References

1.https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/stream/operators/asyncio.html