本文主要對Flink 1.5中在network stack中引入的credit base模式進行說明。首先，會對flink 1.5之前的通信模型進行簡單的說明，然后在對Credit Based網(wǎng)絡(luò)通信模型進行說明，這樣更容易理解為什么需要Credit Based這樣的通信模型。
老的網(wǎng)絡(luò)通信模型
數(shù)據(jù)生產(chǎn)節(jié)點
- 數(shù)據(jù)寫入端將數(shù)據(jù)寫入到buffer中。
- 判斷當前channel是否可寫
- 判斷buffer中是否有數(shù)據(jù)，有則通過該channel發(fā)送到消費端。
數(shù)據(jù)消費節(jié)點
- 接受到數(shù)據(jù)包msg后，首先向LocalBufferPool申請內(nèi)存buffer。
- 申請成功則將msg的內(nèi)容拷貝到buffer中，并釋放msg。
- 如果申請內(nèi)存失敗，則監(jiān)聽LocalBufferPool中的內(nèi)存情況（這樣當有可用的內(nèi)存時，會被通知到），并將channel的autoRead屬性設(shè)置為false（導(dǎo)致發(fā)送的端的channel變?yōu)椴豢蓪憼顟B(tài)），同時將當前的msg放到staged的集合中（staged集合用來存儲那些申請內(nèi)存失敗的msg），當LocalBufferPool中的有可用內(nèi)存時會先處理staged中的msg集合，staged中的msg處理完畢后才會將channel的autoRead屬性設(shè)置為true。
缺點
可以看到在老的網(wǎng)絡(luò)模型中，數(shù)據(jù)生成節(jié)點只能通過檢查當前的channel是否可寫來決定自己是否要向消費端發(fā)送數(shù)據(jù)，它對下游數(shù)據(jù)消費端的真實容量情況一概不知。這就導(dǎo)致，當生成節(jié)點發(fā)現(xiàn)channel已經(jīng)不可寫的時候，有可能下游消費節(jié)點已經(jīng)積壓了很多數(shù)據(jù)（這些數(shù)據(jù)最終會被發(fā)在staged結(jié)合中）。
Credit Based網(wǎng)絡(luò)通信模型
為了解決老的通信模型中，生成節(jié)點對消費節(jié)點的真實容量情況一無所知而導(dǎo)致數(shù)據(jù)積壓的情況。Flink引入了Credit Based網(wǎng)絡(luò)通信模型。簡單的說，它的工作方式就是，在數(shù)據(jù)傳輸?shù)倪^程中。消費節(jié)點主動告知生成節(jié)點它的容量情況，也就是消費節(jié)點讓生成節(jié)點發(fā)多少數(shù)據(jù)，生成節(jié)點才發(fā)多少數(shù)據(jù)。
數(shù)據(jù)生產(chǎn)節(jié)點
數(shù)據(jù)寫出流程
- 數(shù)據(jù)寫入端將數(shù)據(jù)寫入到buffer中。
- 判斷當前的credit值是否大于0。
- 如果credit > 0，則將數(shù)據(jù)寫出，并更新credit值。數(shù)據(jù)寫出的時候會在msg上帶上當前生成端的數(shù)據(jù)量backlog。
- 如果credit <= 0，則不寫。
數(shù)據(jù)消費節(jié)點
- 節(jié)點接到msg包后，從msg包中拿出backlog值，更具該值判斷是否需要從LocalBufferPool中申請新的內(nèi)存塊用于接受數(shù)據(jù)。
- 如果backlong >= 消費端當前緩存的buffer數(shù)，則想向LocalBufferPool中申請新的內(nèi)存塊。然后再將新申請到的內(nèi)存塊作為credit值的增量通過Credit消息發(fā)送到數(shù)據(jù)生產(chǎn)端。
- 接著msg被接著處理，從消費端緩存的內(nèi)存塊中取出內(nèi)存塊buffer，并將msg拷貝到buffer中，釋放msg。
以上便是Credit Based網(wǎng)絡(luò)模型的大致過程，為了把主要的流程說清楚省略了很多的細節(jié)，包括Credit信息的同步、LocalBufferPool的內(nèi)存控制等。感興趣的同學(xué)看下下面這些相關(guān)的源碼，畢竟看代碼還是更加清楚些：
- Credit Based 數(shù)據(jù)發(fā)送端
CreditBasedPartitionRequestClientHandler.java - Credit Based 數(shù)據(jù)接收端
CreditBasedSequenceNumberingViewReader - 非Credit Based數(shù)據(jù)接受端
PartitionRequestClientHandler.java - LocalBufferPool相關(guān)
LocalBufferPool.java
