快捷導(dǎo)航

關(guān)于Kafka消息隊(duì)列原理的總結(jié)

更新時(shí)間：2022年05月05日 14:59:05 作者：qq838642798

這篇文章主要介紹了關(guān)于Kafka消息隊(duì)列原理的總結(jié)，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

Kafka消息隊(duì)列原理

最近在測(cè)試kafka的讀寫性能，所以借這個(gè)機(jī)會(huì)了解了kafka的一些設(shè)計(jì)原理，既然作為分布式系統(tǒng)，我們還是按照分布式的套路進(jìn)行分析。

Kafka的邏輯數(shù)據(jù)模型

生產(chǎn)者發(fā)送數(shù)據(jù)給服務(wù)端時(shí)，構(gòu)造的是ProducerRecord<Integer, String>(String topic, Integer key,String value)對(duì)象并發(fā)送，從這個(gè)構(gòu)造函數(shù)可以看到，kafka的表面邏輯數(shù)據(jù)模型是key-value。

當(dāng)然api再發(fā)送前還會(huì)在這個(gè)基礎(chǔ)上加入若干校驗(yàn)信息，不過(guò)這個(gè)對(duì)用戶而言是透明的。

Kafka的分發(fā)策略

跟很多分布式多備份系統(tǒng)類似，kafka的基本網(wǎng)絡(luò)結(jié)構(gòu)如下：

一個(gè)節(jié)點(diǎn)(Broker)中存有不同partition的備份，一個(gè)parittion存在多份備份保存在不同節(jié)點(diǎn)上并且選舉出一個(gè)作為leader跟客戶端交互，一個(gè)topic擁有多個(gè)parittion。

默認(rèn)的kafka分發(fā)算法是hash(key)%numPartitions，簡(jiǎn)單來(lái)就是哈希再取模。當(dāng)然這個(gè)算法可以自定義，只要重寫相關(guān)接口。

如上圖在一個(gè)四臺(tái)主機(jī)上創(chuàng)建了一個(gè)有兩個(gè)備份，四個(gè)分區(qū)partion的話題topic，但生產(chǎn)者需要發(fā)送某個(gè)key-value對(duì)象到消息隊(duì)列里面時(shí)，創(chuàng)建連接時(shí)通過(guò)訪問(wèn)zookeeper，獲取到一份leader partion列表（Broker1. Partition-0, Broker2. Partition1, Broker3. Partition-2, Broker4.Partition-3），再根據(jù)分發(fā)算法計(jì)算出這個(gè)對(duì)象應(yīng)該要發(fā)送到哪個(gè)leader partion中。

Kafka的物理存儲(chǔ)模型和查找數(shù)據(jù)的設(shè)計(jì)

Kafka的物理存儲(chǔ)模型比較簡(jiǎn)單，在kafka的物理持久化的存儲(chǔ)中有分Segment的概念，每個(gè)Segment有兩種類型的文件：索引文件***.index和日志文件（數(shù)據(jù)文件）***.log。兩者的命名規(guī)則都是以這個(gè)Segment的第一條的消息邏輯偏移量作為文件名。索引是稀疏索引，目的在于減少索引文件的數(shù)據(jù)量，其文件的內(nèi)容是key-value結(jié)構(gòu)，key是消息的偏移量offeset（就是一個(gè)自增的序列號(hào)），value是對(duì)應(yīng)的log文件的實(shí)際物理磁盤偏移量。

值得一提的是，跟其他正常分布式不一樣，kafka并不支持根據(jù)給定的key查找該key對(duì)應(yīng)的value值的能力，某種意義而言，邏輯數(shù)據(jù)模型中的key只是用來(lái)實(shí)現(xiàn)分發(fā)計(jì)算用的，所以使用kafka查找數(shù)據(jù)只能以指定消息的偏移量的放松實(shí)現(xiàn)。

