Kafka是一個(gè)分布式發(fā)布訂閱消息系統(tǒng)。由LinkedIn開發(fā)并已經(jīng)在2011年7月成為apache***項(xiàng)目。kafka在LinkedIn, Twitte等許多公司都得到廣泛使用，主要用于：日志聚合，消息隊(duì)列，實(shí)時(shí)監(jiān)控等。

0.8版本開始，kafka支持集群內(nèi)復(fù)制，從而提高可用性和系統(tǒng)穩(wěn)定性，這篇文章主要概述kafka復(fù)制的設(shè)計(jì)。

復(fù)制

有了復(fù)制后，kafka客戶端將會(huì)得到如下好處：

• 生產(chǎn)者能在出現(xiàn)故障的時(shí)候繼續(xù)發(fā)布消息，并且能在延遲和持久性之間選擇，取決于應(yīng)用。
• 消費(fèi)者能在出現(xiàn)故障的時(shí)候繼續(xù)實(shí)時(shí)接受正確的消息。
• 所有的分布式系統(tǒng)必須在一致性，可用性，分區(qū)容錯(cuò)性之間進(jìn)行權(quán)衡并做出取舍(參考CAP定理)，kafka的目標(biāo)是在單個(gè)數(shù)據(jù)中心里的kafka集群也支持復(fù)制。網(wǎng)絡(luò)分區(qū)是比較少見的，因此kafka設(shè)計(jì)專注于高可用和強(qiáng)一致。強(qiáng)一致意味著所有副本數(shù)據(jù)完全一致，這簡(jiǎn)化了應(yīng)用程序開發(fā)人員的工作。

kafka是一個(gè)基于CA的系統(tǒng)(???)，zookeeper是一個(gè)基于CP的系統(tǒng)(很確定)，eureka是一個(gè)基于AP的系統(tǒng)(很確定)。

復(fù)制強(qiáng)一致

現(xiàn)有比較成熟的方案中，有兩種保持強(qiáng)一致性復(fù)制的典型方法。這兩種方法都需要副本中的一個(gè)被設(shè)計(jì)為leader，所有寫入都需要發(fā)布到該副本。leader負(fù)責(zé)處理所有的接入。并廣播這些寫到其他follower副本，并且要保證復(fù)制順序和leader的順序一致。

• ***種方法是基于法定人數(shù)。leader等待直到大多數(shù)副本收到數(shù)據(jù)。當(dāng)leader出現(xiàn)故障，大多數(shù)follower會(huì)協(xié)調(diào)選舉出新的leader。這種方法被用于Apache Zookeeper 和Google's Spanner.
• 第二種方法是leader等待所有副本收到數(shù)據(jù)(重要說(shuō)明：在kafka中這個(gè)"所有"是所有的In-Sync Replicas)。如果leader出現(xiàn)故障，其他副本能被選舉為新的leader。

kafka復(fù)制選擇的是第二種方法，有兩個(gè)主要原因：

相同數(shù)量的副本情況下，第二種方法能承受更多的容錯(cuò)。例如，總計(jì)2n+1個(gè)副本，第二種方法能承受2n個(gè)副本故障(只要還有一個(gè)ISR，就能正常寫入)，而***種方法只能承受n個(gè)副本故障。如果在只有兩個(gè)副本的情況下，***種方法不能容忍任意一個(gè)副本故障。

***種方法延遲性表現(xiàn)更好，因?yàn)橹恍枰ǘㄈ藬?shù)確認(rèn)即可，所以會(huì)隱藏掉一些比較慢的副本的影響。而kafka副本被設(shè)計(jì)在相同的數(shù)據(jù)中心的一個(gè)集群下。所以網(wǎng)絡(luò)延遲這種變數(shù)是比較小的。

術(shù)語(yǔ)

為了了解kafka中的副本是如何實(shí)現(xiàn)的，我們首先需要介紹一些基本概念。在kafka中，消息流由topic定義，topic被切分為1個(gè)或者多個(gè)分區(qū)(partition)，復(fù)制發(fā)生在分區(qū)級(jí)別，每個(gè)分區(qū)有一個(gè)或者多個(gè)副本。

