點選標題下「非同步社群」可快速關注

Kafka 由於高吞吐量、可持久化、分散式、支援流資料處理等特性而被廣泛應用。但當前關於Kafka原理及應用的相關資料較少，在我打算編寫本文時，還沒有見到中文版本的Kafka相關書籍，對於初學者甚至是一些中高階應用者來說學習成本還是比較高的，因此我打算在對Kafka進行深入而系統的研究基礎上，結合自己在工作中的實踐經驗，編寫一本介紹Kafka原理及其基本應用的書籍，以幫助Kafka初、中、高階應用者更快、更好地全面掌握Kafka的基礎理論及其基本應用，從而解決實際業務中的問題。同時，一直以來我都考慮在技術方面寫點什麼，將自己所學、所積累的知識沉澱下來。

1.1　Kafka背景

隨著資訊科技的快速發展及網際網路使用者規模的急劇增長，計算機所儲存的資訊量正呈爆炸式增長，目前資料量已進入大規模和超大規模的海量資料時代，如何高效地儲存、分析、處理和挖掘海量資料已成為技術研究領域的熱點和難點問題。當前出現的雲端儲存、分散式儲存系統、NoSQL資料庫及列儲存等前沿技術在海量資料的驅使下，正日新月異地向前發展，採用這些技術來處理大資料成為一種發展趨勢。而如何採集和運營管理、分析這些資料也是大資料處理中一個至關重要的組成環節，這就需要相應的基礎設施對其提供支援。針對這個需求，當前業界已有很多開源的訊息系統應運而生，本書介紹的Kafka就是當前流行的一款非常優秀的訊息系統。

Kafka 是一款開源的、輕量級的、分散式、可分割槽和具有複製備份的（Replicated）、基於ZooKeeper 協調管理的分散式流平臺的功能強大的訊息系統。與傳統的訊息系統相比，Kafka能夠很好地處理活躍的流資料，使得資料在各個子系統中高效能、低延遲地不停流轉。

據Kafka官方網站介紹，Kafka定位就是一個分散式流處理平臺。在官方看來，作為一個流式處理平臺，必須具備以下3個關鍵特性。

• 能夠允許釋出和訂閱流資料。從這個角度來講，平臺更像一個訊息佇列或者企業級的訊息系統。

• 儲存流資料時提供相應的容錯機制。

• 當流資料到達時能夠被及時處理。

Kafka能夠很好滿足以上3個特性，通過Kafka能夠很好地建立實時流式資料通道，由該通道可靠地獲取系統或應用程式的資料，也可以通過Kafka方便地構建實時流資料應用來轉換或是對流式資料進行響應處理。特別是在0.10版本之後，Kafka推出了Kafka Streams，這讓Kafka對流資料處理變得更加方便。

Kafka已釋出多個版本。截止到編寫本書時，Kafka的最新版本為0.10.1.1，因此本書內容都是基於該版本進行講解。

1.2　Kafka基本結構

通過前面對Kafka背景知識的簡短介紹，我們對Kafka是什麼有了初步的瞭解，本節我們將進一步介紹Kafka作為訊息系統的基本結構。我們知道，作為一個訊息系統，其基本結構中至少要有產生訊息的元件（訊息生產者，Producer）以及消費訊息的元件（消費者，Consumer）。雖然消費者並不是必需的，但離開了消費者構建一個訊息系統終究是毫無意義的。Kafka訊息系統最基本的體系結構如圖1-1所示。

圖1-1　Kafka訊息系統最基本的體系結構

生產者負責生產訊息，將訊息寫入Kafka叢集；消費者從Kafka叢集中拉取訊息。至於生產者如何將生產的訊息寫入 Kafka，消費者如何從 Kafka 叢集消費訊息，Kafka 如何儲存訊息，Kafka 叢集如何管理排程，如何進行訊息負載均衡，以及各元件間如何進行通訊等諸多問題，我們將在後續章節進行詳細闡述，在本節我們只需對Kafka基本結構輪廓有個清晰認識即可。隨著對Kafka相關知識的深入學習，我們將逐步對Kafka的結構圖進行完善。

