在聊Kafka高可靠之前，先在評(píng)論區(qū)來(lái)波RNG NB好不好!

什么叫可靠性?

大家都知道，系統(tǒng)架構(gòu)有三高：「高性能、高并發(fā)和高可用」，三者的重要性不言而喻。

對(duì)于任意系統(tǒng)，想要同時(shí)滿(mǎn)足三高都是一件非常困難的事情，大型業(yè)務(wù)系統(tǒng)或者傳統(tǒng)中間件都會(huì)搭建復(fù)雜的架構(gòu)來(lái)保證。

除以上三種模式之外，還有一個(gè)指標(biāo)方向也很重要，那就是高可靠，甚至你可能會(huì)將它和「高可用」混淆起來(lái)。

事實(shí)上兩者并不一樣，高可用會(huì)更偏向于整體服務(wù)的可用性，防止系統(tǒng)宕機(jī)等等。而高可靠是指數(shù)據(jù)的可靠性保證嘛，你可以理解”高可靠“相比于系統(tǒng)三高會(huì)是一個(gè)更細(xì)一點(diǎn)的概念。

那么什么是數(shù)據(jù)的高可靠呢，總結(jié)一下就是系統(tǒng)要提供可靠的數(shù)據(jù)支撐，不能發(fā)生丟失、重復(fù)等錯(cuò)誤現(xiàn)象。

所以每個(gè)開(kāi)源中間件在發(fā)布版本時(shí)都會(huì)通過(guò)文檔聲明自己是超可靠的，就像520那天每一位暖男說(shuō)的那樣。

咱今天的主角kafka就是這么一個(gè)例子。

一些重要概念

因?yàn)橛幸欢螘r(shí)間沒(méi)講消息隊(duì)列了嘛，為了幫助你更好理解文章，我們來(lái)先復(fù)習(xí)一下kafka的基礎(chǔ)概念：

• record：消息，消息隊(duì)列基礎(chǔ)通信單位。
• topic：主題，目的就是將消息進(jìn)行分類(lèi)，不同業(yè)務(wù)類(lèi)型的消息通常會(huì)被分發(fā)到不同的主題。
• partition：分區(qū)，每個(gè)主題可以創(chuàng)建多個(gè)分區(qū)，每個(gè)分區(qū)都由一系列有序和不可變的消息組成。
• replica：副本，每個(gè)分區(qū)都有一個(gè)至多個(gè)副本存在，它的主要作用是存儲(chǔ)保存數(shù)據(jù)，以日志(Log)對(duì)象的形式體現(xiàn)。副本又分為leader副本和follower副本。
• offset：偏移量，每一個(gè)消息在日志文件中的位置都對(duì)應(yīng)一個(gè)按序遞增的偏移量，你可以理解為類(lèi)似數(shù)組的存儲(chǔ)形式。
• producer：生產(chǎn)者，生產(chǎn)消息的那一方。
• consumer：消費(fèi)者，通常不同的業(yè)務(wù)都會(huì)有一到多個(gè)消費(fèi)者組成消費(fèi)者集群。
• broker：代理，一個(gè)Kafka集群由一個(gè)或多個(gè)Kafka實(shí)例構(gòu)成，每一個(gè)Kafka實(shí)例就稱(chēng)為代理。

如上圖所示，一共存在主題1和主題2，主題1有兩個(gè)分區(qū)，主題2只有一個(gè)分區(qū)，并且每個(gè)分區(qū)都存在一個(gè)leader副本和兩個(gè)follower副本，它們分布在每個(gè)不同的代理節(jié)點(diǎn)上。

partition里只有l(wèi)eader副本負(fù)責(zé)與生產(chǎn)者、消費(fèi)者之間數(shù)據(jù)的交互，follower副本會(huì)定期從leader副本拉取數(shù)據(jù)以保證整個(gè)集群數(shù)據(jù)可用性。

如何保證數(shù)據(jù)高可靠

Kafka是通過(guò)副本機(jī)制實(shí)現(xiàn)數(shù)據(jù)的存儲(chǔ)的，所以就需要一些機(jī)制保證數(shù)據(jù)在跨集群的副本之間能夠可靠地傳輸。

1、副本同步集合

