Kafka消息隊(duì)列原理

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

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

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

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

Kafka的分發(fā)策略

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

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

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

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

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

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

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

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

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

Kafka的持久化設(shè)計(jì)是非常有特色的，和其他分布式系統(tǒng)不同，它沒有自己維護(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)需要持久化時再調(diào)用操作系統(tǒng)的文件系統(tǒng)寫入到文件中，這樣就多了一次datacache到pagecache的拷貝消耗。這樣的話，kafka的持久化管理關(guān)鍵是管理文件系統(tǒng)的pagecache的刷盤。

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

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

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

但是實(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á)到一定容量時刷新到磁盤的data文件中。

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

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

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

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

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

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

其他備份不參與跟客戶端的交互。而是去跟leader partion交互同步數(shù)據(jù)。這樣一來就可能出現(xiàn)主備之間數(shù)據(jù)不一致的情況。Kafka在客戶端提供了一個配置選項(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)會異步去同步leader partion的數(shù)據(jù)，保持一致，當(dāng)然如果在同步的過程中，leader partion出現(xiàn)數(shù)據(jù)丟失，那么這部分?jǐn)?shù)據(jù)將永遠(yuǎn)丟失。

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

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

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

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

Kafka消息隊(duì)列內(nèi)部實(shí)現(xiàn)原理

以上為個人經(jīng)驗(yàn)，希望能給大家一個參考，也希望大家多多支持腳本之家。