kafka是如何保證消息的可靠性與一致性

這篇文章將為大家詳細(xì)講解有關(guān)kafka是如何保證消息的可靠性與一致性，文章內(nèi)容質(zhì)量較高，因此小編分享給大家做個(gè)參考，希望大家閱讀完這篇文章后對(duì)相關(guān)知識(shí)有一定的了解。

在kafka中主要通過(guò)ISR機(jī)制來(lái)保證消息的可靠性。下面通過(guò)幾個(gè)問(wèn)題來(lái)說(shuō)明kafka如何來(lái)保證消息可靠性與一致性

在kafka中ISR是什么？

在zk中會(huì)保存AR（Assigned Replicas）列表，其中包含了分區(qū)所有的副本，其中 AR = ISR+OSR

• ISR（in sync replica）：是kafka動(dòng)態(tài)維護(hù)的一組同步副本，在ISR中有成員存活時(shí)，只有這個(gè)組的成員才可以成為leader，內(nèi)部保存的為每次提交信息時(shí)必須同步的副本（acks = all時(shí)），每當(dāng)leader掛掉時(shí)，在ISR集合中選舉出一個(gè)follower作為leader提供服務(wù)，當(dāng)ISR中的副本被認(rèn)為壞掉的時(shí)候，會(huì)被踢出ISR，當(dāng)重新跟上leader的消息數(shù)據(jù)時(shí)，重新進(jìn)入ISR。

• OSR（out sync replica）: 保存的副本不必保證必須同步完成才進(jìn)行確認(rèn)，OSR內(nèi)的副本是否同步了leader的數(shù)據(jù)，不影響數(shù)據(jù)的提交，OSR內(nèi)的follower盡力的去同步leader，可能數(shù)據(jù)版本會(huì)落后。

kafka如何控制需要同步多少副本才可以返回確定到生產(chǎn)者消息才可用？

• 當(dāng)寫(xiě)入到kakfa時(shí)，生產(chǎn)者可以選擇是否等待0（只需寫(xiě)入leader）,1（只需同步一個(gè)副本） 或 -1（全部副本）的消息確認(rèn)（這里的副本指的是ISR中的副本）。

• 需要注意的是“所有副本確認(rèn)”并不能保證全部分配副本已收到消息。默認(rèn)情況下，當(dāng)acks=all時(shí)，只要當(dāng)前所有在同步中的副本（ISR中的副本）收到消息，就會(huì)進(jìn)行確認(rèn)。所以Kafka的交付承諾可以這樣理解：對(duì)沒(méi)有提交成功的消息不做任何交付保證，而對(duì)于ISR中至少有一個(gè)存活的完全同步的副本的情況下的“成功提交”的消息保證不會(huì)丟失。

對(duì)于kafka節(jié)點(diǎn)活著的條件是什么？

• 第一點(diǎn)：一個(gè)節(jié)點(diǎn)必須維持和zk的會(huì)話，通過(guò)zk的心跳檢測(cè)實(shí)現(xiàn)

• 第二點(diǎn)：如果節(jié)點(diǎn)是一個(gè)slave也就是復(fù)制節(jié)點(diǎn)，那么他必須復(fù)制leader節(jié)點(diǎn)不能太落后。這里的落后可以指兩種情況

• 1：數(shù)據(jù)復(fù)制落后，slave節(jié)點(diǎn)和leader節(jié)點(diǎn)的數(shù)據(jù)相差較大，這種情況有一個(gè)缺點(diǎn)，在生產(chǎn)者突然發(fā)送大量消息導(dǎo)致網(wǎng)絡(luò)堵塞后，大量的slav復(fù)制受阻，導(dǎo)致數(shù)據(jù)復(fù)制落后被大量的踢出ISR。

• 2：時(shí)間相差過(guò)大，指的是slave向leader請(qǐng)求復(fù)制的時(shí)間距離上次請(qǐng)求相隔時(shí)間過(guò)大。通過(guò)配置 replica.lag.time.max就可以配置這個(gè)時(shí)間參數(shù)。這種方式解決了上述第一種方式導(dǎo)致的問(wèn)題。