整個(gè)查找過(guò)程：當(dāng)要查找offset=888及后續(xù)的消息時(shí)，kafka先到該節(jié)點(diǎn)上找到對(duì)應(yīng)的Segment。通過(guò)該Segment的index文件上用二分查找的方法找到最接近offset=888的紀(jì)錄，比如886，然后找到886對(duì)應(yīng)的物理磁盤偏移量999，這樣就從log的磁盤偏移量找起，連續(xù)遍歷了兩個(gè)消息后就能找到888這個(gè)消息的數(shù)據(jù)（log文件中保留了每條消息的邏輯偏移量,長(zhǎng)度和數(shù)據(jù)）。

Kafka的持久化策略設(shè)計(jì)

Kafka的持久化設(shè)計(jì)是非常有特色的，和其他分布式系統(tǒng)不同，它沒(méi)有自己維護(hù)一套緩存機(jī)制，而是直接使用了操作系統(tǒng)的文件系統(tǒng)（操作系統(tǒng)的文件系統(tǒng)自帶pagecache）。這樣的好處是減少了一次內(nèi)存拷貝的消耗。其他分布式系統(tǒng)比如cassandra,自己在服務(wù)端維護(hù)了一份數(shù)據(jù)緩沖內(nèi)存塊datacache，當(dāng)需要持久化時(shí)再調(diào)用操作系統(tǒng)的文件系統(tǒng)寫入到文件中，這樣就多了一次datacache到pagecache的拷貝消耗。這樣的話，kafka的持久化管理關(guān)鍵是管理文件系統(tǒng)的pagecache的刷盤。

由于kafka采用了這種特別的持久化策略，所以在kafka中并沒(méi)有其他分布式系統(tǒng)的重做日志。所以kafka在出現(xiàn)故障后的數(shù)據(jù)恢復(fù)策略有自己的一套：首先，kafka會(huì)通過(guò)配置文件配置pagecache定時(shí)或者定量刷盤的頻率以保證即使出現(xiàn)故障也能把丟失的數(shù)據(jù)降低到最少。其次，pageche本身是操作系統(tǒng)管理維護(hù)的，跟kafka自身的服務(wù)進(jìn)程沒(méi)有關(guān)系，如果是kafka本身掛了的話，重啟后還是能訪問(wèn)到pageche中的數(shù)據(jù)的。最后如果很不幸是kafka所在的一個(gè)節(jié)點(diǎn)的主機(jī)掛掉的話，那么重啟主機(jī)和kafka后也可以從其他備份節(jié)點(diǎn)重新同步丟失的數(shù)據(jù)。

Kafka高性能的和持久化策略關(guān)系非常密切,這部分內(nèi)容，也是整個(gè)kafka設(shè)計(jì)的精髓所在：

傳統(tǒng)的觀念認(rèn)為磁盤的讀寫是非常低效的，所以一般系統(tǒng)都會(huì)自己管理一塊內(nèi)存datacache充當(dāng)磁盤的緩存，只有需要的時(shí)候才去和磁盤交互。

但是實(shí)際上，磁盤的低效的原因不在于磁盤io,而在于磁頭的隨機(jī)尋址。如果數(shù)據(jù)是順序讀寫的話（也就是一次磁頭尋址，連續(xù)io）,其實(shí)速度是非?？斓模?Raid-5，7200rpm)：順序 I/O: 600MB/s）。

而在傳統(tǒng)的設(shè)計(jì)中雖然加入了內(nèi)存作為緩存，但是為了保證數(shù)據(jù)的安全性還是得提供一份重做日志（每次的修改操作都要記錄在重做日志redo.log中，以保證內(nèi)存丟失后能根據(jù)重做日志進(jìn)行恢復(fù)），并且當(dāng)datacache里面的數(shù)據(jù)達(dá)到一定容量時(shí)刷新到磁盤的data文件中。

但是kafka并沒(méi)有使用這套常規(guī)設(shè)計(jì)，并沒(méi)有自己維護(hù)一套datacache而是另辟蹊徑，直接使用操作系統(tǒng)中的文件系統(tǒng)，并利用文件系統(tǒng)原有的pagecache作為數(shù)據(jù)緩存。

減少了datacache到pagecache的拷貝消耗。并且順序地進(jìn)行磁盤io，這樣大大提高了kafka寫數(shù)據(jù)時(shí)持久化的效率。