1.3　Kafka基本概念

在對Kafka基本體系結構有了一定了解後，本節我們對Kafka的基本概念進行詳細闡述。

1．主題

Kafka將一組訊息抽象歸納為一個主題（Topic），也就是說，一個主題就是對訊息的一個分類。生產者將訊息傳送到特定主題，消費者訂閱主題或主題的某些分割槽進行消費。

2．訊息

訊息是Kafka通訊的基本單位，由一個固定長度的訊息頭和一個可變長度的訊息體構成。在老版本中，每一條訊息稱為Message；在由Java重新實現的客戶端中，每一條訊息稱為Record。

3．分割槽和副本

Kafka將一組訊息歸納為一個主題，而每個主題又被分成一個或多個分割槽（Partition）。每個分割槽由一系列有序、不可變的訊息組成，是一個有序佇列。

每個分割槽在物理上對應為一個資料夾，分割槽的命名規則為主題名稱後接“—”連線符，之後再接分割槽編號，分割槽編號從0開始，編號最大值為分割槽的總數減1。每個分割槽又有一至多個副本（Replica），分割槽的副本分佈在叢集的不同代理上，以提高可用性。從儲存角度上分析，分割槽的每個副本在邏輯上抽象為一個日誌（Log）物件，即分割槽的副本與日誌物件是一一對應的。每個主題對應的分割槽數可以在Kafka啟動時所載入的配置檔案中配置，也可以在建立主題時指定。當然，客戶端還可以在主題建立後修改主題的分割槽數。

分割槽使得Kafka在併發處理上變得更加容易，理論上來說，分割槽數越多吞吐量越高，但這要根據叢集實際環境及業務場景而定。同時，分割槽也是Kafka保證訊息被順序消費以及對訊息進行負載均衡的基礎。

Kafka只能保證一個分割槽之內訊息的有序性，並不能保證跨分割槽訊息的有序性。每條訊息被追加到相應的分割槽中，是順序寫磁碟，因此效率非常高，這是Kafka高吞吐率的一個重要保證。同時與傳統訊息系統不同的是，Kafka並不會立即刪除已被消費的訊息，由於磁碟的限制訊息也不會一直被儲存（事實上這也是沒有必要的），因此Kafka提供兩種刪除老資料的策略，一是基於訊息已儲存的時間長度，二是基於分割槽的大小。這兩種策略都能通過配置檔案進行配置，在這裡不展開探討，在3.5.4節將詳細介紹。

4．Leader副本和Follower副本

由於Kafka副本的存在，就需要保證一個分割槽的多個副本之間資料的一致性，Kafka會選擇該分割槽的一個副本作為Leader副本，而該分割槽其他副本即為Follower副本，只有Leader副本才負責處理客戶端讀/寫請求，Follower副本從Leader副本同步資料。如果沒有Leader副本，那就需要所有的副本都同時負責讀/寫請求處理，同時還得保證這些副本之間資料的一致性，假設有n個副本則需要有n×n條通路來同步資料，這樣資料的一致性和有序性就很難保證。

引入Leader副本後客戶端只需與Leader副本進行互動，這樣資料一致性及順序性就有了保證。Follower副本從Leader副本同步訊息，對於n個副本只需n−1條通路即可，這樣就使得系統更加簡單而高效。副本Follower與Leader的角色並不是固定不變的，如果Leader失效，通過相應的選舉演算法將從其他Follower副本中選出新的Leader副本。

5．偏移量