kafka分區(qū)partition掛掉之后如何恢復(fù)？

在kafka中有一個(gè)partition recovery機(jī)制用于恢復(fù)掛掉的partition。

每個(gè)Partition會(huì)在磁盤(pán)記錄一個(gè)RecoveryPoint（恢復(fù)點(diǎn)）, 記錄已經(jīng)flush到磁盤(pán)的最大offset。當(dāng)broker fail 重啟時(shí),會(huì)進(jìn)行l(wèi)oadLogs。首先會(huì)讀取該P(yáng)artition的RecoveryPoint,找到包含RecoveryPoint點(diǎn)上的segment及以后的segment, 這些segment就是可能沒(méi)有完全flush到磁盤(pán)segments。然后調(diào)用segment的recover,重新讀取各個(gè)segment的msg,并重建索引。

優(yōu)點(diǎn)：

1. 以segment為單位管理Partition數(shù)據(jù),方便數(shù)據(jù)生命周期的管理,刪除過(guò)期數(shù)據(jù)簡(jiǎn)單

2. 在程序崩潰重啟時(shí),加快recovery速度,只需恢復(fù)未完全flush到磁盤(pán)的segment即可

什么原因?qū)е赂北九cleader不同步的呢？

• 慢副本：在一定周期時(shí)間內(nèi)follower不能追趕上leader。最常見(jiàn)的原因之一是I / O瓶頸導(dǎo)致follower追加復(fù)制消息速度慢于從leader拉取速度。

• 卡住副本：在一定周期時(shí)間內(nèi)follower停止從leader拉取請(qǐng)求。follower replica卡住了是由于GC暫?；騠ollower失效或死亡。

• 新啟動(dòng)副本：當(dāng)用戶(hù)給主題增加副本因子時(shí)，新的follower不在同步副本列表中，直到他們完全趕上了leader日志。

一個(gè)partition的follower落后于leader足夠多時(shí)，被認(rèn)為不在同步副本列表或處于滯后狀態(tài)。

正如上述所說(shuō)，現(xiàn)在kafka判定落后有兩種，副本滯后判斷依據(jù)是副本落后于leader最大消息數(shù)量(replica.lag.max.messages)或replicas響應(yīng)partition leader的最長(zhǎng)等待時(shí)間(replica.lag.time.max.ms)。前者是用來(lái)檢測(cè)緩慢的副本,而后者是用來(lái)檢測(cè)失效或死亡的副本

如果ISR內(nèi)的副本掛掉怎么辦？

1. 兩種選擇：服務(wù)直接不可用一段時(shí)間等待ISR中副本恢復(fù)（祈禱恢復(fù)的副本有數(shù)據(jù)吧） 或者 直接選用第一個(gè)副本（這個(gè)副本不一定在ISR中）作為leader，這兩種方法也是在可用性和一致性之間的權(quán)衡。

2. 服務(wù)不可用方式這種適用在不允許消息丟失的情況下使用，適用于一致性大于可用性，可以有兩種做法

2.1 設(shè)置ISR最小同步副本數(shù)量，如果ISR的當(dāng)前數(shù)量大于設(shè)置的最小同步值，那么該分區(qū)才會(huì)接受寫(xiě)入，避免了ISR同步副本過(guò)少。如果小于最小值那么該分區(qū)將不接收寫(xiě)入。這個(gè)最小值設(shè)置只有在acks = all的時(shí)候才會(huì)生效。

2.2 禁用unclean-leader選舉，當(dāng)isr中的所有副本全部不可用時(shí)，不可以使用OSR 中的副本作為leader，直接使服務(wù)不可用，直到等到ISR 中副本恢復(fù)再進(jìn)行選舉leader。

3. 直接選擇第一個(gè)副本作為leader的方式，適用于可用性大于一致性的場(chǎng)景，這也是kafka在isr中所有副本都死亡了的情況采用的默認(rèn)處理方式，我們可以通過(guò)配置參數(shù) unclean.leader.election.enable來(lái)禁止這種行為，采用第一種方法。

