[[379096]]

在日常開發(fā)過程中使用kafka來實限流削峰作用但是往往kafka會存放多份副本來防止數(shù)據(jù)丟失，那你知道他的機(jī)制是什么樣的嗎?本篇文章就帶給大家講解下。

一、Kafka集群

Kafka 使用 Zookeeper 來維護(hù)集群成員 (brokers) 的信息。每個 broker 都有一個唯一標(biāo)識 broker.id，用于標(biāo)識自己在集群中的身份，可以在配置文件 server.properties 中進(jìn)行配置，或者由程序自動生成。下面是 Kafka brokers 集群自動創(chuàng)建的過程：

• 每一個 broker 啟動的時候，它會在 Zookeeper 的 /brokers/ids 路徑下創(chuàng)建一個 臨時節(jié)點，并將自己的 broker.id 寫入，從而將自身注冊到集群;
• 當(dāng)有多個 broker 時，所有 broker 會競爭性地在 Zookeeper 上創(chuàng)建 /controller 節(jié)點，由于 Zookeeper 上的節(jié)點不會重復(fù)，所以必然只會有一個 broker 創(chuàng)建成功，此時該 broker 稱為 controller broker。它除了具備其他 broker 的功能外，還負(fù)責(zé)管理主題分區(qū)及其副本的狀態(tài)。
• 當(dāng) broker 出現(xiàn)宕機(jī)或者主動退出從而導(dǎo)致其持有的 Zookeeper 會話超時時，會觸發(fā)注冊在 Zookeeper 上的 watcher 事件，此時 Kafka 會進(jìn)行相應(yīng)的容錯處理;如果宕機(jī)的是 controller broker 時，還會觸發(fā)新的 controller 選舉。

二、副本機(jī)制

為了保證高可用，kafka 的分區(qū)是多副本的，如果一個副本丟失了，那么還可以從其他副本中獲取分區(qū)數(shù)據(jù)。但是這要求對應(yīng)副本的數(shù)據(jù)必須是完整的，這是 Kafka 數(shù)據(jù)一致性的基礎(chǔ)，所以才需要使用 controller broker 來進(jìn)行專門的管理。下面將詳解介紹 Kafka 的副本機(jī)制。

2.1 分區(qū)和副本

Kafka 的主題被分為多個分區(qū) ，分區(qū)是 Kafka 最基本的存儲單位。每個分區(qū)可以有多個副本 (可以在創(chuàng)建主題時使用 replication-factor 參數(shù)進(jìn)行指定)。其中一個副本是首領(lǐng)副本 (Leader replica)，所有的事件都直接發(fā)送給首領(lǐng)副本;其他副本是跟隨者副本 (Follower replica)，需要通過復(fù)制來保持與首領(lǐng)副本數(shù)據(jù)一致，當(dāng)首領(lǐng)副本不可用時，其中一個跟隨者副本將成為新首領(lǐng)。

2.2 ISR機(jī)制

每個分區(qū)都有一個 ISR(in-sync Replica) 列表，用于維護(hù)所有同步的、可用的副本。首領(lǐng)副本必然是同步副本，而對于跟隨者副本來說，它需要滿足以下條件才能被認(rèn)為是同步副本：

• 與 Zookeeper 之間有一個活躍的會話，即必須定時向 Zookeeper 發(fā)送心跳;
• 在規(guī)定的時間內(nèi)從首領(lǐng)副本那里低延遲地獲取過消息。

如果副本不滿足上面條件的話，就會被從 ISR 列表中移除，直到滿足條件才會被再次加入。

這里給出一個主題創(chuàng)建的示例：使用 --replication-factor 指定副本系數(shù)為 3，創(chuàng)建成功后使用 --describe 命令可以看到分區(qū) 0 的有 0,1,2 三個副本，且三個副本都在 ISR 列表中，其中 1 為首領(lǐng)副本。

2.3 不完全的首領(lǐng)選舉

對于副本機(jī)制，在 broker 級別有一個可選的配置參數(shù) unclean.leader.election.enable，默認(rèn)值為 fasle，代表禁止不完全的首領(lǐng)選舉。這是針對當(dāng)首領(lǐng)副本掛掉且 ISR 中沒有其他可用副本時，是否允許某個不完全同步的副本成為首領(lǐng)副本，這可能會導(dǎo)致數(shù)據(jù)丟失或者數(shù)據(jù)不一致，在某些對數(shù)據(jù)一致性要求較高的場景 (如金融領(lǐng)域)，這可能無法容忍的，所以其默認(rèn)值為 false，如果你能夠允許部分?jǐn)?shù)據(jù)不一致的話，可以配置為 true。

2.4 最少同步副本

ISR 機(jī)制的另外一個相關(guān)參數(shù)是 min.insync.replicas , 可以在 broker 或者主題級別進(jìn)行配置，代表 ISR 列表中至少要有幾個可用副本。這里假設(shè)設(shè)置為 2，那么當(dāng)可用副本數(shù)量小于該值時，就認(rèn)為整個分區(qū)處于不可用狀態(tài)。此時客戶端再向分區(qū)寫入數(shù)據(jù)時候就會拋出異常 org.apache.kafka.common.errors.NotEnoughReplicasExceptoin: Messages are rejected since there are fewer in-sync replicas than required。

