一、Kafka簡介

Kafka是一種高吞吐量、分布式、基于發布/訂閱的消息系統，最初由LinkedIn公司開發，使用Scala語言編寫，目前是Apache的開源項目。

跟RabbitMQ、RocketMQ等目前流行的開源消息中間件相比，Kakfa具有高吞吐、低延遲等特點，在大數據、日志收集等應用場景下被廣泛使用。

本文主要簡單介紹Kafka的設計原理。

二、Kafka架構

基本概念：

• broker：Kafka服務器，負責消息存儲和轉發
• topic：消息類別，Kafka按照topic來分類消息
• partition：topic的分區，一個topic可以包含多個partition，topic消息保存在各個partition上
• offset：消息在日志中的位置，可以理解是消息在partition上的偏移量，也是代表該消息的唯一序號
• Producer：消息生產者
• Consumer：消息消費者
• Consumer Group：消費者分組，每個Consumer必須屬于一個group
• Zookeeper：保存著集群broker、topic、partition等meta數據;另外，還負責broker故障發現，partition leader選舉，負載均衡等功能

三、Kafka設計原理

3.1 數據存儲設計

partition以文件形式存儲在文件系統，目錄命名規則：<topic_name>-<partition_id>，例如，名為test的topic，其有3個partition，則Kafka數據目錄中有3個目錄：test-0, test-1, test-2，分別存儲相應partition的數據。

partition的數據文件

partition中的每條Message包含了以下三個屬性：

• offset
• MessageSize
• data

其中offset表示Message在這個partition中的偏移量，offset不是該Message在partition數據文件中的實際存儲位置，而是邏輯上一個值，它唯一確定了partition中的一條Message，可以認為offset是partition中Message的id;MessageSize表示消息內容data的大小;data為Message的具體內容。

partition的數據文件由以上格式的Message組成，按offset由小到大排列在一起。

如果一個partition只有一個數據文件：

1. 新數據是添加在文件末尾，不論文件數據文件有多大，這個操作永遠都是O(1)的。
2. 查找某個offset的Message是順序查找的。因此，如果數據文件很大的話，查找的效率就低。

Kafka通過分段和索引來提高查找效率。

數據文件分段segment

partition物理上由多個segment文件組成，每個segment大小相等，順序讀寫。每個segment數據文件以該段中最小的offset命名，文件擴展名為.log。這樣在查找指定offset的Message的時候，用二分查找就可以定位到該Message在哪個segment數據文件中。

數據文件索引

數據文件分段使得可以在一個較小的數據文件中查找對應offset的Message了，但是這依然需要順序掃描才能找到對應offset的Message。為了進一步提高查找的效率，Kafka為每個分段后的數據文件建立了索引文件，文件名與數據文件的名字是一樣的，只是文件擴展名為.index。

索引文件中包含若干個索引條目，每個條目表示數據文件中一條Message的索引。索引包含兩個部分，分別為相對offset和position。

• 相對offset：因為數據文件分段以后，每個數據文件的起始offset不為0，相對offset表示這條Message相對于其所屬數據文件中最小的offset的大小。舉例，分段后的一個數據文件的offset是從20開始，那么offset為25的Message在index文件中的相對offset就是25-20 = 5。存儲相對offset可以減小索引文件占用的空間。
• position，表示該條Message在數據文件中的絕對位置。只要打開文件并移動文件指針到這個position就可以讀取對應的Message了。

index文件中并沒有為數據文件中的每條Message建立索引，而是采用了稀疏存儲的方式，每隔一定字節的數據建立一條索引。這樣避免了索引文件占用過多的空間，從而可以將索引文件保留在內存中。但缺點是沒有建立索引的Message也不能一次定位到其在數據文件的位置，從而需要做一次順序掃描，但是這次順序掃描的范圍就很小了。

總結

查找某個offset的消息，先二分法找出消息所在的segment文件(因為每個segment的命名都是以該文件中消息offset最小的值命名);然后，加載對應的.index索引文件到內存，同樣二分法找出小于等于給定offset的***的那個offset記錄(相對offset，position);***，根據position到.log文件中，順序查找出offset等于給定offset值的消息。

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

3.2 生產者設計

