淺談Kafka特性與架構

Kafka特性

kafka做爲一種消息中間件，有如下特性：

• 高吞吐量：吞吐量高達數十萬
• 高併發：支持數千個客戶端同時讀寫
• 低延遲：延遲最低只有幾毫秒
• 消息持久性和可靠性：消息被持久化到本地磁盤，同時支持數據備份
• 集羣容錯性：容許n-1個節點失敗（n爲副本個數）
• 可擴展性：支持集羣動態擴展

應用場景

根據Kafka的特性，有如下應用場景：

• 消息中間件：kafka自己做爲標準的消息中間件，能夠用於producer和consumer之間的異步消息通訊
• 日誌：出於kafka的高吞吐量特性，能夠進行高效地日誌收集
• 數據收集：出於高吞吐量和高併發特性，可使用kafka記錄用戶/系統的一些實時數據

主要名詞

• Broker：每一臺Kafka服務器就叫作一個Broker，支持水平擴展，一個集羣中一般有多臺Broker，各個Broker地位一致，不存在主從關係
• Coordinator：集羣的協調者，kafka會將負載最小的broker指定爲Coordinator
• Topic：全部消息都有本身的所屬分類，這個分類就叫作Topic。一個Topic下的消息能夠保存在多個Broker上（對於Producer和Consumer是無感知的）
• Producer：產生消息的主體叫作Producer，負責發佈消息到指定Topic中
• Consumer：消費對象的主體叫作Consumer，負責消費指定Topic中的消息
• ConsumerGroup（CG）：每個Consumer均屬於一個特定CG，一個Topic能夠對應多個CG，Topic的消息會發送到全部CG，可是CG能夠選擇發送給全部Consumer仍是指定的Consumer，經過這種方式能夠方便的實現單播和廣播。同時，同一個CG下的Consumer能夠實現負載均衡
• Partition：存放數據的具體物理實體，每個Topic會分爲多個Partition。每個Partition對應一個文件夾，在文件夾下存放數據和索引文件。每個Partition中的消息是有序的，可是不一樣Partition的數據不能肯定順序
• Replication：Partition的備份，一個Partition會有多個Replication，存放在不一樣的Broker上
• Segment：指每個數據文件，一個Partition對應多個Segment，每個Segment會有一個索引文件與之對應
• Offset：指消息的序列號，是連續遞增的，Partition中的每個消息都會有本身的Offset，用於惟一標識一條消息。由於是有序的，因此能夠根據Offset快速定位一個數據文件

基本架構

kafka是一個自然支持分佈式架構的發佈訂閱模式的rpc通訊框架，kafka集羣爲典型的去中心化的設計，主體設計以下：

生產者向Kafka集羣提供數據，消費者從Kafka集羣拉取數據，Kafka集羣的調度由Zookeeper負責

Zookeeper

Kafka集羣的元數據保存在Zookeeper中，除此以外不存儲任何消息數據。每個Broker都須要在Zookeeper上註冊並不斷在上面更新本身的元數據（Topic和Partition信息），Zookeeper會使用這些數據信息來實現動態的集羣擴容

Producer和Consumer都會在Zookeeper上註冊監聽器（Watcher），用於在Zookeeper發生變化時做出響應的調整。同時，Consumer還會向Zookeeper中註冊本身消費的Partition列表，用於發現Broker並與Partition創建socket鏈接

核心組件

Partition

Kafka中的Topic是以Partition的形式存放的，一個Topic會被拆分爲多個Partition，存放在多臺服務器上。Producer在生產數據時會根據必定的規則將數據寫入指定Topic下的Partition中

能夠設置每個Topic的Partition數量，可是須要注意的是，一個Partition只能供一個Consumer消費，若是Partition過少，就可能會有Consumer消費不到數據。另外，建議partition的數量也須要大於集羣中Broker的數量，這樣可讓Partition Leader儘可能均勻地分佈在各個Broker中。同時也須要注意，拆分的Partiton越多，也就意味着須要更多的空間

