本次的聊法，還是要通過以kafka（source）->Flink,Flink(source)->Kafka來分別展開討論。
kafka是壹個具有數據保存、數據回放能力的消息隊列，說白了就是kafka中的每壹個數據，都有壹個專門的標記作為標識。而在Flink消費kafka傳入的數據的時候，source任務就能夠將這個偏移量以算子狀態的角色進行保存，寫入到設定好的檢查點中。這樣壹旦發生故障，Flink中的FlinkKafkaProduce連接器就i能夠按照自己保存的偏移量，自己去Kafka中重新拉取數據，也正是通過這種方式，就能夠確保Kafka到Flink之間的精準壹次性。
在上壹篇文章當中，已經表明了，如果想要讓輸出端能夠進行精準壹次性消費，就需要使用到冪等性或者是事務。而事務中的兩階段提交是所有方案裏面最好的實現。
其實Flink到Kafak之間也是采用了這種方式，具體的可以看壹下ctrl進到FlinkKafkaProduce連接器內部去看壹看：
這也就表明了，當數據通過Flink發送給sink端Kafka的時候，是經歷了兩個階段的處理的。第壹階段就是Flink向Kafka中插入數據，進入預提交階段。當JobManager發送的Ckeckpoint保存成功信號過來之後，才會提交事務進行正式的數據發送，也就是讓原來不可用的數據可以被使用了。
這個實現過程到目前階段就很清晰了，它的主體流程無非就是在開啟檢查點之後，由JobManager向各個階段的處理邏輯發送有關於檢查點的barrier。所有的計算任務接收到之後，就會根據自己當前的狀態做壹個檢查點保存。而當這個barrier來到sink任務的時候，sink就會開啟壹個事務，然後通過這個事務向外預寫數據。直到Jobmanager來告訴它這壹次的檢查點已經保存完成了，sink就會進行第二次提交，數據也就算是成功寫出了。
1.必須要保證檢查點被打開了，如果檢查點沒有打開，那麽之前說的壹切話都是空談。因為Flink默認檢查點是關著的。
2.在FlinkKafakProducer連接器的構造函數中要傳入參數，這個參數就是用來保證狀態壹致性的。就是在構造函數的最後壹個參數輸入如下：
3.配置Kafka讀取數據的隔離級別
在kafka中有個配置，這個配置用來管理Kafka讀取數據的級別。而這個配置默認是能夠讀取預提交階段的數據的，所以如果妳沒改這個配置，那兩階段提交的第壹階段就是白費了。所以需要改壹下這個配置，來更換壹下隔離級別：
4.事務超時時間
這個配置也很有意思，大家試想壹下。如果要進行兩階段提交，就要保證sink端支持事務，Kafka是支持事務的，但是像這個組件對於很多機制都有壹個超時時間的概念，也就是說如果時間到了這個界限還沒完成工作，那就會默認這個工作失敗。Kafka中由這個概念，Flink中同樣由這個概念。但是flink默認超時時間是1小時，而Kafka默認是15分鐘，這就有可能出現檢查點保存東西的時間大於15分鐘，假如說是16分鐘保存完成然後給sink發送檢查點保存陳功可以提交事務的信號，但是這個時候Kafka已經認為事務失敗，把之前的數據都扔了。那數據不就是丟失了麽。所以說Kafka的超時時間要大於Flink的超時時間才好。
截止到目前為止，基本上把有關於狀態維護的壹些東西都說完了，有狀態後端、有檢查點。還通過檢查點完成可端到端的數據精準壹次性消費。但是想到這我又感覺，如果有學習進度比我差壹些的，萬壹沒辦法很好的理解怎麽辦。所以在下壹篇文章當中我就聊聊Flink中的“狀態”到底是個什麽東西，都有什麽類型，都怎麽去用。