• 負載均衡：由于消息topic由多個partition組成，且partition會均衡分布到不同broker上，因此，為了有效利用broker集群的性能，提高消息的吞吐量，producer可以通過隨機或者hash等方式，將消息平均發送到多個partition上，以實現負載均衡。
• 批量發送：是提高消息吞吐量重要的方式，Producer端可以在內存中合并多條消息后，以一次請求的方式發送了批量的消息給broker，從而大大減少broker存儲消息的IO操作次數。但也一定程度上影響了消息的實時性，相當于以時延代價，換取更好的吞吐量。

3.3 消費者設計

• 任何Consumer必須屬于一個Consumer Group
• 同一Consumer Group中的多個Consumer實例，不同時消費同一個partition，等效于隊列模式。如圖，Consumer Group 1的三個Consumer實例分別消費不同的partition的消息，即，TopicA-part0、TopicA-part1、TopicA-part2。
• 不同Consumer Group的Consumer實例可以同時消費同一個partition，等效于發布訂閱模式。如圖，Consumer Group 1的Consumer1和Consumer Group 2的Consumer4，同時消費TopicA-part0的消息。
• partition內消息是有序的，Consumer通過pull方式消費消息。
• Kafka不刪除已消費的消息

隊列模式

隊列模式，指每條消息只會有一個Consumer消費到。Kafka保證同一Consumer Group中只有一個Consumer會消費某條消息。

• 在Consumer Group穩定狀態下，每一個Consumer實例只會消費某一個或多個特定partition的數據，而某個partition的數據只會被某一個特定的Consumer實例所消費，也就是說Kafka對消息的分配是以partition為單位分配的，而非以每一條消息作為分配單元;
• 同一Consumer Group中，如果Consumer實例數量少于partition數量，則至少有一個Consumer會消費多個partition的數據;如果Consumer的數量與partition數量相同，則正好一個Consumer消費一個partition的數據;而如果Consumer的數量多于partition的數量時，會有部分Consumer無法消費該Topic下任何一條消息;
• 設計的優勢是：每個Consumer不用都跟大量的broker通信，減少通信開銷，同時也降低了分配難度，實現也更簡單;可以保證每個partition里的數據可以被Consumer有序消費。
• 設計的劣勢是：無法保證同一個Consumer Group里的Consumer均勻消費數據，且在Consumer實例多于partition個數時導致有些Consumer會餓死。

如果有partition或者Consumer的增減，為了保證均衡消費，需要實現Consumer Rebalance，分配算法如下：

broker對Consumer設計原理：

• 對于每個Consumer Group，選舉出一個Broker作為Coordinator(0.9版本以上)，由它Watch Zookeeper，從而監控判斷是否有partition或者Consumer的增減，然后生成Rebalance命令，按照以上算法重新分配。
• 當Consumer Group***次被初始化時，Consumer通常會讀取每個partition的最早或最近的offset(Zookeeper記錄)，然后順序地讀取每個partition log的消息，在Consumer讀取過程中，它會提交已經成功處理的消息的offsets(由Zookeeper記錄)。
• 當一個partition被重新分配給Consumer Group中的其他Consumer，新的Consumer消費的初始位置會設置為(原來Consumer)最近提交的offset。

如圖，Last Commited Offset指Consumer最近一次提交的消費記錄offset，Current Position是當前消費的位置，High Watermark是成功拷貝到log的所有副本節點(partition的所有ISR節點，下文介紹)的最近消息的offset，Log End Offset是寫入log中***一條消息的offset+1。

從Consumer的角度來看，最多只能讀取到High watermark的位置，后面的消息對消費者不可見，因為未完全復制的數據還沒可靠存儲，有丟失可能。

發布訂閱模式

發布訂閱模式，又指廣播模式，Kafka保證topic的每條消息會被所有Consumer Group消費到，而對于同一個Consumer Group，還是保證只有一個Consumer實例消費到這條消息。

3.4 Replication設計

作為消息中間件，數據的可靠性以及系統的可用性，必然依賴數據副本的設計。

Kafka的replica副本單元是topic的partition，一個partition的replica數量不能超過broker的數量，因為一個broker最多只會存儲這個partition的一個副本。所有消息生產、消費請求都是由partition的leader replica來處理，其他follower replica負責從leader復制數據進行備份。

Replica均勻分布到整個集群，Replica的算法如下：

• 將所有Broker(假設共n個Broker)和待分配的Partition排序
• 將第i個Partition分配到第(i mod n)個Broker上
• 將第i個Partition的第j個Replica分配到第((i + j) mode n)個Broker上