一般一個Partition須要有數個副本（Replication），Kafka容許用戶設置一份數據的備份個數，副本會存儲在不一樣的Broker上。在全部的副本中（包括本身），會存在一個Partition Leader用於進行讀寫，Leader的選舉調度等操做由Zookeeper來完成

Producer

Producer直接將消息發送到Broker的Partition Leader上，不須要通過代理中轉等操做，由於在設計時，Kafka集羣中的每個Broker均可以單獨響應Producer的操做，並返回Topic的一些信息（存活的機器/Leader位置/...）

Producer客戶端負責採用指定的負載均衡算法，管理消息會被推送到哪些Partition上。同時Producer能夠將消息在內存中累計到必定數量時，做爲一個Batch進行發送，可以有效減小IO次數，進而提升效率。具體的Batch參數能夠手動設置，能夠是累計的數量大小/時間間隔等

Producer能夠異步地向Kafka發送數據，在發送後會收到一個Futrue響應，包含offset值等信息。能夠經過指定acks參數來控制Producer要求收到的確認消息個數

• acks參數爲n時：只有當n個partition副本收到消息後，producer纔會收到broker的確認
• acks參數爲-1時：producer會在全部partition副本收到消息後獲得broker的確認
• acks參數爲0時：producer不會等待broker的響應，能夠獲得最大的吞吐量，可是可能會致使數據丟失

Consumer

Kafka中，讀取消息的offset值由Consumer進行維護，所以consumer能夠自由選取讀取消息的方式。同時，無論消息有沒有被消費，數據都會在kafka中保存一段時間

Kafka提供了兩種consumer api，分別是high-level api和sample api。Sample api只維持了和單一Broker的鏈接，同時是無狀態的，每次請求都須要指定offset值，因此也更爲靈活

High-Level api封裝了對集羣中broker的訪問，能夠透明的訪問一個topic，同時也維持了已消費消息的狀態，每次消費的都是下一個消息。High-Level api還支持以組（CG）的形式消費消息，消息會被髮送給全部的CG，CG內部會選擇按順序發送給全部Consumer或是指定的Consumer

核心機制

消息壓縮

Kafka能夠以集合（batch）形式發送數據，在此基礎上，kafka能夠對batch進行壓縮。在producer端進行壓縮後，在consumer進行解壓，減小了傳輸所需的數據量，減輕對網絡的壓力。kafka在消息頭部增長了一個字節用於描述壓縮屬性，這個字節後兩位表示壓縮採用的編碼，若是後兩位爲0，表示消息未被壓縮

消息可靠性

最理想的狀況是消息發送成功，而且只發送了一次，這種狀況叫作exactly-once，可是不可避免的會發生消息發送失敗以及消息重複發送的狀況

爲了解決這類問題，在producer端，當一個消息被髮送後，producer會等待broker發送響應，收到響應後producer會確認消息已經被正確發送給kafka，不然就會從新發送

在consumer端，由於broker記錄了partition中的offset值，這個值指向consumer下一個消費的消息，若是consumer收到消息可是消費失敗，broker能夠根據offset值來找到上一個消息，同時consumer還能夠控制offset值，來對消息進行任意處理

備份機制

（在「核心組件-Partition」中已經對此部分作了敘述）

消息消費策略

消費策略分類

固定分區消費

consumer在進行消息消費時，能夠指定消息某分區的消息

Rebalance分區消費

通常地，一個topic下會有多個partition，而一個partition只能被一個CG中的consumer消費，能夠經過指定rebalance策略，來採用不一樣的消費方式。Rebalance策略有兩種，範圍分區（Range）和輪詢分區（RoundRobin），範圍分區策略，即對topic下的partition進行排序，將partition數量除以CG下的consumer數量，從而得出每個consumer消費哪幾個分區

輪詢分區策略則是將partition按照hashcode進行排序，而後經過分區取模來給consumer分配partition