任何釋出到分割槽的訊息會被直接追加到日誌檔案（分割槽目錄下以“.log”為檔名字尾的資料檔案）的尾部，而每條訊息在日誌檔案中的位置都會對應一個按序遞增的偏移量。偏移量是一個分割槽下嚴格有序的邏輯值，它並不表示訊息在磁碟上的物理位置。由於Kafka幾乎不允許對訊息進行隨機讀寫，因此Kafka並沒有提供額外索引機制到儲存偏移量，也就是說並不會給偏移量再提供索引。消費者可以通過控制訊息偏移量來對訊息進行消費，如消費者可以指定消費的起始偏移量。為了保證訊息被順序消費，消費者已消費的訊息對應的偏移量也需要儲存。需要說明的是，消費者對訊息偏移量的操作並不會影響訊息本身的偏移量。舊版消費者將消費偏移量儲存到ZooKeeper當中，而新版消費者是將消費偏移量儲存到Kafka內部一個主題當中。當然，消費者也可以自己在外部系統儲存消費偏移量，而無需儲存到Kafka中。

6．日誌段

一個日誌又被劃分為多個日誌段（LogSegment），日誌段是Kafka日誌物件分片的最小單位。與日誌物件一樣，日誌段也是一個邏輯概念，一個日誌段對應磁碟上一個具體日誌檔案和兩個索引檔案。日誌檔案是以“.log”為檔名字尾的資料檔案，用於儲存訊息實際資料。兩個索引檔案分別以“.index”和“.timeindex”作為檔名字尾，分別表示訊息偏移量索引檔案和訊息時間戳索引檔案。

7．代理

在Kafka基本體系結構中我們提到了Kafka叢集。Kafka叢集就是由一個或多個Kafka例項構成，我們將每一個Kafka例項稱為代理（Broker），通常也稱代理為Kafka伺服器（KafkaServer）。在生產環境中Kafka叢集一般包括一臺或多臺伺服器，我們可以在一臺伺服器上配置一個或多個代理。每一個代理都有唯一的標識id，這個id是一個非負整數。在一個Kafka叢集中，每增加一個代理就需要為這個代理配置一個與該叢集中其他代理不同的id，id值可以選擇任意非負整數即可，只要保證它在整個Kafka叢集中唯一，這個id就是代理的名字，也就是在啟動代理時配置的broker.id對應的值，因此在本書中有時我們也稱為brokerId。由於給每個代理分配了不同的brokerId，這樣對代理進行遷移就變得更方便，從而對消費者來說是透明的，不會影響消費者對訊息的消費。代理有很多個引數配置，由於在本節只是對其概念進行闡述，因此不做深入展開，對於代理相關配置將穿插在本書具體元件實現原理、流程分析及相關實戰操作章節進行介紹。

8．生產者

生產者（Producer）負責將訊息傳送給代理，也就是向Kafka代理髮送訊息的客戶端。

9．消費者和消費組

消費者（Comsumer）以拉取（pull）方式拉取資料，它是消費的客戶端。在Kafka中每一個消費者都屬於一個特定消費組（ConsumerGroup），我們可以為每個消費者指定一個消費組，以groupId代表消費組名稱，通過group.id配置設定。如果不指定消費組，則該消費者屬於預設消費組test-consumer-group。同時，每個消費者也有一個全域性唯一的id，通過配置項client.id指定，如果客戶端沒有指定消費者的id，Kafka會自動為該消費者生成一個全域性唯一的id，格式為${groupId}-${hostName}-${timestamp}-${UUID前8位字元}。同一個主題的一條訊息只能被同一個消費組下某一個消費者消費，但不同消費組的消費者可同時消費該訊息。消費組是Kafka用來實現對一個主題訊息進行廣播和單播的手段，實現訊息廣播只需指定各消費者均屬於不同的消費組，訊息單播則只需讓各消費者屬於同一個消費組。

10．ISR

Kafka在ZooKeeper中動態維護了一個ISR（In-sync Replica），即儲存同步的副本列表，該列表中儲存的是與Leader副本保持訊息同步的所有副本對應的代理節點id。如果一個Follower副本宕機（本書用宕機來特指某個代理失效的情景，包括但不限於代理被關閉，如代理被人為關閉或是發生物理故障、心跳檢測過期、網路延遲、程序崩潰等）或是落後太多，則該Follower副本節點將從ISR列表中移除。