如圖，TopicA有三個partition：part0、part1、part2，每個partition的replica數等于2(一個是leader，另一個是follower)，按照以上算法會均勻落到三個broker上。

broker對replica管理：

選舉出一個broker作為controller，由它Watch Zookeeper，負責partition的replica的集群分配，以及leader切換選舉等流程。

In-Sync-Replica(ISR)

分布式系統在處理節點故障時，需要預先明確節點的”failure”和”alive”的定義。對于Kafka節點，判斷是”alive”有以下兩個條件：

• 節點必須和Zookeeper保持心跳連接
• 如果節點是follower，必須從leader節點上復制數據來備份，而且備份的數據相比leader而言，不能落后太多。

Kafka將滿足以上條件的replica節點認為是”in sync”(同步中)，稱為In-Sync-Replica(ISR)。

Kafka的Zookeeper維護了每個partition的ISR信息，理想情況下，ISR包含了partition的所有replica所在的broker節點信息，而當某些節點不滿足以上條件時，ISR可能只包含部分replica。例如，上圖中的TopicA-part0的ISR列表可能是[broker1,broker2,broker3]，也可能是[broker1,broker3]和[broker1]。

數據可靠性

Kafka如何保證數據可靠性?首先看下，Producer生產一條消息，該消息被認為是”committed”(即broker認為消息已經可靠存儲)的過程：

• 消息所在partition的ISR replicas會定時異步從leader上批量復制數據log
• 當所有ISR replica都返回ack，告訴leader該消息已經寫log成功后，leader認為該消息committed，并告訴Producer生產成功。這里和以上”alive”條件的第二點是不矛盾的，因為leader有超時機制，leader等ISR的follower復制數據，如果一定時間不返回ack(可能數據復制進度落后太多)，則leader將該follower replica從ISR中剔除。
• 消息committed之后，Consumer才能消費到。

ISR機制下的數據復制，既不是完全的同步復制，也不是單純的異步復制，這是Kafka高吞吐很重要的機制。同步復制要求所有能工作的follower都復制完，這條消息才會被認為committed，這種復制方式極大的影響了吞吐量。而異步復制方式下，follower異步的從leader復制數據，數據只要被leader寫入log就被認為已經committed，這種情況下如果follower都復制完都落后于leader，而如果leader突然宕機，則會丟失數據。而Kafka的這種使用ISR的方式則很好的均衡了確保數據不丟失以及吞吐量，follower可以批量的從leader復制數據，數據復制到內存即返回ack，這樣極大的提高復制性能，當然數據仍然是有丟失風險的。

Kafka本身定位于高性能的MQ，更多注重消息吞吐量，在此基礎上結合ISR的機制去盡量保證消息的可靠性，但不是絕對可靠的。

服務可用性

Kafka所有收發消息請求都由leader節點處理，由以上數據可靠性設計可知，當ISR的follower replica故障后，leader會及時地從ISR列表中把它剔除掉，并不影響服務可用性，那么當leader故障后會怎樣呢?如何選舉新的leader?

leader選舉

• Kafka在Zookeeper存儲partition的ISR信息，并且能動態調整ISR列表的成員，只有ISR里的成員replica才會被選為leader，并且ISR所有的replica都有可能成為leader;
• leader節點宕機后，Zookeeper能監控發現，并由broker的controller節點從ISR中選舉出新的leader，并通知ISR內的所有broker節點。

因此，可以看出，只要ISR中至少有一個replica，Kafka就能保證服務的可用性(但不保證網絡分區下的可用性)。

容災和數據一致性

分布式系統的容災能力，跟其本身針對數據一致性考慮所選擇的算法有關，例如，Zookeeper的Zab算法，raft算法等。Kafka的ISR機制和這些Majority Vote算法對比如下：

ISR機制能容忍更多的節點失敗。假如replica節點有2f+1個，每個partition最多能容忍2f個失敗，且不丟失消息數據;但相對Majority Vote選舉算法，只能最多容忍f個失敗。

在消息committed持久化上，ISR需要等2f個節點返回ack，但Majority Vote只需等f+1個節點返回ack，且不依賴處理最慢的follower節點，因此Majority Vote有優勢

ISR機制能節省更多replica節點數。例如，要保證f個節點可用，ISR方式至少要f個節點，而Majority Vote至少需要2f+1個節點。