2.5 發(fā)送確認(rèn)

Kafka 在生產(chǎn)者上有一個可選的參數(shù) ack，該參數(shù)指定了必須要有多少個分區(qū)副本收到消息，生產(chǎn)者才會認(rèn)為消息寫入成功：

• acks=0 ：消息發(fā)送出去就認(rèn)為已經(jīng)成功了，不會等待任何來自服務(wù)器的響應(yīng);
• acks=1 ：只要集群的首領(lǐng)節(jié)點收到消息，生產(chǎn)者就會收到一個來自服務(wù)器成功響應(yīng);
• acks=all ：只有當(dāng)所有參與復(fù)制的節(jié)點全部收到消息時，生產(chǎn)者才會收到一個來自服務(wù)器的成功響應(yīng)。

三、數(shù)據(jù)請求

3.1 元數(shù)據(jù)請求機(jī)制

在所有副本中，只有領(lǐng)導(dǎo)副本才能進(jìn)行消息的讀寫處理。由于不同分區(qū)的領(lǐng)導(dǎo)副本可能在不同的 broker 上，如果某個 broker 收到了一個分區(qū)請求，但是該分區(qū)的領(lǐng)導(dǎo)副本并不在該 broker 上，那么它就會向客戶端返回一個 Not a Leader for Partition 的錯誤響應(yīng)。為了解決這個問題，Kafka 提供了元數(shù)據(jù)請求機(jī)制。

首先集群中的每個 broker 都會緩存所有主題的分區(qū)副本信息，客戶端會定期發(fā)送發(fā)送元數(shù)據(jù)請求，然后將獲取的元數(shù)據(jù)進(jìn)行緩存。定時刷新元數(shù)據(jù)的時間間隔可以通過為客戶端配置 metadata.max.age.ms 來進(jìn)行指定。有了元數(shù)據(jù)信息后，客戶端就知道了領(lǐng)導(dǎo)副本所在的 broker，之后直接將讀寫請求發(fā)送給對應(yīng)的 broker 即可。

如果在定時請求的時間間隔內(nèi)發(fā)生的分區(qū)副本的選舉，則意味著原來緩存的信息可能已經(jīng)過時了，此時還有可能會收到 Not a Leader for Partition 的錯誤響應(yīng)，這種情況下客戶端會再次求發(fā)出元數(shù)據(jù)請求，然后刷新本地緩存，之后再去正確的 broker 上執(zhí)行對應(yīng)的操作，過程如下圖：

3.2 數(shù)據(jù)可見性

需要注意的是，并不是所有保存在分區(qū)首領(lǐng)上的數(shù)據(jù)都可以被客戶端讀取到，為了保證數(shù)據(jù)一致性，只有被所有同步副本 (ISR 中所有副本) 都保存了的數(shù)據(jù)才能被客戶端讀取到。

3.3 零拷貝

Kafka 所有數(shù)據(jù)的寫入和讀取都是通過零拷貝來實現(xiàn)的。傳統(tǒng)拷貝與零拷貝的區(qū)別如下：

傳統(tǒng)模式下的四次拷貝與四次上下文切換

以將磁盤文件通過網(wǎng)絡(luò)發(fā)送為例。傳統(tǒng)模式下，一般使用如下偽代碼所示的方法先將文件數(shù)據(jù)讀入內(nèi)存，然后通過 Socket 將內(nèi)存中的數(shù)據(jù)發(fā)送出去。

buffer = File.read 
Socket.send(buffer) 

這一過程實際上發(fā)生了四次數(shù)據(jù)拷貝。首先通過系統(tǒng)調(diào)用將文件數(shù)據(jù)讀入到內(nèi)核態(tài) Buffer(DMA 拷貝)，然后應(yīng)用程序?qū)?nèi)存態(tài) Buffer 數(shù)據(jù)讀入到用戶態(tài) Buffer(CPU 拷貝)，接著用戶程序通過 Socket 發(fā)送數(shù)據(jù)時將用戶態(tài) Buffer 數(shù)據(jù)拷貝到內(nèi)核態(tài) Buffer(CPU 拷貝)，最后通過 DMA 拷貝將數(shù)據(jù)拷貝到 NIC Buffer。同時，還伴隨著四次上下文切換，如下圖所示：

sendfile和transferTo實現(xiàn)零拷貝

Linux 2.4+ 內(nèi)核通過 sendfile 系統(tǒng)調(diào)用，提供了零拷貝。數(shù)據(jù)通過 DMA 拷貝到內(nèi)核態(tài) Buffer 后，直接通過 DMA 拷貝到 NIC Buffer，無需 CPU 拷貝。這也是零拷貝這一說法的來源。除了減少數(shù)據(jù)拷貝外，因為整個讀文件到網(wǎng)絡(luò)發(fā)送由一個 sendfile 調(diào)用完成，整個過程只有兩次上下文切換，因此大大提高了性能。零拷貝過程如下圖所示：圖片 從具體實現(xiàn)來看，Kafka 的數(shù)據(jù)傳輸通過 TransportLayer 來完成，其子類 PlaintextTransportLayer 的 transferFrom 方法通過調(diào)用 Java NIO 中 FileChannel 的 transferTo 方法實現(xiàn)零拷貝，如下所示：