11．ZooKeeper

這裡我們並不打算介紹ZooKeeper的相關知識，只是簡要介紹ZooKeeper在Kafka中的作用。Kafka利用ZooKeeper儲存相應元資料資訊，Kafka元資料資訊包括如代理節點資訊、Kafka叢集資訊、舊版消費者資訊及其消費偏移量資訊、主題資訊、分割槽狀態資訊、分割槽副本分配方案資訊、動態配置資訊等。Kafka在啟動或執行過程當中會在ZooKeeper上建立相應節點來儲存元資料資訊，Kafka通過監聽機制在這些節點註冊相應監聽器來監聽節點元資料的變化，從而由ZooKeeper負責管理維護Kafka叢集，同時通過ZooKeeper我們能夠很方便地對Kafka叢集進行水平擴充套件及資料遷移。

通過以上Kafka基本概念的介紹，我們可以對Kafka基本結構圖進行完善，如圖1-2所示。

圖1-2　Kafka的叢集結構

1.4　Kafka設計概述

1.4.1　Kafka設計動機

Kafka的設計初衷是使Kafka能夠成為統一、實時處理大規模資料的平臺。為了達到這個目標，Kafka必須支援以下幾個應用場景。

（1）具有高吞吐量來支援諸如實時的日誌集這樣的大規模事件流。

（2）能夠很好地處理大量積壓的資料，以便能夠週期性地載入離線資料進行處理。

（3）能夠低延遲地處理傳統訊息應用場景。

（4）能夠支援分割槽、分散式，實時地處理訊息，同時具有容錯保障機制。

滿足以上功能的Kafka與傳統的訊息系統相比更像是一個數據庫日誌系統。瞭解了Kafka的設計動機之後，在下一節我們將看看Kafka發展至今已具有哪些特性。

1.4.2　Kafka特性

上一節對Kafka的設計動機進行了介紹。隨著Kafka的不斷更新發展，當前版本的Kafka又增加了一些新特性，下面就來逐個介紹Kafka的這些新特性。

1．訊息持久化

Kafka高度依賴於檔案系統來儲存和快取訊息。說到檔案系統，大家普遍認為磁碟讀寫慢，依賴於檔案系統進行儲存和快取訊息勢必在效能上會大打折扣，其實檔案系統儲存速度快慢一定程度上也取決於我們對磁碟的用法。據Kafka官方網站介紹：6塊7200r/min SATA RAID-5陣列的磁碟線性寫的速度為600 MB/s，而隨機寫的速度為100KB/s，線性寫的速度約是隨機寫的6000多倍。由此看來磁碟的快慢取決於我們是如何去應用磁碟。加之現代的作業系統提供了預讀（read-ahead）和延遲寫（write-behind）技術，使得磁碟的寫速度並不是大家想象的那麼慢。同時，由於Kafka是基於JVM（Java Virtual Machine）的，而Java物件記憶體消耗非常高，且隨著Java物件的增加JVM的垃圾回收也越來越頻繁和繁瑣，這些都加大了記憶體的消耗。鑑於以上因素，使用檔案系統和依賴於頁快取（page cache）的儲存比維護一個記憶體的儲存或是應用其他結構來儲存訊息更有優勢，因此Kafka選擇以檔案系統來儲存資料。

