Kafka 是一個消息系統，原本開發自 LinkedIn，用作 LinkedIn 的活動流(Activity Stream)和運營數據處理管道(Pipeline)的基礎。

[[273550]]

圖片來自 Pexels

現在它已被多家不同類型的公司作為多種類型的數據管道和消息系統使用。活動流數據是幾乎所有站點在對其網站使用情況做報表時都要用到的數據中最常規的部分。

活動數據包括頁面訪問量(Page View)、被查看內容方面的信息以及搜索情況等內容。

這種數據通常的處理方式是先把各種活動以日志的形式寫入某種文件，然后周期性地對這些文件進行統計分析。

運營數據指的是服務器的性能數據(CPU、IO 使用率、請求時間、服務日志等等數據)。運營數據的統計方法種類繁多。

近年來，活動和運營數據處理已經成為了網站軟件產品特性中一個至關重要的組成部分，這就需要一套稍微更加復雜的基礎設施對其提供支持。

Kafka 基礎概念

Kafka 是一種分布式的，基于發布/訂閱的消息系統，主要設計目標如下：

• 以時間復雜度為 O(1) 的方式提供消息持久化能力，即使對 TB 級以上數據也能保證常數時間復雜度的訪問性能。
• 高吞吐率。即使在非常廉價的商用機器上也能做到單機支持每秒 100K 條以上消息的傳輸。
• 支持 Kafka Server 間的消息分區，及分布式消費，同時保證每個 Partition 內的消息順序傳輸。
• 同時支持離線數據處理和實時數據處理。
• Scale out：支持在線水平擴展。

生產者與消費者

對于 Kafka 來說客戶端有兩種基本類型：

• 生產者(Producer)
• 消費者(Consumer)

除此之外，還有用來做數據集成的 Kafka Connect API 和流式處理的 Kafka Streams 等高階客戶端，但這些高階客戶端底層仍然是生產者和消費者 API，它們只不過是在上層做了封裝。

這很容易理解，生產者(也稱為發布者)創建消息，而消費者(也稱為訂閱者)負責消費 or 讀取消息。

主題(Topic)與分區(Partition)

在 Kafka 中，消息以主題(Topic)來分類，每一個主題都對應一個「消息隊列」，這有點兒類似于數據庫中的表。

但是如果我們把所有同類的消息都塞入到一個“中心”隊列中，勢必缺少可伸縮性，無論是生產者/消費者數目的增加，還是消息數量的增加，都可能耗盡系統的性能或存儲。

我們使用一個生活中的例子來說明：現在 A 城市生產的某商品需要運輸到 B 城市，走的是公路。

那么單通道的高速公路不論是在「A 城市商品增多」還是「現在 C 城市也要往 B 城市運輸東西」這樣的情況下都會出現「吞吐量不足」的問題。

所以我們現在引入分區(Partition)的概念，類似“允許多修幾條道”的方式對我們的主題完成了水平擴展。

Broker 和集群(Cluster)

一個 Kafka 服務器也稱為 Broker，它接受生產者發送的消息并存入磁盤;Broker 同時服務消費者拉取分區消息的請求，返回目前已經提交的消息。

使用特定的機器硬件，一個 Broker 每秒可以處理成千上萬的分區和百萬量級的消息。(現在動不動就百萬量級，我特地去查了一把，好像確實集群的情況下吞吐量挺高的。)

若干個 Broker 組成一個集群(Cluster)，其中集群內某個 Broker 會成為集群控制器(Cluster Controller)，它負責管理集群，包括分配分區到 Broker、監控 Broker 故障等。

在集群內，一個分區由一個 Broker 負責，這個 Broker 也稱為這個分區的 Leader。

當然一個分區可以被復制到多個 Broker 上來實現冗余，這樣當存在 Broker 故障時可以將其分區重新分配到其他 Broker 來負責。

下圖是一個樣例：

Kafka 的一個關鍵性質是日志保留(Retention)，我們可以配置主題的消息保留策略，譬如只保留一段時間的日志或者只保留特定大小的日志。

當超過這些限制時，老的消息會被刪除。我們也可以針對某個主題單獨設置消息過期策略，這樣對于不同應用可以實現個性化。

多集群

隨著業務發展，我們往往需要多集群，通常處于下面幾個原因：

• 基于數據的隔離
• 基于安全的隔離
• 多數據中心(容災)