@Override 
public long transferFrom(FileChannel fileChannel, long position, long count) throws IOException { 
    return fileChannel.transferTo(position, count, socketChannel); 
} 

注： transferTo 和 transferFrom 并不保證一定能使用零拷貝。實際上是否能使用零拷貝與操作系統(tǒng)相關(guān)，如果操作系統(tǒng)提供 sendfile 這樣的零拷貝系統(tǒng)調(diào)用，則這兩個方法會通過這樣的系統(tǒng)調(diào)用充分利用零拷貝的優(yōu)勢，否則并不能通過這兩個方法本身實現(xiàn)零拷貝。

四、物理存儲

4.1 分區(qū)分配

在創(chuàng)建主題時，Kafka 會首先決定如何在 broker 間分配分區(qū)副本，它遵循以下原則：

• 在所有 broker 上均勻地分配分區(qū)副本;
• 確保分區(qū)的每個副本分布在不同的 broker 上;
• 如果使用了 broker.rack 參數(shù)為 broker 指定了機(jī)架信息，那么會盡可能的把每個分區(qū)的副本分配到不同機(jī)架的 broker 上，以避免一個機(jī)架不可用而導(dǎo)致整個分區(qū)不可用。

基于以上原因，如果你在一個單節(jié)點上創(chuàng)建一個 3 副本的主題，通常會拋出下面的異常：

Error while executing topic command : org.apache.kafka.common.errors.InvalidReplicationFactor    
Exception: Replication factor: 3 larger than available brokers: 1. 

4.2 分區(qū)數(shù)據(jù)保留規(guī)則

保留數(shù)據(jù)是 Kafka 的一個基本特性， 但是 Kafka 不會一直保留數(shù)據(jù)，也不會等到所有消費者都讀取了消息之后才刪除消息。相反， Kafka 為每個主題配置了數(shù)據(jù)保留期限，規(guī)定數(shù)據(jù)被刪除之前可以保留多長時間，或者清理數(shù)據(jù)之前可以保留的數(shù)據(jù)量大小。分別對應(yīng)以下四個參數(shù)：

• log.retention.bytes ：刪除數(shù)據(jù)前允許的最大數(shù)據(jù)量;默認(rèn)值-1，代表沒有限制;
• log.retention.ms：保存數(shù)據(jù)文件的毫秒數(shù)，如果未設(shè)置，則使用 log.retention.minutes 中的值，默認(rèn)為 null;
• log.retention.minutes：保留數(shù)據(jù)文件的分鐘數(shù)，如果未設(shè)置，則使用 log.retention.hours 中的值，默認(rèn)為 null;
• log.retention.hours：保留數(shù)據(jù)文件的小時數(shù)，默認(rèn)值為 168，也就是一周。

因為在一個大文件里查找和刪除消息是很費時的，也很容易出錯，所以 Kafka 把分區(qū)分成若干個片段，當(dāng)前正在寫入數(shù)據(jù)的片段叫作活躍片段?；顒悠斡肋h(yuǎn)不會被刪除。如果按照默認(rèn)值保留數(shù)據(jù)一周，而且每天使用一個新片段，那么你就會看到，在每天使用一個新片段的同時會刪除一個最老的片段，所以大部分時間該分區(qū)會有 7 個片段存在。

4.3 文件格式

通常保存在磁盤上的數(shù)據(jù)格式與生產(chǎn)者發(fā)送過來消息格式是一樣的。如果生產(chǎn)者發(fā)送的是壓縮過的消息，那么同一個批次的消息會被壓縮在一起，被當(dāng)作“包裝消息”進(jìn)行發(fā)送 (格式如下所示) ，然后保存到磁盤上。之后消費者讀取后再自己解壓這個包裝消息，獲取每條消息的具體信息。

小結(jié)

本篇文章講解了關(guān)于kafka的存放副本的機(jī)制的原理，以及數(shù)據(jù)是如何存儲的kafka為了防止數(shù)據(jù)丟失添加了ack的方式，這個ack可能會影響一些效率，這ack的值可以根據(jù)場景進(jìn)行設(shè)置比如說丟失一些數(shù)據(jù)沒有問題那就設(shè)置為0我將消息發(fā)出去我就不管了。我在這里為大家提供大數(shù)據(jù)的資料需要的朋友可以去下面GitHub去下載，信自己，努力和汗水總會能得到回報的。我是大數(shù)據(jù)老哥，我們下期見~~~

本文轉(zhuǎn)載自微信公眾號「大數(shù)據(jù)老哥」，可以通過以下二維碼關(guān)注。轉(zhuǎn)載本文請聯(lián)系大數(shù)據(jù)老哥公眾號。