1. 程式人生 > >深入理解 Kafka 副本機制

深入理解 Kafka 副本機制

一、Kafka叢集

Kafka使用Zookeeper來維護叢集成員(brokers)的資訊。每個broker都有一個唯一標識broker.id

,用於標識自己在叢集中的身份,可以在配置檔案server.properties中進行配置,或者由程式自動生成。下面是Kafka brokers叢集自動建立的過程:

  • 每一個broker啟動的時候,它會在Zookeeper的/brokers/ids路徑下建立一個臨時節點,並將自己的broker.id寫入,從而將自身註冊到叢集;
  • 當有多個broker時,所有broker會競爭性地在Zookeeper上建立/controller節點,由於Zookeeper上的節點不會重複,所以必然只會有一個broker建立成功,此時該broker稱為controller broker。它除了具備其他broker的功能外,還負責管理主題分割槽及其副本的狀態。
  • 當broker出現宕機或者主動退出從而導致其持有的Zookeeper會話超時時,會觸發註冊在Zookeeper上的watcher事件,此時Kafka會進行相應的容錯處理;如果宕機的是controller broker時,還會觸發新的controller選舉。

二、副本機制

為了保證高可用,kafka的分割槽是多副本的,如果一個副本丟失了,那麼還可以從其他副本中獲取分割槽資料。但是這要求對應副本的資料必須是完整的,這是Kafka資料一致性的基礎,所以才需要使用controller broker來進行專門的管理。下面將詳解介紹Kafka的副本機制。

2.1 分割槽和副本

Kafka 的主題被分為多個分割槽 ,分割槽是Kafka最基本的儲存單位。每個分割槽可以有多個副本(可以在建立主題時使用replication-factor

引數進行指定)。其中一個副本是首領副本(Leader replica),所有的事件都直接傳送給首領副本;其他副本是跟隨者副本(Follower replica),需要通過複製來保持與首領副本資料一致,當首領副本不可用時,其中一個跟隨者副本將成為新首領。

2.2 ISR機制

每個分割槽都有一個ISR(in-sync Replica)列表,用於維護所有同步的、可用的副本。首領副本必然是同步副本,而對於跟隨者副本來說,它需要滿足以下條件才能被認為是同步副本:

  • 與Zookeeper之間有一個活躍的會話,即必須定時向Zookeeper傳送心跳;
  • 在規定的時間內從首領副本那裡低延遲地獲取過訊息。

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

這裡給出一個主題建立的示例:使用--replication-factor指定副本系數為3,建立成功後使用--describe命令可以看到分割槽0的有0,1,2三個副本,且三個副本都在ISR列表中,其中1為首領副本。

2.3 不完全的首領選舉

對於副本機制,在broker級別有一個可選的配置引數unclean.leader.election.enable,預設值為fasle,代表禁止不完全的首領選舉。這是針對當首領副本掛掉且ISR中沒有其他可用副本時,是否允許某個不完全同步的副本成為首領副本,這可能會導致資料丟失或者資料不一致,在某些對資料一致性要求較高的場景(如金融領域),這可能無法容忍的,所以其預設值為false,如果你能夠允許部分資料不一致的話,可以配置為true。

2.4 最少同步副本

ISR機制的另外一個相關引數是min.insync.replicas , 可以在broker或者主題級別進行配置,代表ISR列表中至少要有幾個可用副本。這裡假設設定為2,那麼當可用副本數量小於該值時,就認為整個分割槽處於不可用狀態。此時客戶端再向分割槽寫入資料時候就會丟擲異常org.apache.kafka.common.errors.NotEnoughReplicasExceptoin: Messages are rejected since there are fewer in-sync replicas than required。

2.5 傳送確認

Kafka在生產者上有一個可選的引數ack,該引數指定了必須要有多少個分割槽副本收到訊息,生產者才會認為訊息寫入成功:

  • acks=0 :訊息傳送出去就認為已經成功了,不會等待任何來自伺服器的響應;
  • acks=1 : 只要叢集的首領節點收到訊息,生產者就會收到一個來自伺服器成功響應;
  • acks=all :只有當所有參與複製的節點全部收到訊息時,生產者才會收到一個來自伺服器的成功響應。

三、資料請求

3.1 元資料請求機制

在所有副本中,只有領導副本才能進行訊息的讀寫處理。由於不同分割槽的領導副本可能在不同的broker上,如果某個broker收到了一個分割槽請求,但是該分割槽的領導副本並不在該broker上,那麼它就會向客戶端返回一個Not a Leader for Partition的錯誤響應。 為了解決這個問題,Kafka提供了元資料請求機制。

首先叢集中的每個broker都會快取所有主題的分割槽副本資訊,客戶端會定期傳送傳送元資料請求,然後將獲取的元資料進行快取。定時重新整理元資料的時間間隔可以通過為客戶端配置metadata.max.age.ms來進行指定。有了元資料資訊後,客戶端就知道了領導副本所在的broker,之後直接將讀寫請求傳送給對應的broker即可。