業(yè)務(wù)數(shù)據(jù)封裝成消息在系統(tǒng)中流轉(zhuǎn)，由于各個(gè)組件都是分布在不同的服務(wù)器上的，所以主題和生產(chǎn)者、消費(fèi)者之間的數(shù)據(jù)同步可能存在一定的時(shí)間延遲，Kafka通過(guò)延遲范圍劃分了幾個(gè)不同的集合：

AR(Assigned Replicas)

指的是已經(jīng)分配數(shù)據(jù)的分區(qū)副本，通常指的是leader副本 + follower副本。

ISR(In Sync Replicas)

指的是和leader副本數(shù)據(jù)保持同步的副本集合。當(dāng)follower副本數(shù)據(jù)和leader副本數(shù)據(jù)保持同步，那么這些副本就處在ISR里面，ISR集合會(huì)根據(jù)數(shù)據(jù)的同步狀態(tài)動(dòng)態(tài)變化。

OSR(Out Sync Replicas)

一旦follower副本的數(shù)據(jù)同步進(jìn)度跟不上leader了，那么它就會(huì)被放進(jìn)叫做OSR的集合里。也就是這個(gè)集合包含的是不處于同步狀態(tài)的分區(qū)副本。

OK，那有什么標(biāo)準(zhǔn)判斷它是同步還是不同步呢?

通過(guò)replica.lag.time.max.ms這個(gè)參數(shù)來(lái)設(shè)置數(shù)據(jù)同步時(shí)間差，它的默認(rèn)值是10s。

一旦從分區(qū)副本和主分區(qū)副本的消息相差10s以上，那么就認(rèn)為消息處于OSR不同步的狀態(tài)。若follower處于OSR集合里，那么在選取新的leader的時(shí)候就不會(huì)選舉它作為新leader。

2、ACK應(yīng)答機(jī)制

我們剛剛說(shuō)了kafka是通過(guò)ack來(lái)發(fā)送數(shù)據(jù)同步信號(hào)的，那信號(hào)發(fā)送頻率又有幾種設(shè)定呢?

• ack = 0

生產(chǎn)者發(fā)送一次消息就不再發(fā)送。不管是否發(fā)送成功，若發(fā)出去的消息處于通信的路上就丟失，或者還未做磁盤(pán)持久化操作，那么消息就可能丟失。

它的好處就是性能很高，你想呀你發(fā)送消息都不需要等待對(duì)方回復(fù)就持續(xù)發(fā)送下一批，那么消息等待的時(shí)間就節(jié)省出來(lái)了。同一時(shí)間范圍內(nèi)能比別人處理更多數(shù)據(jù)，缺點(diǎn)就是它的可靠性真的很低，數(shù)據(jù)真的是說(shuō)丟就丟。

• ack = 1

leader接收到消息并且寫(xiě)入到本地磁盤(pán)后就認(rèn)為消息處理成功。這種方式可靠性會(huì)比上一種好一些，當(dāng)leader接收到消息并且寫(xiě)入到本地磁盤(pán)后就認(rèn)為消息處理成功，不論follower是否同步完這條消息就會(huì)返回給producer。

但是假如此刻partition leader所在的broker宕機(jī)了，如果那么數(shù)據(jù)也可能會(huì)丟失，所以follower副本的數(shù)據(jù)同步就很重要。

Kafka默認(rèn)就采用這種方式。

• ack = -1

producer只有收到分區(qū)內(nèi)所有副本的響應(yīng)ACK才會(huì)認(rèn)為消息已經(jīng)push成功。

這種方式雖然對(duì)于數(shù)據(jù)的可靠保障做得很好，但是就是性能很差，影響吞吐量，所以一般也不會(huì)采取。

那么它就絕對(duì)可靠嗎?也不一定。最重要的還是取決于副本數(shù)據(jù)是否同步完成。若producer收到響應(yīng)消息前l(fā)eader副本掛掉，那么producer會(huì)因未收到消息重復(fù)發(fā)送消息，那就可能造成數(shù)據(jù)重復(fù)。怎么解決呢?只要保證業(yè)務(wù)冪等就行。

我們可以通過(guò)request.required.acks這個(gè)參數(shù)控制消息的發(fā)送頻率。

3、消息語(yǔ)義