對(duì)于kafka的讀數(shù)據(jù)這塊，kafka也使用了Sendfile技術(shù)來(lái)提高讀的效率，傳統(tǒng)的讀方案是讀取磁盤的數(shù)據(jù)到pagecache中，然后從pagecache拷貝一份到用戶進(jìn)程的datacache中，datacache再拷貝到內(nèi)核的socket緩存區(qū)中，最后從socket緩存區(qū)拷貝數(shù)據(jù)到網(wǎng)卡中發(fā)送。而Sendfile技術(shù)跳過(guò)了用戶進(jìn)程的datacache這一環(huán)節(jié)，直接讀取磁盤的數(shù)據(jù)到pagecache中，然后從pagecache拷貝一份到socket緩存區(qū)中，最后從socket緩存區(qū)拷貝數(shù)據(jù)到網(wǎng)卡中發(fā)送。整個(gè)過(guò)程減少了兩次拷貝消耗。

Kafka的節(jié)點(diǎn)間的數(shù)據(jù)一致性策略設(shè)計(jì)

對(duì)于任何多節(jié)點(diǎn)多備份的分布式系統(tǒng)而言，數(shù)據(jù)的一致性問(wèn)題都是繞不開(kāi)的難點(diǎn)，一般的選擇是要么優(yōu)先考慮效率，這樣可能就造成數(shù)據(jù)不一致甚至是數(shù)據(jù)丟失，要么選擇保障數(shù)據(jù)一致性和數(shù)據(jù)安全性犧牲效率。在kafka的身上也存在這樣的矛盾。

Kafka是一種分partion,多節(jié)點(diǎn)多備份的分布式系統(tǒng)，每個(gè)partion都可以存在多份備份，每個(gè)備份在不同的節(jié)點(diǎn)上。多個(gè)備份中會(huì)根據(jù)zookpeer的注冊(cè)信息通過(guò)算法選舉出其中一份作為leader,這個(gè)leader負(fù)責(zé)和客戶端的讀寫訪問(wèn)進(jìn)行交互。

其他備份不參與跟客戶端的交互。而是去跟leader partion交互同步數(shù)據(jù)。這樣一來(lái)就可能出現(xiàn)主備之間數(shù)據(jù)不一致的情況。Kafka在客戶端提供了一個(gè)配置選項(xiàng)props.put("acks", "all");--其中all表示生產(chǎn)者等待確認(rèn)所有的備份數(shù)據(jù)都寫入pagecache后再返回。

可以設(shè)置為0（不等待任何確認(rèn)），1（leader確認(rèn)）或者其他小于備份數(shù)的數(shù)字。其他備份節(jié)點(diǎn)會(huì)異步去同步leader partion的數(shù)據(jù)，保持一致，當(dāng)然如果在同步的過(guò)程中，leader partion出現(xiàn)數(shù)據(jù)丟失，那么這部分?jǐn)?shù)據(jù)將永遠(yuǎn)丟失。

Kafka的備份和負(fù)載均衡

Kafka的備份很明顯，上文已經(jīng)說(shuō)過(guò)是通過(guò)討論一致性問(wèn)題已經(jīng)交待清楚，至于Kafka的負(fù)載均衡，個(gè)人發(fā)現(xiàn)是嚴(yán)重依賴于zookeeper上的注冊(cè)信息，通過(guò)一套算法來(lái)選取leader partion來(lái)實(shí)現(xiàn)kafka多節(jié)點(diǎn)的負(fù)載均衡。

Zookeeper中保存了kafka幾乎一切的重要信息，比如topic，每個(gè)topic下面的多個(gè)partion信息，主機(jī)節(jié)點(diǎn)信息（包括ip和端口），每個(gè)節(jié)點(diǎn)下的多個(gè)partion信息，每個(gè)partion的主備份信息，消費(fèi)客戶端的group_id分組信息，每個(gè)消費(fèi)者信息等。

通過(guò)這一堆信息進(jìn)行算法計(jì)算最后得出負(fù)載均衡的方案，主要體現(xiàn)是選出讓kafka效率性能達(dá)到最好的每個(gè)partion的leader。并且在zookeeper中注冊(cè)監(jiān)視器，一旦發(fā)現(xiàn)上述信息有變動(dòng)則更新負(fù)載均衡方案。