當構建多個數據中心時，往往需要實現消息互通。舉個例子，假如用戶修改了個人資料，那么后續的請求無論被哪個數據中心處理，這個更新需要反映出來。又或者，多個數據中心的數據需要匯總到一個總控中心來做數據分析。

上面說的分區復制冗余機制只適用于同一個 Kafka 集群內部，對于多個 Kafka 集群消息同步可以使用 Kafka 提供的 MirrorMaker 工具。

本質上來說，MirrorMaker 只是一個 Kafka 消費者和生產者，并使用一個隊列連接起來而已。它從一個集群中消費消息，然后往另一個集群生產消息。

Kafka 的設計與實現

上面我們知道了 Kafka 中的一些基本概念，但作為一個成熟的「消息隊列」中間件，其中有許多有意思的設計值得我們思考，下面我們簡單列舉一些。

Kafka 存儲在文件系統上

是的，您首先應該知道 Kafka 的消息是存在于文件系統之上的。Kafka 高度依賴文件系統來存儲和緩存消息，一般的人認為 “磁盤是緩慢的”，所以對這樣的設計持有懷疑態度。

實際上，磁盤比人們預想的快很多也慢很多，這取決于它們如何被使用;一個好的磁盤結構設計可以使之跟網絡速度一樣快。

現代的操作系統針對磁盤的讀寫已經做了一些優化方案來加快磁盤的訪問速度。

比如，預讀會提前將一個比較大的磁盤快讀入內存。后寫會將很多小的邏輯寫操作合并起來組合成一個大的物理寫操作。

并且，操作系統還會將主內存剩余的所有空閑內存空間都用作磁盤緩存，所有的磁盤讀寫操作都會經過統一的磁盤緩存(除了直接 I/O 會繞過磁盤緩存)。

綜合這幾點優化特點，如果是針對磁盤的順序訪問，某些情況下它可能比隨機的內存訪問都要快，甚至可以和網絡的速度相差無幾。

上述的 Topic 其實是邏輯上的概念，面向消費者和生產者，物理上存儲的其實是 Partition，每一個 Partition 最終對應一個目錄，里面存儲所有的消息和索引文件。

默認情況下，每一個 Topic 在創建時如果不指定 Partition 數量時只會創建 1 個 Partition。

比如，我創建了一個 Topic 名字為 test ，沒有指定 Partition 的數量，那么會默認創建一個 test-0 的文件夾，這里的命名規則是：- 。

任何發布到 Partition 的消息都會被追加到 Partition 數據文件的尾部，這樣的順序寫磁盤操作讓 Kafka 的效率非常高(經驗證，順序寫磁盤效率比隨機寫內存還要高，這是 Kafka 高吞吐率的一個很重要的保證)。

每一條消息被發送到 Broker 中，會根據 Partition 規則選擇被存儲到哪一個 Partition。如果 Partition 規則設置的合理，所有消息可以均勻分布到不同的 Partition中。

Kafka 中的底層存儲設計

假設我們現在 Kafka 集群只有一個 Broker，我們創建 2 個 Topic 名稱分別為：「Topic1」和「Topic2」，Partition 數量分別為 1、2。

那么我們的根目錄下就會創建如下三個文件夾：

| --topic1-0 
| --topic2-0 
| --topic2-1 

在 Kafka 的文件存儲中，同一個 Topic 下有多個不同的 Partition，每個 Partition 都為一個目錄。

而每一個目錄又被平均分配成多個大小相等的 Segment File 中，Segment File 又由 index file 和 data file 組成，他們總是成對出現，后綴 ".index" 和 ".log" 分表表示 Segment 索引文件和數據文件。

現在假設我們設置每個 Segment 大小為 500 MB，并啟動生產者向 topic1 中寫入大量數據，topic1-0 文件夾中就會產生類似如下的一些文件：

| --topic1-0 
    | --00000000000000000000.index 
    | --00000000000000000000.log 
    | --00000000000000368769.index 
    | --00000000000000368769.log 
    | --00000000000000737337.index 
    | --00000000000000737337.log 
    | --00000000000001105814.index 
    | --00000000000001105814.log 
| --topic2-0 
| --topic2-1 

Segment 是 Kafka 文件存儲的最小單位。Segment 文件命名規則：Partition 全局的第一個 Segment 從 0 開始，后續每個 Segment 文件名為上一個 Segment 文件最后一條消息的 offset 值。