消息集群整體是一個(gè)復(fù)雜的系統(tǒng)，所以過(guò)程中可能會(huì)因?yàn)楦鞣N原因?qū)е孪鬟f出錯(cuò)，Kafka對(duì)于這些可能遇到的場(chǎng)景定義了對(duì)應(yīng)的的消息語(yǔ)義。

at most once

它代表消息可能被消費(fèi)者消費(fèi)0次或者1次。若場(chǎng)景如下：

• 消息從partition分發(fā)給消費(fèi)者集群。
• 消費(fèi)者把自己收到的消息告訴集群，集群收到之后offset就會(huì)往后移動(dòng)。
• 消費(fèi)者將數(shù)據(jù)入庫(kù)做持久化。

你一定想到了。在第三步消費(fèi)者將消息入庫(kù)時(shí)若因任何原因消費(fèi)者A掛了，那么在將消費(fèi)者切換到集群的消費(fèi)者B后，數(shù)據(jù)還沒(méi)入庫(kù)呢。此時(shí)partition是渾然不知的呀，那么這就會(huì)造成一個(gè)問(wèn)題：數(shù)據(jù)丟失。

at least once

它代表partition分發(fā)的消息至少被消費(fèi)一次。其通信過(guò)程如下：

• 消息從partition分發(fā)給消費(fèi)者集群。
• 消費(fèi)者將數(shù)據(jù)入庫(kù)做持久化。
• 消費(fèi)者把自己收到的消息告訴集群，集群收到之后offset就會(huì)往后移動(dòng)。

假設(shè)consumer group在數(shù)據(jù)入庫(kù)之后，在將數(shù)據(jù)返回給partition的過(guò)程中消費(fèi)者A掛了，那么partition會(huì)因?yàn)榻邮詹坏巾憫?yīng)ACK而重新發(fā)送數(shù)據(jù)，此時(shí)消費(fèi)者B可能再次將原先的消息入庫(kù)，這就造成了數(shù)據(jù)重復(fù)了。

在沒(méi)有做任何冪等性保護(hù)的情況下，像重復(fù)轉(zhuǎn)賬，重付疊加積分這種業(yè)務(wù)，那么結(jié)果可能是致命的。

exactly once

代表消息正好能被消費(fèi)一次，不丟失，不重復(fù)。

在at least once的情況基礎(chǔ)上，假設(shè)consumerA在返回ack給partition的過(guò)程中宕機(jī)了。那么consumerB不會(huì)跟著partition的offset走，它會(huì)先去數(shù)據(jù)庫(kù)里面查看最新消息對(duì)應(yīng)的偏移位，再根據(jù)這個(gè)偏移位返回Kafka集群從對(duì)應(yīng)的偏移位置出發(fā)，這就可以避免消息重復(fù)和消息丟失。

4、數(shù)據(jù)截?cái)鄼C(jī)制

我們開(kāi)頭說(shuō)了真正處理數(shù)據(jù)的是leader副本，follower副本只負(fù)責(zé)數(shù)據(jù)的同步和保存，那如果因?yàn)閘eader宕機(jī)了二者數(shù)據(jù)不一致會(huì)怎么樣呢?

在講一致性保證過(guò)程之前還需了解兩個(gè)Kafka用于表示副本數(shù)據(jù)同步的概念：

HW(High Watermark)：中文翻譯為高水位，用來(lái)體現(xiàn)副本間數(shù)據(jù)同步的相對(duì)位置，consumer最多只能消費(fèi)到HW所在的位置，通過(guò)HW我們可以判斷數(shù)據(jù)對(duì)副本是否可見(jiàn)。

LEO(Log End Offset)：下一條待寫(xiě)入消息的記錄位置。

leader副本從生產(chǎn)者獲取消息，follower副本實(shí)時(shí)從leder同步數(shù)據(jù)，此時(shí)它們的同步數(shù)據(jù)是一致的都同步到2這個(gè)位置，并且下一個(gè)寫(xiě)入的消息都是偏移位4：

假設(shè)因?yàn)橐馔鈒eader發(fā)生宕機(jī)，follower即被選為新leader，此后從生產(chǎn)者寫(xiě)入最新的偏移位4和5：

過(guò)了一段時(shí)間原leader通過(guò)修復(fù)恢復(fù)服務(wù)，它就會(huì)發(fā)現(xiàn)自己和新leader的數(shù)據(jù)是不一致的：