Rebalance的觸發時機

當如下三種狀況發生時，會觸發rebalance操做，從新指定分區：

• CG內部加入了新的consumer
• consumer離開CG
• topic新增partition

Rebalance的執行過程

rebalance的執行由CG Leader來完成，並負責在執行結束後將執行結果經過broker集羣中的coordinator廣播到CG。當CG的第一個consumer啓動後，這個consumer會和kafka肯定組內的coordinator，以後CG內的全部成員都會和該coordinator進行通訊

CG Leader的選舉有兩個階段，Join Group和Synchronizing Group State。

1. Join Group階段，全部成員都會向coordinator發送JoinGroup請求，當全部consumer都發送請求後， coordinator會選擇一個consumer擔任leader，並把CG的信息發送給該leader
2. Synchronizing Group State階段，全部consumer都會向coordinator發送SynchronizingGroupState請求，而leader則將分區方案發送給coordinator，coordinator會在接受到分區方案後，將分區結果返回給全部consumer，這樣就完成分區方案的同步

高效性設計

消息持久化

消息的持久化並不只僅是出於數據備份的須要，一個事實是，線性讀寫的時間遠遠高於隨機讀寫，對磁盤的線性讀所消耗的時間在有些狀況下能夠比內存的隨機訪問更快，因此現代不少操做系統會把空閒的內存用做磁盤緩存，儘管會在內存回收時帶來性能損耗，可是在讀寫上帶來的效率提高是顯著的

基於這樣的事實，利用文件系統依靠頁緩存來維護數據，會比維護一個內存緩存更好，由於採用了更爲緊湊的數據結構。不一樣於維護儘量多的內存緩存，若是咱們將數據寫入到一個持久化日誌中，不調用刷新程序，這意味着數據將被傳輸到內核中並在稍後被刷新，咱們也能夠經過配置來控制數據在何時刷新到物理磁盤上

常數時間的保證

kafka中持久化消息隊列採用對文件的讀寫來實現，相似日誌的形式。儘管這種操做不支持豐富的語義，可是能夠很高效的進行並行操做，而且全部的操做都是常數時間，最終系統的性能和數據大小徹底無關，能夠充分利用硬盤來進行高效的消息服務

字節拷貝

爲了解決字節拷貝的問題，kafka採用「標準字節消息」這種消息格式，這種格式在producer、consumer和broker間共享，kafka的日誌文件都是按「標準字節消息」這種格式寫入磁盤中。unix系統爲了提升頁面緩存和socket之間的數據傳遞效率，使用了「零拷貝」機制，即sendfile system call 系統調用，java中也提供了訪問這個系統調用的接口

爲了解釋爲何這種方式能解決字節拷貝帶來的性能損耗，咱們先來描述將數據從文件發送到socket的通常步驟：

1. os將數據從磁盤讀到內核空間的頁緩存中
2. 應用將數據從內核空間讀到用戶空間的頁緩存中
3. 應用將數據寫回內核空間的socket緩存中
4. os將數據從socket緩存寫到網卡緩存中
5. 數據經網絡發出

咱們能夠發現這個過程至少涉及4次字節拷貝，2次系統調用，2次內核態到用戶態的切換，而若是咱們可以直接將數據寫入socket緩存中，就能減小不少沒必要要的切換。若是使用了sendfile的方式，數據能夠直接由內核頁緩存直接拷貝到內核socket緩存中，不須要進行額外的系統狀態切換。經過這種方式，即便下游有不少consumer，也不會對集羣服務形成壓力

想更詳細瞭解零拷貝機制的可見個人另外一篇文章：淺談零拷貝機制

頻繁小IO

頻繁的小io能夠經過一次性發送一個消息集合，而不是隻發送一條消息來解決，消息在服務器以消息塊的形式添加到日誌中。同時consumer在查詢時也會一次查詢大量的線性數據塊。消息集合（Message Set）將一個字節數組或文件進行打包，同時能夠有選擇地進行反序列化