那么ISR是如何實(shí)現(xiàn)同步的呢？

broker的offset大致分為三種：base offset、high watemark（HW）、log end offset（LEO）

• base offset：起始位移，replica中第一天消息的offset

• HW：replica高水印值，副本中最新一條已提交消息的位移。leader 的HW值也就是實(shí)際已提交消息的范圍，每個(gè)replica都有HW值，但僅僅leader中的HW才能作為標(biāo)示信息。什么意思呢，就是說(shuō)當(dāng)按照參數(shù)標(biāo)準(zhǔn)成功完成消息備份（成功同步給follower replica后）才會(huì)更新HW的值，代表消息理論上已經(jīng)不會(huì)丟失，可以認(rèn)為“已提交”。

• LEO：日志末端位移，也就是replica中下一條待寫(xiě)入消息的offset，注意哈，是下一條并且是待寫(xiě)入的，并不是最后一條。這個(gè)LEO個(gè)人感覺(jué)也就是用來(lái)標(biāo)示follower的同步進(jìn)度的。所以HW代表已經(jīng)完成同步的數(shù)據(jù)的位置，LEO代表已經(jīng)寫(xiě)入的最新位置，只有HW位置之前的才是可以被外界訪問(wèn)的數(shù)據(jù)?，F(xiàn)在就來(lái)看一下之前，broker從收到消息到返回響應(yīng)這個(gè)黑盒子里發(fā)生了什么。 

1. broker 收到producer的請(qǐng)求

2. leader 收到消息，并成功寫(xiě)入，LEO 值+1

3. broker 將消息推給follower replica，follower 成功寫(xiě)入 LEO +1 …

4. 所有LEO 寫(xiě)入后，leader HW +1

5. 消息可被消費(fèi)，并成功響應(yīng)

上述過(guò)程從下面的圖便可以看出：

解決上一個(gè)問(wèn)題后，接下來(lái)就是kafka如何選用leader呢？

選舉leader常用的方法是多數(shù)選舉法，比如Redis等，但是kafka沒(méi)有選用多數(shù)選舉法，kafka采用的是quorum（法定人數(shù)）。

quorum是一種在分布式系統(tǒng)中常用的算法，主要用來(lái)通過(guò)數(shù)據(jù)冗余來(lái)保證數(shù)據(jù)一致性的投票算法。在kafka中該算法的實(shí)現(xiàn)就是ISR，在ISR中就是可以被選舉為leader的法定人數(shù)。

• 在leader宕機(jī)后，只能從ISR列表中選取新的leader，無(wú)論ISR中哪個(gè)副本被選為新的leader，它都知道HW之前的數(shù)據(jù)，可以保證在切換了leader后，消費(fèi)者可以繼續(xù)看到HW之前已經(jīng)提交的數(shù)據(jù)。

• HW的截?cái)鄼C(jī)制：選出了新的leader，而新的leader并不能保證已經(jīng)完全同步了之前l(fā)eader的所有數(shù)據(jù)，只能保證HW之前的數(shù)據(jù)是同步過(guò)的，此時(shí)所有的follower都要將數(shù)據(jù)截?cái)嗟紿W的位置，再和新的leader同步數(shù)據(jù)，來(lái)保證數(shù)據(jù)一致。當(dāng)宕機(jī)的leader恢復(fù)，發(fā)現(xiàn)新的leader中的數(shù)據(jù)和自己持有的數(shù)據(jù)不一致，此時(shí)宕機(jī)的leader會(huì)將自己的數(shù)據(jù)截?cái)嗟藉礄C(jī)之前的hw位置，然后同步新leader的數(shù)據(jù)。宕機(jī)的leader活過(guò)來(lái)也像follower一樣同步數(shù)據(jù)，來(lái)保證數(shù)據(jù)的一致性。

關(guān)于kafka是如何保證消息的可靠性與一致性就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺(jué)得文章不錯(cuò)，可以把它分享出去讓更多的人看到。