數值最大為 64 位 long 大小，19 位數字字符長度，沒有數字用 0 填充。如 00000000000000368769.index 和 00000000000000368769.log。

以上面的一對 Segment File 為例，說明一下索引文件和數據文件對應關系：

其中以索引文件中元數據 <3, 497> 為例，依次在數據文件中表示第 3 個 Message(在全局 Partition 表示第 368769 + 3 = 368772 個 message)以及該消息的物理偏移地址為 497。

注意該 Index 文件并不是從0開始，也不是每次遞增 1 的，這是因為 Kafka 采取稀疏索引存儲的方式，每隔一定字節的數據建立一條索引。

它減少了索引文件大小，使得能夠把 Index 映射到內存，降低了查詢時的磁盤 IO 開銷，同時也并沒有給查詢帶來太多的時間消耗。

因為其文件名為上一個 Segment 最后一條消息的 Offset ，所以當需要查找一個指定 Offset 的 Message 時，通過在所有 Segment 的文件名中進行二分查找就能找到它歸屬的 Segment。

再在其 Index 文件中找到其對應到文件上的物理位置，就能拿出該 Message。

由于消息在 Partition 的 Segment 數據文件中是順序讀寫的，且消息消費后不會刪除(刪除策略是針對過期的 Segment 文件)，這是順序磁盤 IO 存儲設計師 Kafka 高性能很重要的原因。

Kafka 是如何準確的知道 Message 的偏移的呢?這是因為在 Kafka 定義了標準的數據存儲結構，在 Partition 中的每一條 Message 都包含了以下三個屬性：

• Offset：表示 Message 在當前 Partition 中的偏移量，是一個邏輯上的值，唯一確定了 Partition 中的一條 Message，可以簡單的認為是一個 ID。
• MessageSize：表示 Message 內容 Data 的大小。
• Data：Message 的具體內容。

生產者設計概要

當我們發送消息之前，先問幾個問題：每條消息都是很關鍵且不能容忍丟失么?偶爾重復消息可以么?我們關注的是消息延遲還是寫入消息的吞吐量?

舉個例子，有一個信用卡交易處理系統，當交易發生時會發送一條消息到 Kafka，另一個服務來讀取消息并根據規則引擎來檢查交易是否通過，將結果通過 Kafka 返回。

對于這樣的業務，消息既不能丟失也不能重復，由于交易量大因此吞吐量需要盡可能大，延遲可以稍微高一點。

再舉個例子，假如我們需要收集用戶在網頁上的點擊數據，對于這樣的場景，少量消息丟失或者重復是可以容忍的，延遲多大都不重要只要不影響用戶體驗，吞吐則根據實時用戶數來決定。

不同的業務需要使用不同的寫入方式和配置。具體的方式我們在這里不做討論，現在先看下生產者寫消息的基本流程：

流程如下：

• 首先，我們需要創建一個 ProducerRecord，這個對象需要包含消息的主題(Topic)和值(Value)，可以選擇性指定一個鍵值(Key)或者分區(Partition)。
• 發送消息時，生產者會對鍵值和值序列化成字節數組，然后發送到分配器(Partitioner)。
• 如果我們指定了分區，那么分配器返回該分區即可;否則，分配器將會基于鍵值來選擇一個分區并返回。
• 選擇完分區后，生產者知道了消息所屬的主題和分區，它將這條記錄添加到相同主題和分區的批量消息中，另一個線程負責發送這些批量消息到對應的Kafka Broker。
• 當 Broker 接收到消息后，如果成功寫入則返回一個包含消息的主題、分區及位移的 RecordMetadata 對象，否則返回異常。
• 生產者接收到結果后，對于異常可能會進行重試。

消費者設計概要

①消費者與消費組

假設這么個場景：我們從 Kafka 中讀取消息，并且進行檢查，最后產生結果數據。

我們可以創建一個消費者實例去做這件事情，但如果生產者寫入消息的速度比消費者讀取的速度快怎么辦呢?

這樣隨著時間增長，消息堆積越來越嚴重。對于這種場景，我們需要增加多個消費者來進行水平擴展。

Kafka 消費者是消費組的一部分，當多個消費者形成一個消費組來消費主題時，每個消費者會收到不同分區的消息。

