Kafka 工作流程?

Kafka 中消息是以 topic 進行分類的，生產者生產消息，消費者消費消息，都是面向 topic的。?
topic 是邏輯上的概念，而 partition 是物理上的概念，每個 partition 對應于一個 log 文件，該 log 文件中存儲的就是 producer 生產的數據。Producer 生產的數據會被不斷追加到該log 文件末端，且每條數據都有自己的 offset。消費者組中的每個消費者，都會實時記錄自己消費到了哪個 offset，以便出錯恢復時，從上次的位置繼續消費?

Kafka文件存儲機制

由于生產者生產的消息會不斷追加到 log 文件末尾，為防止 log 文件過大導致數據定位效率低下，Kafka 采取了分片和索引機制，將每個 partition 分為多個 segment。每個 segment對應兩個文件——“.index”文件和“.log”文件。這些文件位于一個文件夾下，該文件夾的命名規則為：topic 名稱+分區序號。例如，first 這個 topic 有三個分區，則其對應的文件夾為 first0,first-1,first-2。

[root@backup01 logs]# cd /usr/local/hadoop/kafka/kafka_2.12-2.4.1/logs/first-0 [root@backup01 first-0]# ll total 8 -rw-r--r--. 1 root root 10485760 Mar 29 16:32 00000000000000000000.index -rw-r--r--. 1 root root 146 Mar 29 17:34 00000000000000000000.log -rw-r--r--. 1 root root 10485756 Mar 29 16:32 00000000000000000000.timeindex -rw-r--r--. 1 root root 8 Mar 29 16:32 leader-epoch-checkpoint [root@backup01 first-0]#

00000000000000000000.index 00000000000000000000.log 00000000000000170410.index 00000000000000170410.log 00000000000000239430.index 00000000000000239430.log

index 和 log 文件以當前 segment 的第一條消息的 offset 命名。下圖為 index 文件和 log 文件的結構示意圖

“.index”文件存儲大量的索引信息，“.log”文件存儲大量的數據，索引文件中的元 數據指向對應數據文件中 message 的物理偏移地址。??

Kafka中的Message是以topic為基本單位組織的，不同的topic之間是相互獨立的。每個topic又可以分成幾個不同的partition(每個topic有幾個partition是在創建topic時指定的)，每個partition存儲一部分Message。借用官方的一張圖，可以直觀地看到topic和partition的關系。

partition是以文件的形式存儲在文件系統中，比如，創建了一個名為page_visits的topic，其有5個partition，那么在Kafka的數據目錄中(由配置文件中的log.dirs指定的)中就有這樣5個目錄: page_visits-0， page_visits-1，page_visits-2，page_visits-3，page_visits-4，其命名規則為<topic_name>-<partition_id>，里面存儲的分別就是這5個partition的數據。

接下來，本文將分析partition目錄中的文件的存儲格式和相關的代碼所在的位置。

Partition的數據文件

Partition中的每條Message由offset來表示它在這個partition中的偏移量，這個offset不是該Message在partition數據文件中的實際存儲位置，而是邏輯上一個值，它唯一確定了partition中的一條Message。因此，可以認為offset是partition中Message的id。partition中的每條Message包含了以下三個屬性：

? ?offset
? ? MessageSize
? ? data
其中offset為long型，MessageSize為int32，表示data有多大，data為message的具體內容。它的格式和Kafka通訊協議中介紹的MessageSet格式是一致。

Partition的數據文件則包含了若干條上述格式的Message，按offset由小到大排列在一起。它的實現類為FileMessageSet，類圖如下：

它的主要方法如下：

append: 把給定的ByteBufferMessageSet中的Message寫入到這個數據文件中。
searchFor: 從指定的startingPosition開始搜索找到第一個Message其offset是大于或者等于指定的offset，并返回其在文件中的位置Position。它的實現方式是從startingPosition開始讀取12個字節，分別是當前MessageSet的offset和size。如果當前offset小于指定的offset，那么將position向后移動LogOverHead+MessageSize（其中LogOverHead為offset+messagesize，為12個字節）。
read：準確名字應該是slice，它截取其中一部分返回一個新的FileMessageSet。它不保證截取的位置數據的完整性。
sizeInBytes: 表示這個FileMessageSet占有了多少字節的空間。
truncateTo: 把這個文件截斷，這個方法不保證截斷位置的Message的完整性。
readInto: 從指定的相對位置開始把文件的內容讀取到對應的ByteBuffer中。
我們來思考一下，如果一個partition只有一個數據文件會怎么樣？

新數據是添加在文件末尾（調用FileMessageSet的append方法），不論文件數據文件有多大，這個操作永遠都是O(1)的。
查找某個offset的Message（調用FileMessageSet的searchFor方法）是順序查找的。因此，如果數據文件很大的話，查找的效率就低。
那Kafka是如何解決查找效率的的問題呢？有兩大法寶：1) 分段 2) 索引。

數據文件的分段
Kafka解決查詢效率的手段之一是將數據文件分段，比如有100條Message，它們的offset是從0到99。假設將數據文件分成5段，第一段為0-19，第二段為20-39，以此類推，每段放在一個單獨的數據文件里面，數據文件以該段中最小的offset命名。這樣在查找指定offset的Message的時候，用二分查找就可以定位到該Message在哪個段中。

為數據文件建索引
數據文件分段使得可以在一個較小的數據文件中查找對應offset的Message了，但是這依然需要順序掃描才能找到對應offset的Message。為了進一步提高查找的效率，Kafka為每個分段后的數據文件建立了索引文件，文件名與數據文件的名字是一樣的，只是文件擴展名為.index。
索引文件中包含若干個索引條目，每個條目表示數據文件中一條Message的索引。索引包含兩個部分（均為4個字節的數字），分別為相對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也不能一次定位到其在數據文件的位置，從而需要做一次順序掃描，但是這次順序掃描的范圍就很小了。

在Kafka中，索引文件的實現類為OffsetIndex，它的類圖如下：

主要的方法有：

append方法，添加一對offset和position到index文件中，這里的offset將會被轉成相對的offset。
lookup, 用二分查找的方式去查找小于或等于給定offset的最大的那個offset
小結
我們以幾張圖來總結一下Message是如何在Kafka中存儲的，以及如何查找指定offset的Message的。

Message是按照topic來組織，每個topic可以分成多個的partition，比如：有5個partition的名為為page_visits的topic的目錄結構為：

partition是分段的，每個段叫LogSegment，包括了一個數據文件和一個索引文件，下圖是某個partition目錄下的文件：

可以看到，這個partition有4個LogSegment。

借用博主@lizhitao博客上的一張圖來展示是如何查找Message的。

比如：要查找絕對offset為7的Message：

首先是用二分查找確定它是在哪個LogSegment中，自然是在第一個Segment中。
打開這個Segment的index文件，也是用二分查找找到offset小于或者等于指定offset的索引條目中最大的那個offset。自然offset為6的那個索引是我們要找的，通過索引文件我們知道offset為6的Message在數據文件中的位置為9807。
打開數據文件，從位置為9807的那個地方開始順序掃描直到找到offset為7的那條Message。
這套機制是建立在offset是有序的。索引文件被映射到內存中，所以查找的速度還是很快的。

一句話，Kafka的Message存儲采用了分區(partition)，分段(LogSegment)和稀疏索引這幾個手段來達到了高效性。
?

總結

以上是生活随笔為你收集整理的大数据技术之 Kafka (第 3 章 Kafka 架构深入 ) Log存储解析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。