Kafka offset

引言
Kafka中的Message是以topic爲基本單位組織的，不一樣的topic之間是相互獨立的。每一個topic又能夠分紅幾個不一樣的partition(每一個topic有幾個partition是在建立topic時指定的)，每一個partition存儲一部分Message。借用官方的一張圖，能夠直觀地看到topic和partition的關係。
Anatomy of a Topic

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包含了如下三個屬性：

1.offset

2.MessageSize

3.data

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

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

FileMessageSet類圖
它的主要方法以下：

1.append: 把給定的ByteBufferMessageSet中的Message寫入到這個數據文件中。

2.searchFor: 從指定的startingPosition開始搜索找到第一個Message其offset是大於或者等於指定的offset，並返回其在文件中的位置Position。它的實現方式是從startingPosition開始讀取12個字節，分別是當前MessageSet的offset和size。若是當前offset小於指定的offset，那麼將position向後移動LogOverHead+MessageSize（其中LogOverHead爲offset+messagesize，爲12個字節）。

3.read：準確名字應該是slice，它截取其中一部分返回一個新的FileMessageSet。它不保證截取的位置數據的完整性。

4.sizeInBytes: 表示這個FileMessageSet佔有了多少字節的空間。

5.truncateTo: 把這個文件截斷，這個方法不保證截斷位置的Message的完整性。

6.readInto: 從指定的相對位置開始把文件的內容讀取到對應的ByteBuffer中。

咱們來思考一下，若是一個partition只有一個數據文件會怎麼樣？

1.新數據是添加在文件末尾（調用FileMessageSet的append方法），不論文件數據文件有多大，這個操做永遠都是O(1)的。

2.查找某個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。

1.相對offset：由於數據文件分段之後，每一個數據文件的起始offset不爲0，相對offset表示這條Message相對於其所屬數據文件中最小的offset的大小。舉例，分段後的一個數據文件的offset是從20開始，那麼offset爲25的Message在index文件中的相對offset就是25-20 = 5。存儲相對offset能夠減少索引文件佔用的空間。

2.position，表示該條Message在數據文件中的絕對位置。只要打開文件並移動文件指針到這個position就能夠讀取對應的Message了。

index文件中並無爲數據文件中的每條Message創建索引，而是採用了稀疏存儲的方式，每隔必定字節的數據創建一條索引。這樣避免了索引文件佔用過多的空間，從而能夠將索引文件保留在內存中。但缺點是沒有創建索引的Message也不能一次定位到其在數據文件的位置，從而須要作一次順序掃描，可是此次順序掃描的範圍就很小了。

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

主要的方法有：

1.append方法，添加一對offset和position到index文件中，這裏的offset將會被轉成相對的offset。

2.lookup, 用二分查找的方式去查找小於或等於給定offset的最大的那個offset

小結
咱們以幾張圖來總結一下Message是如何在Kafka中存儲的，以及如何查找指定offset的Message的。

Message是按照topic來組織，每一個topic能夠分紅多個的partition，好比：有5個partition的名爲爲page_visits的topic的目錄結構爲：
topic_partition

partition是分段的，每一個段叫LogSegment，包括了一個數據文件和一個索引文件，下圖是某個partition目錄下的文件：
partition
能夠看到，這個partition有4個LogSegment。

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

好比：要查找絕對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的Log存儲解析