副本被均勻分配到kafka集群的不同服務(wù)器(稱為broker)上。每個(gè)副本都維護(hù)磁盤上的日志。生產(chǎn)者發(fā)布的消息順序追加到日志中，日志中每條消息被一個(gè)單調(diào)遞增的offset標(biāo)識(shí)。

offset是分區(qū)內(nèi)的邏輯概念， 給定偏移量，可以在分區(qū)的每個(gè)副本中標(biāo)識(shí)相同的消息。 當(dāng)消費(fèi)者訂閱某個(gè)主題時(shí)，它會(huì)跟蹤每個(gè)分區(qū)中的偏移量以供使用，并使用它來(lái)向broker發(fā)出獲取消息的請(qǐng)求。

設(shè)計(jì)

kafka中增加副本的目標(biāo)是為了更強(qiáng)的持久性和高可用。kafka要保證任何成功發(fā)布的消息不會(huì)丟失，且能被消費(fèi)，即使在有一些服務(wù)器宕機(jī)的情況下。kafka復(fù)制的主要目標(biāo)有：

可配置的持久化保證：例如，某些數(shù)據(jù)不能容忍丟失的應(yīng)用，可以選擇更強(qiáng)的持久性，當(dāng)然會(huì)伴隨延遲的增長(zhǎng)。另一個(gè)產(chǎn)生海量允許部分?jǐn)?shù)據(jù)丟失的應(yīng)用，可以選擇稍微弱一點(diǎn)的持久性，但是更獲得更好的寫入響應(yīng)時(shí)間，得到更好的吞吐量。

自動(dòng)化的副本管理：kafka要簡(jiǎn)化向broker分配副本的指配過程，并且能支持集群逐步擴(kuò)容&縮容。

這樣的話，有兩個(gè)主要問題需要解決：

• 如何均勻的指配分區(qū)的副本給broker?
• 對(duì)于一個(gè)給定的分區(qū)，如何廣播每條消息到其他副本?

數(shù)據(jù)復(fù)制

kafka允許客戶端選擇異步或者同步復(fù)制，異步復(fù)制的話，發(fā)布的消息，當(dāng)被1個(gè)副本接收到就能確認(rèn)。同步復(fù)制的話，kafka盡***努力確保消息到達(dá)多個(gè)副本(所以有效的ISR)后才確認(rèn)。當(dāng)客戶端嘗試發(fā)布消息到一個(gè)topic的分區(qū)時(shí)，kafka必須傳播這個(gè)消息到所有副本，kafka必須決定：

• 怎樣傳播;
• 在向客戶端確認(rèn)前，需要多少副本接收消息;
• 一個(gè)副本故障后，該怎么處理;
• 一個(gè)故障的副本恢復(fù)后該怎么處理;

實(shí)現(xiàn)

保持副本同步有兩種常用的策略：主備復(fù)制和基于仲裁復(fù)制。這兩種情況下，一個(gè)副本被設(shè)計(jì)為leader，其他副本被稱為follower，所有寫請(qǐng)求都由leader處理，leader傳播寫請(qǐng)求給follower。

在主備復(fù)制下，leader等待直到寫在這個(gè)組里每個(gè)副本都完成，才向客戶端發(fā)送確認(rèn)。如果某個(gè)副本故障，leader把它從這個(gè)組移除，并繼續(xù)寫到剩余的副本。一個(gè)故障副本也被允許從新加入組，只要它恢復(fù)，并追趕上leader。在用n個(gè)副本的前提下，主備復(fù)制模式能容忍n-1個(gè)副本故障。

在基于仲裁方法下，leader等待直到寫在大多數(shù)副本上完成，副本組的大小不會(huì)因?yàn)槟承└北竟收习l(fā)生改變(例如某個(gè)分區(qū)有5個(gè)副本，即使有2個(gè)副本故障，我們還是認(rèn)為這個(gè)副本組有5個(gè)副本)。因此如果有2n+1個(gè)副本，基于仲裁復(fù)制的話，只能容忍n個(gè)副本故障。如果leader出現(xiàn)故障，需要至少n+1個(gè)副本才能選舉一個(gè)新的leader。