訊息系統資料持久化一般採用為每個消費者佇列提供一個 B 樹或其他通用的隨機訪問資料結構來維護訊息的元資料，B樹操作的時間複雜度為O(log n)，O(log n)的時間複雜度可以看成是一個常量時間，而且B樹可以支援各種各樣的事務性和非事務性語義訊息的傳遞。儘管B樹具有這些優點，但這並不適合磁碟操作。目前的磁碟尋道時間一般在10ms以內，對一塊磁碟來說，在同一時刻只能有一個磁頭來讀寫磁碟，這樣在併發IO能力上就有問題。同時，對樹結構效能的觀察結果表明：其效能會隨著資料的增長而線性下降。鑑於訊息系統本身的作用考慮，資料的持久化佇列可以建立在簡單地對檔案進行追加的實現方案上。因為是順序追加，所以Kafka在設計上是採用時間複雜度O(1)的磁碟結構，它提供了常量時間的效能，即使是儲存海量的資訊（TB級）也如此，效能和資料的大小關係也不大，同時Kafka將資料持久化到磁碟上，這樣只要磁碟空間足夠大資料就可以一直追加，而不會像一般的訊息系統在訊息被消費後就刪除掉，Kafka提供了相關配置讓使用者自己決定訊息要儲存多久，這樣為消費者提供了更靈活的處理方式，因此Kafka能夠在沒有效能損失的情況下提供一般訊息系統不具備的特性。

正是由於Kafka將訊息進行持久化，使得Kafka在機器重啟後，已儲存的訊息可繼續恢復使用。同時Kafka能夠很好地支援線上或離線處理、與其他儲存及流處理框架的整合。

2．高吞吐量

高吞吐量是Kafka設計的主要目標，Kafka將資料寫到磁碟，充分利用磁碟的順序讀寫。同時，Kafka在資料寫入及資料同步採用了零拷貝（zero-copy）技術，採用sendFile()函式呼叫，sendFile()函式是在兩個檔案描述符之間直接傳遞資料，完全在核心中操作，從而避免了核心緩衝區與使用者緩衝區之間資料的拷貝，操作效率極高。Kafka還支援資料壓縮及批量傳送，同時Kafka將每個主題劃分為多個分割槽，這一系列的優化及實現方法使得Kafka具有很高的吞吐量。經大多數公司對Kafka應用的驗證，Kafka支援每秒數百萬級別的訊息。

3．擴充套件性

Kafka要支援對大規模資料的處理，就必須能夠對叢集進行擴充套件，分散式必須是其特性之一，這樣就可以將多臺廉價的PC伺服器搭建成一個大規模的訊息系統。Kafka依賴ZooKeeper來對叢集進行協調管理，這樣使得Kafka更加容易進行水平擴充套件，生產者、消費者和代理都為分散式，可配置多個。同時在機器擴充套件時無需將整個叢集停機，叢集能夠自動感知，重新進行負責均衡及資料複製。

4．多客戶端支援

Kafka核心模組用Scala語言開發，但Kafka支援不同語言開發生產者和消費者客戶端應用程式。0.8.2之後的版本增加了Java版本的客戶端實現，0.10之後的版本已廢棄Scala語言實現的Producer及Consumer，預設使用Java版本的客戶端。Kafka提供了多種開發語言的接入，如Java、Scala、C、C++、Python、Go、Erlang、Ruby、Node.js等，感興趣的讀者可以自行參考https://cwiki.apache.org/confluence/display/KAFKA/Clients。同時，Kafka支援多種聯結器（Connector）的接入，也提供了Connector API供開發者呼叫。Kafka與當前主流的大資料框架都能很好地整合，如Flume、Hadoop、HBase、Hive、Spark、Storm等。

5．Kafka Streams

Kafka在0.10之後版本中引入Kafak Streams。Kafka Streams是一個用Java語言實現的用於流處理的jar檔案，關於Kafka Streams的相關內容將在第7章中進行講解。

6．安全機制

當前版本的Kafka支援以下幾種安全措施：

• 通過SSL和SASL(Kerberos)，SASL/PLAIN驗證機制支援生產者、消費者與代理連線時的身份認證；

• 支援代理與ZooKeeper連線身份驗證；

• 通訊時資料加密；

• 客戶端讀、寫許可權認證；

• Kafka支援與外部其他認證授權服務的整合。

7．資料備份

Kafka可以為每個主題指定副本數，對資料進行持久化備份，這可以一定程度上防止資料丟失，提高可用性。

8．輕量級