如果在定時請求的時間間隔內發生的分割槽副本的選舉,則意味著原來快取的資訊可能已經過時了,此時還有可能會收到Not a Leader for Partition的錯誤響應,這種情況下客戶端會再次求發出元資料請求,然後重新整理本地快取,之後再去正確的broker上執行對應的操作,過程如下圖:

3.2 資料可見性

需要注意的是,並不是所有儲存在分割槽首領上的資料都可以被客戶端讀取到,為了保證資料一致性,只有被所有同步副本(ISR中所有副本)都儲存了的資料才能被客戶端讀取到。

3.3 零拷貝

Kafka所有資料的寫入和讀取都是通過零拷貝來實現的。傳統拷貝與零拷貝的區別如下:

傳統模式下的四次拷貝與四次上下文切換

以將磁碟檔案通過網路傳送為例。傳統模式下,一般使用如下虛擬碼所示的方法先將檔案資料讀入記憶體,然後通過Socket將記憶體中的資料傳送出去。

buffer = File.read
Socket.send(buffer)

這一過程實際上發生了四次資料拷貝。首先通過系統呼叫將檔案資料讀入到核心態Buffer(DMA拷貝),然後應用程式將記憶體態Buffer資料讀入到使用者態Buffer(CPU拷貝),接著使用者程式通過Socket傳送資料時將使用者態Buffer資料拷貝到核心態Buffer(CPU拷貝),最後通過DMA拷貝將資料拷貝到NIC Buffer。同時,還伴隨著四次上下文切換,如下圖所示:

sendfile和transferTo實現零拷貝

Linux 2.4+核心通過sendfile系統呼叫,提供了零拷貝。資料通過DMA拷貝到核心態Buffer後,直接通過DMA拷貝到NIC Buffer,無需CPU拷貝。這也是零拷貝這一說法的來源。除了減少資料拷貝外,因為整個讀檔案到網路傳送由一個sendfile呼叫完成,整個過程只有兩次上下文切換,因此大大提高了效能。零拷貝過程如下圖所示:

從具體實現來看,Kafka的資料傳輸通過TransportLayer來完成,其子類PlaintextTransportLayertransferFrom方法通過呼叫Java NIO中FileChannel的transferTo方法實現零拷貝,如下所示:

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

注: transferTotransferFrom並不保證一定能使用零拷貝。實際上是否能使用零拷貝與作業系統相關,如果作業系統提供sendfile這樣的零拷貝系統呼叫,則這兩個方法會通過這樣的系統呼叫充分利用零拷貝的優勢,否則並不能通過這兩個方法本身實現零拷貝。

四、物理儲存

4.1 分割槽分配

在建立主題時,Kafka會首先決定如何在broker間分配分割槽副本,它遵循以下原則:

  • 在所有broker上均勻地分配分割槽副本;
  • 確保分割槽的每個副本分佈在不同的broker上;
  • 如果使用了broker.rack引數為broker指定了機架資訊,那麼會盡可能的把每個分割槽的副本分配到不同機架的broker上,以避免一個機架不可用而導致整個分割槽不可用。

基於以上原因,如果你在一個單節點上建立一個3副本的主題,通常會丟擲下面的異常:

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

4.2 分割槽資料保留規則

保留資料是 Kafka 的一個基本特性, 但是Kafka不會一直保留資料,也不會等到所有消費者都讀取了訊息之後才刪除訊息。相反, Kafka為每個主題配置了資料保留期限,規定資料被刪除之前可以保留多長時間,或者清理資料之前可以保留的資料量大小。分別對應以下四個引數:

  • log.retention.bytes :刪除資料前允許的最大資料量;預設值-1,代表沒有限制;
  • log.retention.ms:儲存資料檔案的毫秒數,如果未設定,則使用log.retention.minutes中的值,預設為null;
  • log.retention.minutes:保留資料檔案的分鐘數,如果未設定,則使用log.retention.hours中的值,預設為null;
  • log.retention.hours:保留資料檔案的小時數,預設值為168,也就是一週。

因為在一個大檔案裡查詢和刪除訊息是很費時的,也很容易出錯,所以Kafka把分割槽分成若干個片段,當前正在寫入資料的片段叫作活躍片段。活動片段永遠不會被刪除。如果按照預設值保留資料一週,而且每天使用一個新片段,那麼你就會看到,在每天使用一個新片段的同時會刪除一個最老的片段,所以大部分時間該分割槽會有7個片段存在。

4.3 檔案格式

通常儲存在磁碟上的資料格式與生產者傳送過來訊息格式是一樣的。 如果生產者傳送的是壓縮過的訊息,那麼同一個批次的訊息會被壓縮在一起,被當作“包裝訊息”進行傳送(格式如下所示) ,然後儲存到磁碟上。之後消費者讀取後再自己解壓這個包裝訊息,獲取每條訊息的具體資訊。

參考資料

  1. Neha Narkhede, Gwen Shapira ,Todd Palino(著) , 薛命燈(譯) . Kafka權威指南 . 人民郵電出版社 . 2017-12-26
  2. Kafka高效能架構之道

更多大資料系列文章可以參見個人 GitHub 開源專案: 大資料入門指南