kafka同步數據機制

一、名詞解釋

1. 基本名詞

• 主題：topic。邏輯概念，用於區分業務
• 服務器：broker。接收持久化消息，管理topic，權限管理，consumer重平衡等等
• 分區：partition。一個有序的消息序列，一個topic對應多個分區
• 消息：record。不作解釋
• 消息位移：Offset。分區中每條消息的位置
• 生產者：producer。不作解釋~~~~
• 消費者：consumer。不作解釋
  

2.副本與ISR

這兩個是broker裏面的兩個概念名詞。

所謂_副本_，就是kafka爲了提升可用性，將同一個parition數據分散到不一樣broker的數據備份。同時kafka會選出一個leader副本用於對外讀，follower則須要主動向leader副本請求同步kafka日誌，以保證主從副本數據保持一致。

所謂_ISR_，就是有資格能被評選爲leader副本的follower副本集合。只有leader和ISR中全部副本都同步狀態了，纔會被kafka認爲該消息已經提交。

三、副本中的概念

起始位移（ base offset ）：該副本第一條消息的位移

高水印值（ high watermark, HW ）：該副本最新一條_已經提交_的消息位移。_若是某個消息的offset小於該值，則全部的副本都已經同步這條消息了；若是某個消息的offset大於該值，則肯說明些副本還沒同步到這條消息_。換一種說法，全部offset小於該值的消息對consumer是可見的；而大於該值的是不可見的。該值很是重要，他的值影響到副本主從同步的位置，影響到consumer消費數據的位置。應該注意的是，HW不止在leader裏存在，在follower裏也存在，其緣由就是爲了防止leader崩潰，follower也能當即頂替leader進行正常工做（最終一致性）。

日誌末端位移 （Clog end offset, LEO ）：該副本最後一條信息的位移

二、消息同步流程

一、總體流程

①broker1上的 leader 副本接收到消息，把本身的 LEO 值更新爲 1 。

②broker2 和 broker3 上的 follower 副本各自發送請求給 broker 1。（通常狀況下，是follower定時給leader發送fetch請求的，就好像heartbeat心跳）

③broker1收到fetch請求後，主動分別把該消息推送給 follower 副本 。

④follower 副本接收到消息後各自更新本身的 LEO 爲 1，並返回response。

⑤leader 副本接收到其餘 follower 副本的數據請求響應（ response ）以後，更新 HW 值爲 1 。 此時位移爲 0 的這條消息能夠被 consumer 消費。

二、流程細節

_原則：HW是當前已知的全部LEO的最小值_。爲何呢？正常狀況下，各個broker的partition數據都是順序寫入的，最小的LEO意味着全部的副本都同步到了這個LEO之前的全部數據，就知足了「HW以前的消息都已經同步完成」的要求。

爲了便於描述，咱們假設有兩個副本在同步數據，一個leader一個follower

• 第一輪fetch

在某一時刻，leader收到了一條信息，寫入了底層數據，接下來就是數據同步的過程了。

一、leader的LEO +1，好理解，有了一條信息，尾數須要加一。

二、leader嘗試更新HW，取全部副本LEO最小值，本例是0。那麼，哪裏獲取各個副本最小值呢？leader副本本地有個地方專門負責緩存這個數據，其餘follower經過fetch請求告知leader

這個時候，follower發送了fetch（fetch請求裏會帶着本身如今的LEO，如今是0），leader收到了fetch

三、leader收到了fetch，嘗試更新HW。全部副本LEO最小值。本身的是1，fetch裏是0，那麼是0。

四、獲取offset ＞follower LEO的數據放到response裏，同時將本身的HW(注意！！此時是全局LEO最小值)放到response裏，本例裏是0

五、follower接受到了response，將數據寫入，同時更新LEO，本例裏+1

六、follower嘗試更新HW，是全局LEO最小值，比較response裏的HW和本身的LEO取最小值便可（上面紅字的特性，這裏就用到了）。本例裏更新後是0

此時第一輪fetch結束，應該注意到，第一輪fetch完成後，數據雖然同步過去了，可是還不可見，由於leader此時還不知道follower是否是同步成功了

• 第二輪fetch

一、follower發送了fetch請求，攜帶本身的LEO=1

二、leader嘗試更新HW，全局LEO的最小值，所以是1

三、獲取offset ＞follower LEO的數據放到response裏，此次沒有數據，同時將本身的HW=1放到response裏

四、follower收到信息，沒數據寫入，而後嘗試更新本身的HW，全局最小值，本例HW=1

此時第二輪fetch結束，此時此刻，數據同步才真正結束，這條新數據對外可見了

三、數據丟失問題

綜合上面的論述，通過兩輪fetch過程後纔會對外可見。這個時間差就容易致使數據丟失或者不一致的問題

場景一：兩輪fetch中間發生follwer和leader前後崩潰，前提：leader寫入完成即認爲已提交

如圖，假如第二輪fetch發生，A已經更新了HW，可是尚未包裝response返回給B，此時B發生了崩潰。重啓後的B會將LEO調整成崩潰前的HW值，那麼後面的數據就被刪除了（看，此時出現了一次數據不一致）。這個時候B想要向A發起fetch，若是這個時候剛好A掛掉了，B被選爲leader，A重啓回來後就會fetch取leader的HW和本身的LEO比較取最小值，最後獲得HW=1，這樣原來HW=2的數據就永久丟失了。

場景二：兩輪fetch中，follower和leader同時崩潰

仍是上面那種狀況，第二輪fetch發生，A已經更新了HW，可是尚未包裝response返回給B。這個時候leader和follower同時崩潰，而後B先重啓成爲leader了。這個時候，producer發送了一個消息記錄到B，此時由於沒有follower，所以直接更新HW到2。然後A回來成爲follower，這時，A發現本身的HW和B的HW相等，所以不作變動。可是A的HW指向的消息和B的HW指向的消息並非一回事，這顯然就不是咱們想要的了。

四、數據丟失問題的解決

kafka 0.11版本以後引入了leader epoch（我理解其實就是帶有版本信息的HW）來取代HW，同時重啓後的follower增長了一種請求，解決了這個問題。（這也是爲啥商業使用基本上都用0.11以後的版本）。epoch其實是一對值（epoch,offset）。 epoch 表示 leader 的版本號，offset表示本次寫入的位置。好比（0,0）就表明這是第一次寫入，寫入位置是0 。(1,120)就表明，這是第二次寫入，寫入位置是120（前面已經寫了119個數據）

場景一的解決：

場景二的解決：

**附錄：leader 中HW的更新條件**

1.  副本成爲leader副本的時候
2.  集羣裏有broker崩潰，退出ISR的時候
3.  producer向leader推送數據的時候