Kafka的代理是無狀態的，即代理不記錄訊息是否被消費，消費偏移量的管理交由消費者自己或組協調器來維護。同時叢集本身幾乎不需要生產者和消費者的狀態資訊，這就使得Kafka非常輕量級，同時生產者和消費者客戶端實現也非常輕量級。

9．訊息壓縮

Kafka支援Gzip、Snappy、LZ4這3種壓縮方式，通常把多條訊息放在一起組成MessageSet，然後再把MessageSet放到一條訊息裡面去，從而提高壓縮比率進而提高吞吐量。

1.4.3　Kafka應用場景

訊息系統或是說訊息佇列中介軟體是當前處理大資料一個非常重要的元件，用來解決應用解耦、非同步通訊、流量控制等問題，從而構建一個高效、靈活、訊息同步和非同步傳輸處理、儲存轉發、可伸縮和最終一致性的穩定系統。當前比較流行的訊息中介軟體有Kafka、RocketMQ、RabbitMQ、ZeroMQ、ActiveMQ、MetaMQ、Redis等，這些訊息中介軟體在效能及功能上各有所長。如何選擇一個訊息中介軟體取決於我們的業務場景、系統執行環境、開發及運維人員對訊息中件間掌握的情況等。我認為在下面這些場景中，Kafka是一個不錯的選擇。

（1）訊息系統。Kafka作為一款優秀的訊息系統，具有高吞吐量、內建的分割槽、備份冗餘分散式等特點，為大規模訊息處理提供了一種很好的解決方案。

（2）應用監控。利用Kafka採集應用程式和伺服器健康相關的指標，如CPU佔用率、IO、記憶體、連線數、TPS、QPS等，然後將指標資訊進行處理，從而構建一個具有監控儀表盤、曲線圖等視覺化監控系統。例如，很多公司採用Kafka與ELK（ElasticSearch、Logstash和Kibana）整合構建應用服務監控系統。

（3）網站使用者行為追蹤。為了更好地瞭解使用者行為、操作習慣，改善使用者體驗，進而對產品升級改進，將使用者操作軌跡、內容等資訊傳送到Kafka叢集上，通過Hadoop、Spark或Strom等進行資料分析處理，生成相應的統計報告，為推薦系統推薦物件建模提供資料來源，進而為每個使用者進行個性化推薦。

（4）流處理。需要將已收集的流資料提供給其他流式計算框架進行處理，用Kafka收集流資料是一個不錯的選擇，而且當前版本的Kafka提供了Kafka Streams支援對流資料的處理。

（5）永續性日誌。Kafka可以為外部系統提供一種永續性日誌的分散式系統。日誌可以在多個節點間進行備份，Kafka為故障節點資料恢復提供了一種重新同步的機制。同時，Kafka很方便與HDFS和Flume進行整合，這樣就方便將Kafka採集的資料持久化到其他外部系統。

kafka基礎最佳實戰(覆蓋原書第5章)

5.1　KafkaServer管理

Kafka執行依賴於ZooKeeper，在啟動Kafka之前，首先要保證ZooKeeper已正常啟動。在$KAFKA_HOME/bin目錄下，Kafka自帶了ZooKeeper操作的相應指令碼，讀者可以修改該目錄下ZooKeeper相關指令碼，然後執行。這裡對ZooKeeper操作並未使用Kafka提供的指令碼，而是直接執行$ZOOKEEPER_HOME/bin目錄下提供的相應操作指令碼及命令。Kafka提供了啟動KafkaServer的執行指令碼kafka-server-start.sh，而該指令碼核心程式碼是呼叫kafka-run-class.sh指令碼編譯kafka.Kafka類，程式碼如下：

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka"[email protected]"

kafka-run-class.sh是Kafka自帶的用來直接編譯執行Kafka原始碼中有main方法的Scala檔案。根據啟動KafkaServer時呼叫指令碼的不同，對KafkaServer啟動操作分以下兩小節分別進行介紹。

5.1.1　啟動Kafka單個節點

相關推薦