前言
筆者最近回顧自己對Flink技術(shù)棧細(xì)節(jié)的理解，發(fā)現(xiàn)對Flink的網(wǎng)絡(luò)棧、流控與反壓這一套機(jī)制存在比較大的盲區(qū)。雖然平時多次處理過作業(yè)反壓的問題，但是不完全理解背后的實現(xiàn)顯然說不過去。于是專門寫一篇總結(jié)，站在大佬們的肩膀上徹底搞清楚Flink是怎么做流控與處理反壓的。
Flink網(wǎng)絡(luò)傳輸?shù)臄?shù)據(jù)流向
Flink網(wǎng)絡(luò)傳輸?shù)臄?shù)據(jù)流向如下圖所示。
Sender在發(fā)送數(shù)據(jù)時，首先寫入TaskManager內(nèi)部的網(wǎng)絡(luò)緩存，利用Netty進(jìn)行傳輸——將待發(fā)送的數(shù)據(jù)存入Netty的ChannelOutboundBuffer，再經(jīng)由Socket的發(fā)送緩存發(fā)送出去。Receiver在接收數(shù)據(jù)時是反過來的，同樣要經(jīng)過3層緩存，即Socket接收緩存→Netty ChannelInboundBuffer→TaskManager網(wǎng)絡(luò)緩存。要實現(xiàn)流量控制，就是在上面的流程上做文章。
Flink的反壓傳播
反壓（back pressure）就是流式系統(tǒng)中關(guān)于處理能力的動態(tài)反饋機(jī)制，并且是從下游到上游的反饋。下圖示出數(shù)據(jù)流在Flink TaskManager之間流動的邏輯。
可見，一旦因為下游處理能力不足而出現(xiàn)反壓，反壓信號的傳播應(yīng)該分為兩個階段：一是從下游TaskManager的輸入端（InputGate）傳播到直接上游TaskManager的輸出端（ResultPartition）；二是在TaskManager內(nèi)部從輸出端傳播到輸入端。當(dāng)然，我們要重點考慮的是跨TaskManager的反壓傳播，因為它的鏈路比較長（參考上一節(jié)的數(shù)據(jù)流向圖），更有可能成為瓶頸。
下面先來介紹舊版本中的流控和反壓機(jī)制。
Flink 1.5之前：基于TCP的流控和反壓
在1.5版本之前，F(xiàn)link并沒有特別地去實現(xiàn)自己的流控機(jī)制，而是在傳輸層直接依靠TCP協(xié)議自身具備的滑動窗口機(jī)制（大學(xué)計算機(jī)網(wǎng)絡(luò)課程必講）。下面通過實例來復(fù)習(xí)TCP滑動窗口是如何實現(xiàn)流控的。
- 初始情況如下圖所示。Sender每單位時間發(fā)送3個包，發(fā)送窗口初始大小為3；Receiver每單位時間接收1個包，接收窗口初始大小為5（與接收緩存的大小相同）。
- Sender發(fā)送1~3三個包，Receiver接收到之后將它們放入緩存。
- Receiver消費一個包，接收窗口向前滑動一格，并告知Sender ACK=4（表示可以從第4個包開始發(fā)送），以及Window=3（表示接收窗口當(dāng)前的空余量為3）。
- Sender接收到ACK消息后發(fā)送4~6三個包，Receiver接收到之后將它們放入緩存。
- Receiver消費一個包，接收窗口向前滑動一格，并告知Sender ACK=7（表示可以從第7個包開始發(fā)送），以及Window=1（表示接收窗口當(dāng)前的空余量為1）。Sender接收到ACK消息之后，發(fā)現(xiàn)Receiver只能再接收1個包了，就將發(fā)送窗口的大小調(diào)整為1并發(fā)送包7，達(dá)到了限流的目的。
接著這個流程分析下去，可以得知Sender最終會無法發(fā)送數(shù)據(jù)（因為Receiver報告Window=0），直到Receiver消費掉緩存中的數(shù)據(jù)才能繼續(xù)發(fā)送。同時Sender還會定時向Receiver發(fā)送ZeroWindowProbe探測消息，保證Receiver能夠及時將消費能力報告給Sender。
接下來用實例介紹反壓流程。
- 如圖所示，Sender發(fā)送速度與Receiver接收速度的比是2:1，起初是可以正常發(fā)送與接收的。
- 一段時間過后，Receiver端InputChannel本身的緩存被耗盡，因此會向本地緩存池LocalBufferPool申請新的緩存。
- 一段時間過后，LocalBufferPool的可用額度會被耗盡，因此會向網(wǎng)絡(luò)緩存池NetworkBufferPool申請新的緩存。
- 隨著數(shù)據(jù)不斷積壓，NetworkBufferPool的額度也會被耗盡，此時沒有空間再接收新的數(shù)據(jù)，Netty的auto read會被關(guān)閉，不再從Socket緩存讀取數(shù)據(jù)。
- Socket緩存耗盡后，Receiver報告Window=0（參見上文的滑動窗口），Sender的Socket就會停止發(fā)送數(shù)據(jù)。
- Sender端的Socket緩存積壓，導(dǎo)致Netty無法再發(fā)送數(shù)據(jù)。
- 待發(fā)送的數(shù)據(jù)都積壓在Sender的ChannelOutboundBuffer中，當(dāng)數(shù)據(jù)量超過Netty的high watermark之后，Channel被置為不可寫，ResultSubPartition也就不再向Netty寫數(shù)據(jù)。
- Sender端的ResultSubPartition緩存滿了之后，就會像Receiver端的InputChannel一樣，不斷地向LocalBufferPool和NetworkBufferPool申請新的緩存，直到緩存全部耗盡，RecordWriter不能再寫數(shù)據(jù)。
這樣，我們就實現(xiàn)了反壓向上游TaskManager的傳遞。
Flink 1.5之后：基于Credit的流控和反壓
基于TCP的流控和反壓方案有兩大缺點：
- 只要TaskManager執(zhí)行的一個Task觸發(fā)反壓，該TaskManager與上游TaskManager的Socket就不能再傳輸數(shù)據(jù)，從而影響到所有其他正常的Task，以及Checkpoint Barrier的流動，可能造成作業(yè)雪崩；
- 反壓的傳播鏈路太長，且需要耗盡所有網(wǎng)絡(luò)緩存之后才能有效觸發(fā)，延遲比較大。
Flink 1.5+版本為了解決這兩個問題，引入了基于Credit的流控和反壓機(jī)制。它本質(zhì)上是將TCP的流控機(jī)制從傳輸層提升到了應(yīng)用層——即ResultPartition和InputGate的層級，從而避免在傳輸層造成阻塞。具體來講：
- Sender端的ResultSubPartition會統(tǒng)計累積的消息量（以緩存?zhèn)€數(shù)計），以backlog size的形式通知到Receiver端的InputChannel；
- Receiver端InputChannel會計算有多少空間能夠接收消息（同樣以緩存?zhèn)€數(shù)計），以credit的形式通知到Sender端的ResultSubPartition。
也就是說，Sender和Receiver通過互相告知對方自己的處理能力的方式來精準(zhǔn)地進(jìn)行流控（注意backlog size和credit也是要通過傳輸層的，不是直接交換的）。接下來仍然通過實例來說明基于Credit的流控和反壓流程。
- 仍然是Sender發(fā)送速度與Receiver接收速度的比是2:1的情景。Sender端的ResultSubPartition積壓了2個緩存的數(shù)據(jù)，因此會將該批次要發(fā)送的數(shù)據(jù)與backlog size = 2一同發(fā)往Receiver。
Receiver收到當(dāng)前批數(shù)據(jù)和backlog size之后，會計算InputChannel是否有足夠的緩存來接收下一批數(shù)據(jù)，如果不夠，則會去LocalBufferPool/NetworkBufferPool申請緩存，并將credit = 3通知到上游的ResultSubPartition，表示自己能夠接收3個緩存的消息。
- 隨著Receiver端的數(shù)據(jù)不斷積壓，網(wǎng)絡(luò)緩存最終被耗盡，因此會反饋給上游credit = 0（相當(dāng)于TCP滑動窗口中的window = 0），Sender端ResultPartition到Netty的鏈路會被阻斷。按照上一節(jié)所述的流程，Sender端的網(wǎng)絡(luò)緩存會被更快地耗盡，RecordWriter不能再寫數(shù)據(jù)，從而達(dá)到反壓的效果。
由上可知，反壓信號在TaskManager之間不需要再通過傳輸層隨著數(shù)據(jù)向上反饋，大大降低了反壓的延遲。并且也不會因為一個Task反壓而阻塞整個Socket鏈路，能夠相當(dāng)精確地在Task粒度控制流量，不僅輕量級，而且高效。
The End
筆者之前也寫過Spark Streaming基于PID的流控和反壓機(jī)制（傳送門），橫向?qū)Ρ纫幌?，不得不感嘆技術(shù)是永無止境的。
民那晚安晚安。
