1 Kafka概念和架構

第一講：概念、ZK的存儲結構、Producer、Consumers流程、Kafka Broker的啟動（額外）
從客戶端使用角度來講。

第二講：從設計原理角度來講。

Kafka屬於Apache組織，是一個高性能跨語言分布式發布訂閱消息隊列系統[7]。它的主要特點有：

• 以時間復雜度O(1)的方式提供消息持久化能力，並對大數據量能保證常數時間的訪問性能；
• 高吞吐率，單台服務器可以達到每秒幾十萬的吞吐速率；
• 支持服務器間的消息分區，支持分布式消費，同時保證了每個分區內的消息順序；
• 輕量級，支持實時數據處理和離線數據處理兩種方式。

1 Broker

Kafka是一個高吞吐量分布式消息系統，采用Scala和Java語言編寫，它提供了快速、可擴展的、分布式、分區的和可復制的日志訂閱服務。它由Producer、Broker、Consumer三部分構成，如圖所示。

• Producer向某個Topic發布消息，
• 而Consumer訂閱某個Topic的消息。
• 一旦有某個Topic新產生的消息，Broker會傳遞給訂閱它的所有Consumer。
• ，每個Topic分為多個分區，這樣的設計有利於管理數據和負載均衡。

0. Broker：消息中間件處理結點，一個Kafka節點就是一個broker，多個broker可以組成一個Kafka集群。
1. Controller：中央控制器Control，負責管理分區和副本狀態並執行管理着這些分區的重新分配。（里面涉及到partition leader 選舉）
2. ISR：同步副本組

2 Topic

在Kafka中，消息是按Topic組織的.

1. Partition:topic物理上的分組，一個topic可以分為多個partition，每個partition是一個有序的隊列。
2. Segment：partition物理上由多個segment組成
3. offset：每個partition都由一系列有序的、不可變的消息組成，這些消息被連續的追加到partition中。partition中的每個消息都有一個連續的序列號叫做offset,用於partition唯一標識一條消息.

2.1 topic中partition存儲分布

在Kafka文件存儲中，同一個topic下有多個不同partition，每個partition為一個目錄，partiton命名規則為topic名稱+有序序號，第一個partiton序號從0開始，序號最大值為partitions數量減1。

├── data0
│   ├── cleaner-offset-checkpoint
│   ├── client_mblogduration-35
│   │   ├── 00000000000004909731.index
│   │   ├── 00000000000004909731.log // 1G文件--Segment
│   │   ├── 00000000000005048975.index // 數字是Offset
│   │   ├── 00000000000005048975.log
│   ├── client_mblogduration-37
│   │   ├── 00000000000004955629.index
│   │   ├── 00000000000004955629.log
│   │   ├── 00000000000005098290.index
│   │   ├── 00000000000005098290.log
│   ├── __consumer_offsets-33
│   │   ├── 00000000000000105157.index
│   │   └── 00000000000000105157.log
│   ├── meta.properties
│   ├── recovery-point-offset-checkpoint
│   └── replication-offset-checkpoint

• cleaner-offset-checkpoint:存了每個log的最后清理offset
• meta.properties: broker.id 信息
• recovery-point-offset-checkpoint:表示已經刷寫到磁盤的記錄。recoveryPoint以下的數據都是已經刷到磁盤上的了。
• replication-offset-checkpoint: 用來存儲每個replica的HighWatermark的(high watermark (HW)，表示已經被commited的message，HW以下的數據都是各個replicas間同步的，一致的。)

這些都是歸於LogManager使用。

[huangqiang@yz4098 data1]$ cat cleaner-offset-checkpoint
0
0
[huangqiang@yz4098 data1]$ cat recovery-point-offset-checkpoint
0
2
__consumer_offsets 23 148062
client_mblogduration 1 6244035
[huangqiang@yz4098 data1]$ cat replication-offset-checkpoint
0
2
__consumer_offsets 23 148062
client_mblogduration 1 6244092

2.2 partiton中文件存儲方式

• 每個partion(目錄)由多個大小相等segment(段)數據文件中。但每個段segment file消息數量不一定相等，這種特性方便old segment file快速被刪除。
• 每個partiton只需要支持順序讀寫就行了，segment文件生命周期由服務端配置參數決定。