這兩種方法需要權(quán)衡：

• 基于仲裁比主備有更好的寫延遲，任何副本的延遲(例如FGC造成長(zhǎng)時(shí)間的STW)將增加主備方法的寫延遲，但是不會(huì)增加仲裁方法的寫延遲。
• 在相同數(shù)量副本情況下，主備方法能容忍更多故障。
• 在主備方法前提下，副本因子是2，也能運(yùn)行良好。但是在基于仲裁方法的復(fù)制，兩個(gè)副本必須持續(xù)工作保持有效狀態(tài)。
• kafka選擇主備復(fù)制，因?yàn)樗苋萑谈喔北竟收?，并且只?個(gè)副本也能正常工作。

同步復(fù)制

kafka同步復(fù)制是典型的主備方式，每個(gè)分區(qū)有n個(gè)副本，并且能容忍n-1個(gè)副本故障。只有一個(gè)副本被選舉為leader，其他都是follower。leader維護(hù)了一個(gè)ISR集合：這個(gè)副本集完全和leader保持同步狀態(tài)，kafka還會(huì)把當(dāng)前的leader和當(dāng)前的ISR保持到zookeeper中。

每個(gè)副本保存信息在本地日志中，并且維護(hù)了一個(gè)日志中重要的offset位置。LEO表示日志尾部，HW是***提交消息的offset。每個(gè)日志周期性的同步到磁盤，已經(jīng)刷新的偏移量之前的數(shù)據(jù)保證保留在磁盤上。

寫

為了發(fā)布消息到分區(qū)，客戶端首先從zookeeper中找到分區(qū)的leader，然后發(fā)送消息到這個(gè)leader。leader寫消息到它的本地日志，每個(gè)follower經(jīng)常從leader拉取***的消息。所以，follower接收到的所有消息的順序和leader保持一致，follower把每條接收到的消息寫入它的本地日志，并向leader發(fā)送一個(gè)確認(rèn)。一旦leader接收到所有ISR副本的確認(rèn)，消息就能被提交。leader推進(jìn)HW，然后向客戶端發(fā)送確認(rèn)。為了更好的性能，每個(gè)follower在把消息寫入內(nèi)存后，就發(fā)送確認(rèn)。因此，對(duì)于每條提交的消息，我們保證它被保存到多個(gè)副本的內(nèi)容中然而，不保證任何副本已經(jīng)持久化已提交消息到磁盤上。

由于這種相關(guān)故障相對(duì)罕見，并且這種方法能給我們一個(gè)在響應(yīng)時(shí)間和持久性之間一個(gè)很好的平衡。在將來(lái)，kafka可能考慮增加一個(gè)選項(xiàng)參數(shù)從而提供更強(qiáng)的保證。

讀

為了簡(jiǎn)化，讀也是leader提供服務(wù)，并且只有HW以上的消息才會(huì)被暴露給消費(fèi)者讀取。

異步復(fù)制

為了支持異步復(fù)制，leader可以在消息寫入本地日志后，馬上通知客戶端。唯一需要注意的是在追趕階段，follower必須截?cái)郒W位置以后的數(shù)據(jù)。follower主要是異步復(fù)制，所以不能保證提交的消息在broker故障后不丟失。

復(fù)制實(shí)現(xiàn)

kafka復(fù)制示意圖如下所示： 

• 集群總計(jì)4個(gè)broker(broker1~broker4);
• 1個(gè)topic，2個(gè)分區(qū)，3個(gè)副本;
• 分區(qū)1即topic1-part1的leader在broker1上，分區(qū)2即topic1-part2的leader在broker4上;

producer寫入消息到分區(qū)topic1-part1的leader上(在broker1上)，然后復(fù)制到它的兩個(gè)副本，分別在broker2和broker3上。

