你都知道那些Kafka副本機(jī)制？

作者： 大數(shù)據(jù)老哥 2021-01-28 18:52:57

開(kāi)發(fā)

架構(gòu)

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

我們提供的服務(wù)有：成都網(wǎng)站設(shè)計(jì)、做網(wǎng)站、微信公眾號(hào)開(kāi)發(fā)、網(wǎng)站優(yōu)化、網(wǎng)站認(rèn)證、高青ssl等。為成百上千家企事業(yè)單位解決了網(wǎng)站和推廣的問(wèn)題。提供周到的售前咨詢和貼心的售后服務(wù)，是有科學(xué)管理、有技術(shù)的高青網(wǎng)站制作公司

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

一、Kafka集群

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

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

二、副本機(jī)制

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

2.1 分區(qū)和副本

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

2.2 ISR機(jī)制

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

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

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

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

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

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

2.4 最少同步副本

ISR 機(jī)制的另外一個(gè)相關(guān)參數(shù)是 min.insync.replicas , 可以在 broker 或者主題級(jí)別進(jìn)行配置，代表 ISR 列表中至少要有幾個(gè)可用副本。這里假設(shè)設(shè)置為 2，那么當(dāng)可用副本數(shù)量小于該值時(shí)，就認(rèn)為整個(gè)分區(qū)處于不可用狀態(tài)。此時(shí)客戶端再向分區(qū)寫入數(shù)據(jù)時(shí)候就會(huì)拋出異常 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)者上有一個(gè)可選的參數(shù) ack，該參數(shù)指定了必須要有多少個(gè)分區(qū)副本收到消息，生產(chǎn)者才會(huì)認(rèn)為消息寫入成功：

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

三、數(shù)據(jù)請(qǐng)求

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

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

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

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

3.2 數(shù)據(jù)可見(jiàn)性

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

3.3 零拷貝

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

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

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

  
 
 
 

   
  
  
  
buffer = File.read
   
  
  
  
Socket.send(buffer)
  
 
 
 

這一過(guò)程實(shí)際上發(fā)生了四次數(shù)據(jù)拷貝。首先通過(guò)系統(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 拷貝)，接著用戶程序通過(guò) Socket 發(fā)送數(shù)據(jù)時(shí)將用戶態(tài) Buffer 數(shù)據(jù)拷貝到內(nèi)核態(tài) Buffer(CPU 拷貝)，最后通過(guò) DMA 拷貝將數(shù)據(jù)拷貝到 NIC Buffer。同時(shí)，還伴隨著四次上下文切換，如下圖所示：

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

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

  
 
 
 

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

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

四、物理存儲(chǔ)

4.1 分區(qū)分配

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

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

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

  
 
 
 

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

• log.retention.bytes ：刪除數(shù)據(jù)前允許的最大數(shù)據(jù)量;默認(rèn)值-1，代表沒(méi)有限制;
• 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í)數(shù)，默認(rèn)值為 168，也就是一周。

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

4.3 文件格式

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

小結(jié)

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

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

本文標(biāo)題：你都知道那些Kafka副本機(jī)制？
網(wǎng)站地址：http://m.fisionsoft.com.cn/article/cophssc.html