為了保證數(shù)據(jù)一致性就必須強(qiáng)行讓一方妥協(xié)。因?yàn)閿?shù)據(jù)是不斷在刷新的，所以舊leader此時(shí)的優(yōu)先級(jí)會(huì)小于新leader，因此它會(huì)將自己的數(shù)據(jù)截?cái)嗟脚c新leader相同的HW和LEO位置，確保和新leader的數(shù)據(jù)一定相同，這就是Kafka數(shù)據(jù)截?cái)鄼C(jī)制。

5、數(shù)據(jù)清理機(jī)制

同其它中間件一樣，Kafka的主要作用是通信，所以即使是將數(shù)據(jù)保存在磁盤(pán)上它還是會(huì)占用一定空間。為了節(jié)約存儲(chǔ)空間它會(huì)通過(guò)一些機(jī)制對(duì)過(guò)期數(shù)據(jù)進(jìn)行清理。

日志刪除

日志刪除會(huì)直接刪除日志分段，kafka會(huì)維護(hù)一個(gè)定時(shí)任務(wù)來(lái)周期性檢查和刪除「過(guò)期數(shù)據(jù)」。

• 基于時(shí)間的日志刪除

它在每一個(gè)日志段文件里面都維護(hù)一個(gè)最大時(shí)間戳來(lái)確認(rèn)當(dāng)前配置的刪除時(shí)間，只要日志段寫(xiě)入新消息該字段都會(huì)被更新。一個(gè)日志段被寫(xiě)滿(mǎn)了之后就不會(huì)再接收新的消息，它會(huì)去創(chuàng)建一個(gè)新的日志段文件往里面寫(xiě)數(shù)據(jù)。

每一個(gè)日志段文件被寫(xiě)滿(mǎn)之后它的最大的時(shí)間戳都是保持不變的，Kafka只要通過(guò)當(dāng)前時(shí)間與最大時(shí)間戳進(jìn)行比較就可以判斷該日志段文件是否過(guò)期。

Kafka默認(rèn)配置log.retention.hours = 168，也就是7天的日志保留時(shí)間。

• 基于容量大小的日志刪除

這和以上是異曲同工的方式， 只不過(guò)這次從時(shí)間換成了空間。

Kafka會(huì)通過(guò)每個(gè)日志段空間的大小計(jì)算一個(gè)總?cè)萘块撝?，然后?jì)算出當(dāng)前的實(shí)際空間大小和總?cè)萘块撝档牟钪?，如果這個(gè)差值大于單個(gè)日志段文件的大小那么就會(huì)刪除掉最舊的那個(gè)日志段文件，反之則不做任何處理。

同理，這個(gè)閾值也可以通過(guò)log.retention.bytes參數(shù)來(lái)設(shè)置。

日志壓縮

Kafka的消息是由鍵值組成的，如果日志段里存在多條相同key但是不同value的數(shù)據(jù)，那么它會(huì)選擇性地清除舊數(shù)據(jù)，保留最近一條記錄。

具體的壓縮方式就是創(chuàng)建一個(gè)檢查點(diǎn)文件，從日志起始位置開(kāi)始遍歷到最大結(jié)束位置，然后把每個(gè)消息的key和key對(duì)應(yīng)的offset保存在一個(gè)固定容量的SkimpyOffsetMap中。

這樣前面的值就會(huì)被后面的覆蓋掉，如果日志文件里存在相同的key只有最新的那個(gè)會(huì)被保留。

總結(jié)

Kafka通過(guò)ACK應(yīng)答機(jī)制保證了不同組件之間的通信效率，通過(guò)副本同步機(jī)制、數(shù)據(jù)截?cái)嗪蛿?shù)據(jù)清理機(jī)制實(shí)現(xiàn)了對(duì)于數(shù)據(jù)的管理策略，保證整個(gè)系統(tǒng)運(yùn)行效率。

作為一款高性能又同時(shí)兼顧高可靠性的消息中間件來(lái)說(shuō)，Kafka能吹的點(diǎn)實(shí)在太多。如果本篇文章對(duì)你有所幫助，點(diǎn)擊一下右下角的大拇指，下一次我們來(lái)詳細(xì)講解Kafka是如何實(shí)現(xiàn)副本間數(shù)據(jù)傳遞的。