producer寫入消息到分區(qū)topic1-part2的leader上(在broker4上)，然后復(fù)制到它的兩個(gè)副本，分別在broker2和broker3上。

當(dāng)生產(chǎn)者發(fā)布消息到topic的某個(gè)分區(qū)時(shí)，消息首先被傳遞到leader副本，并追加日志。follower副本從leader中不停的拉取新消息，一旦有足夠的副本收到消息，leader就會(huì)提交這個(gè)消息。

這里有個(gè)問題，leader是怎么決定什么是足夠的。kafka維護(hù)了一個(gè) in-sync replica(ISR)集合。這個(gè)ISR副本集都是存活的，并且完全趕上leader的副本，沒有消息延遲(leader總是在ISR集合中)。當(dāng)分區(qū)初始化創(chuàng)建時(shí)，每個(gè)副本都在ISR集合中。當(dāng)新消息發(fā)布后，leader提交消息前一直等待直到所有ISR副本收到消息。如果某個(gè)follower副本故障，它將會(huì)被從ISR中移除。leader會(huì)繼續(xù)提交新的消息，只不過ISR數(shù)量相比分區(qū)創(chuàng)建時(shí)副本數(shù)量更少。

請(qǐng)注意，現(xiàn)在，系統(tǒng)運(yùn)行在under replicated模式。

leader還會(huì)維護(hù)high watermark (HW，可以翻譯成高水位)，是指分區(qū)中***一次提交消息的offset。HW會(huì)被不斷傳播給follower副本： 

kafka high watermark

當(dāng)一個(gè)故障副本被重啟后，它首先從磁盤上恢復(fù)***的HW，并將日志截?cái)嗟紿W。這是必要的，因?yàn)椴荒鼙ＷC在HW之后的消息被提交，所以可能需要丟棄。然后副本成為follower，并繼續(xù)從leader那里獲取HW以后的消息。一旦完全趕上leader，這個(gè)副本從新被加入到ISR中。系統(tǒng)將重新回到fully replicated模式。

故障處理

kafka依賴zookeeper檢測(cè)broker故障，kafka會(huì)用一個(gè)controller(broker集合中的一個(gè))接收所有zookeeper關(guān)于故障，選舉新leader等相關(guān)通知，這樣還有一個(gè)好處，減少了對(duì)zookeeper的壓力。如果某個(gè)leader故障，controller就會(huì)從ISR副本中選舉一個(gè)新的leader，并發(fā)布新leader的消息給其他follower。

按照設(shè)計(jì)，leader選舉過程中，已經(jīng)提交的消息總是會(huì)被保留，一些未提交的消息可能會(huì)丟失。leader和每個(gè)分區(qū)的ISR也會(huì)被保存在Zookeeper中，controller出現(xiàn)故障轉(zhuǎn)移時(shí)需要用到。由于broker級(jí)別的故障一般會(huì)非常少，所以預(yù)期的leader和ISR都會(huì)不經(jīng)常改變。

對(duì)客戶端來(lái)說(shuō)，broker僅向消費(fèi)者公開已經(jīng)提交的消息。broker故障期間，已提交的數(shù)據(jù)始終被保留。消費(fèi)者使用相同的offset可以從另一個(gè)被選舉為leader的副本拉取消息。

生產(chǎn)者能選擇在broker收到消息后何時(shí)得到broker的確認(rèn)。例如，它能等到消息被leader提交并被所有ISR確認(rèn)(即acks=-1)。另外，也可以選擇消息只要被leader追加到日志中，可能還沒有提交(acks=0表示無(wú)需等待leader確認(rèn)，acks=1表示需要等待leader確認(rèn))。前一種情況即acks=-1，生產(chǎn)者需要等待更長(zhǎng)的時(shí)間。但是確認(rèn)的消息都保證在broker中保留。后一種情況即acks=0或者1，生產(chǎn)者有更低的延遲，更高的吞吐量，但一些確認(rèn)的消息在broker故障時(shí)可能會(huì)丟失。如何抉擇，由你決定。