假設有一個 T1 主題，該主題有 4 個分區;同時我們有一個消費組 G1，這個消費組只有一個消費者 C1。

那么消費者 C1 將會收到這 4 個分區的消息，如下所示：

如果我們增加新的消費者 C2 到消費組 G1，那么每個消費者將會分別收到兩個分區的消息，如下所示：

如果增加到 4 個消費者，那么每個消費者將會分別收到一個分區的消息，如下所示：

但如果我們繼續增加消費者到這個消費組，剩余的消費者將會空閑，不會收到任何消息：

總而言之，我們可以通過增加消費組的消費者來進行水平擴展提升消費能力。

這也是為什么建議創建主題時使用比較多的分區數，這樣可以在消費負載高的情況下增加消費者來提升性能。

另外，消費者的數量不應該比分區數多，因為多出來的消費者是空閑的，沒有任何幫助。

Kafka 一個很重要的特性就是，只需寫入一次消息，可以支持任意多的應用讀取這個消息。

換句話說，每個應用都可以讀到全量的消息。為了使得每個應用都能讀到全量消息，應用需要有不同的消費組。

對于上面的例子，假如我們新增了一個新的消費組 G2，而這個消費組有兩個消費者，那么會是這樣的：

在這個場景中，消費組 G1 和消費組 G2 都能收到 T1 主題的全量消息，在邏輯意義上來說它們屬于不同的應用。

最后，總結起來就是：如果應用需要讀取全量消息，那么請為該應用設置一個消費組;如果該應用消費能力不足，那么可以考慮在這個消費組里增加消費者。

②消費組與分區重平衡

可以看到，當新的消費者加入消費組，它會消費一個或多個分區，而這些分區之前是由其他消費者負責的。

另外，當消費者離開消費組(比如重啟、宕機等)時，它所消費的分區會分配給其他分區。

這種現象稱為重平衡(Rebalance)。重平衡是 Kafka 一個很重要的性質，這個性質保證了高可用和水平擴展。

不過也需要注意到，在重平衡期間，所有消費者都不能消費消息，因此會造成整個消費組短暫的不可用。

而且，將分區進行重平衡也會導致原來的消費者狀態過期，從而導致消費者需要重新更新狀態，這段期間也會降低消費性能。后面我們會討論如何安全的進行重平衡以及如何盡可能避免。

消費者通過定期發送心跳(Hearbeat)到一個作為組協調者(Group Coordinator)的 Broker 來保持在消費組內存活。

這個 Broker 不是固定的，每個消費組都可能不同。當消費者拉取消息或者提交時，便會發送心跳。

如果消費者超過一定時間沒有發送心跳，那么它的會話(Session)就會過期，組協調者會認為該消費者已經宕機，然后觸發重平衡。

可以看到，從消費者宕機到會話過期是有一定時間的，這段時間內該消費者的分區都不能進行消息消費。

通常情況下，我們可以進行優雅關閉，這樣消費者會發送離開的消息到組協調者，這樣組協調者可以立即進行重平衡而不需要等待會話過期。

在 0.10.1 版本，Kafka 對心跳機制進行了修改，將發送心跳與拉取消息進行分離，這樣使得發送心跳的頻率不受拉取的頻率影響。

另外更高版本的 Kafka 支持配置一個消費者多長時間不拉取消息但仍然保持存活，這個配置可以避免活鎖(livelock)。活鎖，是指應用沒有故障但是由于某些原因不能進一步消費。

③Partition 與消費模型

上面提到，Kafka 中一個 Topic 中的消息是被打散分配在多個 Partition(分區)中存儲的， Consumer Group 在消費時需要從不同的 Partition 獲取消息，那最終如何重建出 Topic 中消息的順序呢?

答案是：沒有辦法。Kafka 只會保證在 Partition 內消息是有序的，而不管全局的情況。

下一個問題是：Partition 中的消息可以被(不同的 Consumer Group)多次消費，那 Partition中被消費的消息是何時刪除的?Partition 又是如何知道一個 Consumer Group 當前消費的位置呢?

無論消息是否被消費，除非消息到期 Partition 從不刪除消息。例如設置保留時間為 2 天，則消息發布 2 天內任何 Group 都可以消費，2 天后，消息自動被刪除。

