Flink的異步算子的原理及使用( 五 ) _生活百科

我們可以從上面的ResultHandler類的complete方法開始看，具體可以參見上面注釋，總結(jié)起來就是如下幾步

取消當(dāng)前ResultHandler對象的超時(shí)定時(shí)任務(wù)
調(diào)用StreamRecordQueueEntry的complete方法將成員變量completedElements賦值
判斷OrderedStreamElementQueue隊(duì)列的隊(duì)頭元素StreamRecordQueueEntry的completedElements成員變量是不是不為空
第3步不為空，則調(diào)用OrderedStreamElementQueue隊(duì)列的emitCompletedElement方法移除隊(duì)列的頭元素StreamElementQueueEntry并調(diào)用emitResult方法真正向下游吐出數(shù)據(jù)

從上面可以看出每次隨著completableFuture的complete方法的調(diào)用，都會(huì)判斷隊(duì)頭的元素是否處理完，處理完就移除隊(duì)頭元素并向吐出數(shù)據(jù) 。所以異步算子每次來數(shù)據(jù)經(jīng)過processElement方法，就已經(jīng)將數(shù)據(jù)元素封裝成StreamElementQueueEntry對象并放到了隊(duì)列中，雖然異步算子執(zhí)行過程是異步，每個(gè)元素的完成時(shí)間沒有順序，但是由于每個(gè)元素完成后，都是判斷的隊(duì)頭元素有沒有完成，完成后也是移除隊(duì)頭并向下游吐數(shù)據(jù) 。所以整體過程還是按照processElement處理順序也就是上游給過來的數(shù)據(jù)順序嚴(yán)格有序的。
5、flinksql自定義AsyncLookupFunction通常flinksql使用外部的數(shù)據(jù)源都需要引入一個(gè)flinksql-connector-xx這種jar包，比如我們想以kafka為流表join一個(gè)redis的維表，那么這時(shí)候查詢r(jià)edis的維表，通常使用的就是lookup join 。但是網(wǎng)上提供的例子基本都是同步的lookup join，在有些場景下為了提高吞吐就需要使用異步的lookup join 。詳細(xì)實(shí)現(xiàn)可以直接看代碼：https://gitee.com/rongdi/flinksql-connector-redis