如果所有replica都宕機了，有兩種方式恢復服務：

1. 等ISR任一節點恢復，并選舉為leader;
2. 選擇***個恢復的節點(不一定是ISR中的節點)為leader

***種方式消息不會丟失(只能說這種方式最有可能不丟而已)，第二種方式可能會丟消息，但能盡快恢復服務可用。這是可用性和一致性場景的兩種考慮，Kafka默認選擇第二種，用戶也可以自主配置。

大部分考慮CP的分布式系統(假設2f+1個節點)，為了保證數據一致性，最多只能容忍f個節點的失敗，而Kafka為了兼顧可用性，允許最多2f個節點失敗，因此是無法保證數據強一致的。

如圖所示，一開始ISR數量等于3，正常同步數據，紅色部分開始，leader發現其他兩個follower復制進度太慢或者其他原因(網絡分區、節點故障等)，將其從ISR剔除后，leader單節點存儲數據;然后，leader宕機，觸發重新選舉第二節點為leader，重新開始同步數據，但紅色部分的數據在新leader上是沒有的;***原leader節點恢復服務后，重新從新leader上復制數據，而紅色部分的數據已經消費不到了。

因此，為了減少數據丟失的概率，可以設置Kafka的ISR最小replica數，低于該值后直接返回不可用，當然是以犧牲一定可用性和吞吐量為前提了。

重復消息

消息傳輸有三種方式：

At most once：消息可能會丟失，但不會重復傳輸

At least once：消息不會丟失，但可能重復傳輸

Exactly once：消息保證會被傳輸一次且僅傳輸一次

Kafka實現了第二種方式，即，可能存在重復消息，需要業務自己保證消息冪等性處理。

3.5 高吞吐設計

1. 對于partition，順序讀寫磁盤數據，以時間復雜度O(1)方式提供消息持久化能力。
2. Producer批量向broker寫數據
3. Consumer批量從broker拉數據
4. 日志壓縮
5. Topic分多個partition，提高并發
6. broker零拷貝(Zero Copy)，使用sendfile系統調用，將數據直接從page cache發送到socket上
7. Producer可配置是否等待消息committed。如果Producer生產消息，每次都必須等ISR存儲后才返回，時延會很高，進而影響整體消息的吞吐量。為了解決這個問題，一方面Producer可以配置減少partition的副本數，例如，ISR大小為1;另一方面，在不太關注消息可靠存儲的場景下，Producer可以通過配置選擇是否等待消息committed，如下：

這是用戶在消息吞吐量和持久化之間做的權衡選擇，持久化等級越高，生產消息吞吐量越小，反之，持久化等級越低，吞吐量越高。

3.6 HA基本原理

broker HA

broker集群信息由Zookeeper維護，并選舉出一個controller。所有partition的leader選舉都由controller決定，將leader的變更直接通過rpc方式通知需要為此做出響應的brokers;controller也負責增刪topic以及partition replica的重新分配。

controller在Zookeeper上注冊watch，一旦有broker宕機，其對應在Zookeeper的臨時節點自動被刪除，controller對宕機broker上的所有partition重新分配新leader;如果controller宕機，其他broker通過Zookeeper選舉出新的controller，然后同樣對宕機broker上的所有partition重新分配新leader。

partition HA

partition leader所在的broker宕機，如上所述，broker controller根據動態維護的ISR，會重新在剩下的broker機器中選出ISR里面的一個成員成為新的leader。如果ISR中至少有一個follower，則可以確保已經committed的數據不丟失;否則選擇任意一個replica作為leader，該場景可能會有潛在的數據丟失;如果partition所有的replica都宕機了，就無法保證數據不丟失了，有兩種恢復方案，上文已介紹過。

四、推廣

騰訊云即將推出高性能的消息隊列服務Ckafka，完全兼容開源Kafka API(0.9版本)。Ckafka服務端完全托管在騰訊云上，用戶無需自己維護和搭建，使用開源Kafka API客戶端即可訪問實例，大大降低了用戶使用Kafka的門檻，歡迎體驗：)

原文鏈接：https://cloud.tencent.com/community/article/369570

【本文是51CTO專欄作者“騰訊云技術社區”的原創稿件，轉載請通過51CTO聯系原作者獲取授權】

戳這里，看該作者更多好文