Partition 會為每個 Consumer Group 保存一個偏移量，記錄 Group 消費到的位置。如下圖：

④為什么 Kafka 是 Pull 模型

消費者應該向 Broker 要數據(Pull)還是 Broker 向消費者推送數據(Push)?

作為一個消息系統，Kafka 遵循了傳統的方式，選擇由 Producer 向 Broker Push 消息并由 Consumer 從 Broker Pull 消息。

一些 logging-centric system，比如 Facebook 的 Scribe 和 Cloudera 的 Flume，采用 Push 模式。事實上，Push 模式和 Pull 模式各有優劣。

Push 模式很難適應消費速率不同的消費者，因為消息發送速率是由 Broker 決定的。

Push 模式的目標是盡可能以最快速度傳遞消息，但是這樣很容易造成 Consumer 來不及處理消息，典型的表現就是拒絕服務以及網絡擁塞。

而 Pull 模式則可以根據 Consumer 的消費能力以適當的速率消費消息。

對于 Kafka 而言，Pull 模式更合適。Pull 模式可簡化 Broker 的設計，Consumer 可自主控制消費消息的速率。

同時 Consumer 可以自己控制消費方式，即可批量消費也可逐條消費，同時還能選擇不同的提交方式從而實現不同的傳輸語義。

Kafka 如何保證可靠性

當我們討論可靠性的時候，我們總會提到保證*這個詞語。可靠性保證是基礎，我們基于這些基礎之上構建我們的應用。

比如關系型數據庫的可靠性保證是 ACID，也就是原子性(Atomicity)、一致性(Consistency)、隔離性(Isolation)和持久性(Durability)。

Kafka 中的可靠性保證有如下 4 點：

• 對于一個分區來說，它的消息是有序的。如果一個生產者向一個分區先寫入消息 A，然后寫入消息 B，那么消費者會先讀取消息 A 再讀取消息 B。
• 當消息寫入所有 in-sync 狀態的副本后，消息才會認為已提交(committed)。

這里的寫入有可能只是寫入到文件系統的緩存，不一定刷新到磁盤。生產者可以等待不同時機的確認，比如等待分區主副本寫入即返回，后者等待所有 in-sync 狀態副本寫入才返回。

• 一旦消息已提交，那么只要有一個副本存活，數據不會丟失。
• 消費者只能讀取到已提交的消息。

使用這些基礎保證，我們構建一個可靠的系統，這時候需要考慮一個問題：究竟我們的應用需要多大程度的可靠性?

可靠性不是無償的，它與系統可用性、吞吐量、延遲和硬件價格息息相關，得此失彼。因此，我們往往需要做權衡，一味的追求可靠性并不實際。

動手搭一個 Kafka

通過上面的描述，我們已經大致了解到了 Kafka 是何方神圣了，現在我們開始嘗試自己動手本地搭一個來實際體驗一把。

第一步：下載 Kafka

這里以 Mac OS 為例，在安裝了 Homebrew 的情況下執行下列代碼：

brew install kafka 

由于 Kafka 依賴了 Zookeeper，所以在下載的時候會自動下載。

第二步：啟動服務

我們在啟動之前首先需要修改 Kafka 的監聽地址和端口為 localhost:9092：

vi /usr/local/etc/kafka/server.properties 

然后修改成下圖的樣子：

依次啟動 Zookeeper 和 Kafka：

brew services start zookeeper 
brew services start kafka 

然后執行下列語句來創建一個名字為 "test" 的 Topic：

kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test 

我們可以通過下列的命令查看我們的 Topic 列表：

kafka-topics --list --zookeeper localhost:2181 

第三步：發送消息

然后我們新建一個控制臺，運行下列命令創建一個消費者關注剛才創建的 Topic：

kafka-console-consumer --bootstrap-server localhost:9092 --topic test --from-beginning 

用控制臺往剛才創建的 Topic 中添加消息，并觀察剛才創建的消費者窗口：

kafka-console-producer --broker-list localhost:9092 --topic test 

能通過消費者窗口觀察到正確的消息：

參考資料：

1. Kafka 設計解析(一)：Kafka 背景及架構介紹
2. Kafka系列(一)初識Kafka
3. Kafka 入門介紹
4. Kafka 中的 Topic 為什么要進行分區? - 知乎
5. Kafka 的設計與實踐思考
6. Kafka系列(六)可靠的數據傳輸