[huangqiang@yz4098 kafka]$ ll -ah data9/client_mblogduration-18/
-rw-rw-r-- 1 kafka kafka  607K Nov 21 22:33 00000000000005046268.index
-rw-rw-r-- 1 kafka kafka  1.0G Nov 21 22:33 00000000000005046268.log
-rw-rw-r-- 1 kafka kafka  590K Nov 22 12:26 00000000000005188203.index
-rw-rw-r-- 1 kafka kafka  1.0G Nov 22 12:26 00000000000005188203.log

2.3 partiton中segment文件存儲結構

partion中segment file組成和物理結構。

• segment file組成：由2大部分組成，分別為index file和data file，此2個文件一一對應，成對出現，后綴".index"和“.log”分別表示為segment索引文件、數據文件.
• segment文件命名規則：partion全局的第一個segment從0開始，后續每個segment文件名為上一個segment文件最后一條消息的offset值。數值最大為64位long大小，19位數字字符長度，沒有數字用0填充。

以一對segment file文件為例，說明segment中index<—->data file對應關系物理結構如下

• Index文件存儲大量元數據，指向對應log文件中message的物理偏移地址。
• log數據文件存儲大量消息

其中以Index文件中元數據3,497為例，依次在數據文件中表示第3個message(在全局partiton表示第368772個message)、以及該消息的物理偏移地址為497。

下面看看segment data file的內部

segment data file由許多message組成，下面詳細說明message物理結構如下：

關鍵字	解釋說明
8 byte offset	該message在partition的offset
4 byte message size	message大小
4 byte CRC32	用crc32校驗message
1 byte "magic"	表示本次發布Kafka服務程序協議版本號
1 byte "attributes"	表示為獨立版本、或標識壓縮類型、或編碼類型。
4 byte key length	表示key的長度,當key為-1時，K byte key字段不填
K byte key	可選
value bytes payload	表示實際消息數據。

2.4 在partition中如何通過offset查找message

例如讀取offset=368776的message，需要通過下面2個步驟查找。

1. 第一步查找segment file

   • 上述圖2為例，其中00000000000000000000.index表示最開始的文件，起始偏移量(offset)為0.第二個文件00000000000000368769.index的消息量起始偏移量為368770 = 368769 + 1.同樣，第三個文件00000000000000737337.index的起始偏移量為737338=737337 + 1，其他后續文件依次類推，以起始偏移量命名並排序這些文件，只要根據offset 二分查找文件列表，就可以快速定位到具體文件。
     當offset=368776時定位到00000000000000368769.index|log

2. 第二步通過segment file查找message

   • 通過第一步定位到segment file，當offset=368776時，依次定位到00000000000000368769.index的元數據物理位置（這個較小，可以放在內存中，直接操作）和00000000000000368769.log的物理偏移地址，然后再通過00000000000000368769.log 順序查找 直到offset=368776為止。

從上述圖2.3節可知這樣做的優點，segment index file采取稀疏索引存儲方式，它減少索引文件大小，通過map可以直接內存操作，稀疏索引為數據文件的每個對應message設置一個元數據指針,它比稠密索引節省了更多的存儲空間，但查找起來需要消耗更多的時間。

2.5 讀寫message總結

寫message

• 消息從java堆轉入page cache(即物理內存)。
• 由異步線程刷盤,消息從page cache刷入磁盤。

讀message

• 消息直接從page cache轉入socket發送出去。
• 當從page cache沒有找到相應數據時，此時會產生磁盤IO,從磁
  盤Load消息到page cache,然后直接從socket發出去

Kafka高效文件存儲設計特點

• topic中一個parition大文件分成多個小文件段，通過多個小文件段，就容易定期清除或刪除已經消費完文件，減少磁盤占用。
• 通過索引信息可以快速定位message和確定response的最大大小。
• 通過index元數據全部映射到memory，可以避免segment file的IO磁盤操作。
• 通過索引文件稀疏存儲，可以大幅降低index文件元數據占用空間大小。

---

3 Producer

1. 同步、異步

4 Consumer

1. Partition
2. ConsumerGroup和Consumer
3. Coordinator