來源：https://my.oschina.net/zhangxufeng/blog/3114166
本文主要介紹kafka中日誌的存儲原理，主要內容包括kafka日誌存儲格式、日誌文件的管理方式、日誌索引文件的格式和日誌壓縮等功能。
作為一款消息系統，日誌就是將消息持久化到磁盤上的數據，這份數據的存儲方式將會極大的影響其吞吐量和擴展性，而kafka日誌由於其優秀的設計，為其實現這些特性提供了不可忽略的作用。
總結來說，kafka日誌主要具有如下特點：
極高的壓縮比例。kafka日誌不僅會對其key和value進行壓縮，而且還會對每條消息的偏移量、時間戳等等元數據信息進行壓縮；
batch的方式存儲數據。在存儲上，kafka日誌是以批次的方式進行數據的存儲，每個批次的大小默認為4KB，每個批次的元數據中會存儲其起始偏移量、時間戳和消息長度等信息；
追加的方式寫入數據。由於kafka日誌都是寫入磁盤的，而磁盤的順序寫入效率是非常高的，kafka寫入採用的就是追加的方式寫入消息，這樣可以避免磁頭的隨機移動，從而提升寫入速率；
使用索引文件提升查詢性能。kafka不僅會存儲消息日誌文件，還會為每個消息日誌文件創建一個索引文件，而且索引都是以batch為單位進行存儲的，也即只會為batch的起始位移和時間戳建立索引，而不會為每條消息都建立索引。
1. 日誌存儲格式
最新版本的kafka日誌是以批為單位進行日誌存儲的，所謂的批指的是kafka會將多條日誌壓縮到同一個batch中，然後以batch為單位進行後續的諸如索引的創建和消息的查詢等工作。
對於每個批次而言，其默認大小為4KB，並且保存了整個批次的起始位移和時間戳等元數據信息，而對於每條消息而言，其位移和時間戳等元數據存儲的則是相對於整個批次的元數據的增量，通過這種方式，kafka能夠減少每條消息中數據佔用的磁盤空間。
這裡我們首先展示一下每個批次的數據格式：
圖中消息批次的每個元數據都有固定的長度大小，而只有最後面的消息個數的是可變的。如下是batch中主要的屬性的含義：
起始位移：佔用8字節，其存儲了當前batch中第一條消息的位移；
長度：佔用了4個字節，其存儲了整個batch所佔用的磁盤空間的大小，通過該字段，kafka在進行消息遍歷的時候，可以快速的跳躍到下一個batch進行數據讀取；
分區leader版本號：記錄了當前消息所在分區的leader的服務器版本，主要用於進行一些數據版本的校驗和轉換工作；
CRC：對當前整個batch的數據的CRC校驗碼，主要是用於對數據進行差錯校驗的；
屬性：佔用2個字節，這個字段的最低3位記錄了當前batch中消息的壓縮方式，現在主要有GZIP、LZ4和Snappy三種。第4位記錄了時間戳的類型，第5和6位記錄了新版本引入的事務類型和控制類型；
最大位移增量：最新的消息的位移相對於第一條消息的唯一增量；
起始時間戳：佔用8個字節，記錄了batch中第一條消息的時間戳；
最大時間戳：佔用8個字節，記錄了batch中最新的一條消息的時間戳；
PID、producer epoch和起始序列號：這三個參數主要是為了實現事務和冪等性而使用的，其中PID和producer epoch用於確定當前producer是否合法，而起始序列號則主要用於進行消息的冪等校驗；
消息個數：佔用4個字節，記錄當前batch中所有消息的個數；
通過上面的介紹可以看出，每個batch的頭部數據中佔用的字節數固定為61個字節，可變部分主要是與具體的消息有關，下面我們來看一下batch中每條消息的格式：
這裡的消息的頭部數據就與batch的大不相同，可以看到，其大部分數據的長度都是可變的。既然是可變的，這裡我們需要強調兩個問題：
1、對於數字的存儲，kafka採用的是Zig-Zag的存儲方式，也即負數並不會使用補碼的方式進行編碼，而是將其轉換為對應的正整數，比如-1映射為1、1映射為2、-2映射為3、2映射為4，關係圖如下所示：
通過圖可以看出，在對數據反編碼的時候，我們只需要將對應的整數轉換成其原始值即可；
2、在使用Zig-Zag編碼方式的時候，每個字節最大為128，而其中一半要存儲正數，一半要存儲負數，還有一個0，也就是說每個字節能夠表示的最大整數為64，此時如果有大於64的數字，kafka就會使用多個字節進行存儲。
而這多個字節的表徵方式是通過將每個字節的最大位作為保留位來實現的，如果最高位為1，則表示需要與後續字節共同表徵目標數字，如果最高位為0，則表示當前位即可表示目標數字。
kafka使用這種編碼方式的優點在於，大部分的數據增量都是非常小的數字，因此使用一個字節即可保存，這比直接使用原始類型的數據要節約大概七倍的內存。
對於上面的每條消息的格式，除了消息key和value相關的字段，其還有屬性字段和header，屬性字段的主要作用是存儲當前消息key和value的壓縮方式，而header則供給用戶進行添加一些動態的屬性，從而實現一些定製化的工作。
通過對kafka消息日誌的存儲格式我們可以看出，其使用batch的方式將一些公共信息進行提取，從而保證其只需要存儲一份，雖然看起來每個batch的頭部信息比較多，但其平攤到每條消息上之後使用的字節更少了；
在消息層面，kafka使用了數據增量的方式和Zig-Zag編碼方式對數據進行的壓縮，從而極大地減少其佔用的字節數。總體而言，這種存儲方式極大的減少了kafka佔用的磁盤空間大小。
2. 日誌存儲方式
在使用kafka時，消息都是推送到某個topic中，然後由producer計算當前消息會發送到哪個partition，在partition中，kafka會為每條消息設置一個偏移量，也就是說，如果要唯一定位一條消息，使用<topic>三元組即可。/<topic>
基於kafka的架構模式，其會將各個分區平均分配到每個broker中，也就是說每個broker會被分配用來提供一個或多個分區的日誌存儲服務。在broker服務器上，kafka的日誌也是按照partition進行存儲的，其會在指定的日誌存儲目錄中為每個topic的partition分別創建一個目錄，目錄中存儲的就是這些分區的日誌數據，而目錄的名稱則會以<topic-patition>的格式進行創建。如下是kafka日誌的存儲目錄示意圖：/<topic-patition>
這裡我們需要注意的是，圖中對於分區日誌的存儲，當前broker只會存儲分配給其的分區的日誌，比如圖中的connect-status就只有分區1和分區4的目錄，而沒有分區2和分區3的目錄，這是因為這些分區被分配在了集群的其他節點上。
在每個分區日誌目錄中，存在有三種類型的日誌文件，即後綴分別為log、index和timeindex的文件。其中log文件就是真正存儲消息日誌的文件，index文件存儲的是消息的位移索引數據，而timeindex文件則存儲的是時間索引數據。
如下圖所示為一個分區的消息日誌數據：
從圖中可以看出，每種類型的日誌文件都是分段的，這裡關於分段的規則主要有如下幾點需要說明：
在為日誌進行分段時，每個文件的文件名都是以該段中第一條消息的位移的偏移量來命名的；
kafka會在每個log文件的大小達到1G的時候關閉該文件，而新開一個文件進行數據的寫入。可以看到，圖中除了最新的log文件外，其餘的log文件的大小都是1G；
對於index文件和timeindex文件，在每個log文件進行分段之後，這兩個索引文件也會進行分段，這也就是它們的文件名與log文件一致的原因；
kafka日誌的留存時間默認是7天，也就是說，kafka會刪除存儲時間超過7天的日誌，但是對於某些文件，其部分日誌存儲時間未達到7天，部分達到了7天，此時還是會保留該文件，直至其所有的消息都超過留存時間；
3. 索引文件
kafka主要有兩種類型的索引文件：位移索引文件和時間戳索引文件。位移索引文件中存儲的是消息的位移與該位移所對應的消息的物理地址；時間戳索引文件中則存儲的是消息的時間戳與該消息的位移值。
也就是說，如果需要通過時間戳查詢消息記錄，那麼其首先會通過時間戳索引文件查詢該時間戳對應的位移值，然後通過位移值在位移索引文件中查詢消息具體的物理地址。關於位移索引文件，這裡有兩點需要說明：
1、由於kafka消息都是以batch的形式進行存儲，因而索引文件中索引元素的最小單元是batch，也就是說，通過位移索引文件能夠定位到消息所在的batch，而沒法定位到消息在batch中的具體位置，查找消息的時候，還需要進一步對batch進行遍歷；
2、位移索引文件中記錄的位移值並不是消息真正的位移值，而是該位移相對於該位移索引文件的起始位移的偏移量，通過這種方式能夠極大的減小位移索引文件的大小。
如下圖所示為一個位移索引文件的格式示意圖：
如下則是具體的位移索引文件的示例：
關於時間戳索引文件，由於時間戳的變化比位移的變化幅度要大一些，其即使採用了增量的方式存儲時間戳索引，但也沒法有效地使用Zig-Zag方式對數據進行編碼，因而時間戳索引文件是直接存儲的消息的時間戳數據，
但是對於時間戳索引文件中存儲的位移數據，由於其變化幅度不大，因而其還是使用相對位移的方式進行的存儲，並且這種存儲方式也可以直接映射到位移索引文件中而無需進行計算。如下圖所示為時間戳索引文件的格式圖：
如下則是時間戳索引文件的一個存儲示例：
可以看到，如果需要通過時間戳來定位消息，就需要首先在時間戳索引文件中定位到具體的位移，然後通過位移在位移索引文件中定位到消息的具體物理地址。
4. 日誌壓縮
所謂的日誌壓縮功能，其主要是針對這樣的場景的，比如對某個用戶的郵箱數據進行修改，其總共修改了三次，修改過程如下：
[email protected]
[email protected]
[email protected]
在這麼進行修改之後，很明顯，我們主要需要關心的是最後一次修改，因為其是最終數據記錄，但是如果我們按順序處理上述消息，則需要處理三次消息。
kafka的日誌壓縮就是為了解決這個問題而存在的，對於使用相同key的消息，其會只保留最新的一條消息的記錄，而中間過程的消息都會被kafka cleaner給清理掉。
但是需要注意的是，kafka並不會清理當前處於活躍狀態的日誌文件中的消息記錄。所謂當前處於活躍狀態的日誌文件，也就是當前正在寫入數據的日誌文件。如下圖所示為一個kafka進行日誌壓縮的示例圖：
圖中K1的數據有V1、V3和V4，經過壓縮之後只有V4保留了下來，K2的數據則有V2、V6和V10，壓縮之後也只有V10保留了下來；同理可推斷其他的Key的數據。
另外需要注意的是，kafka開啟日誌壓縮使用的是log.cleanup.policy，其默認值為delete，也即我們正常使用的策略，如果將其設置為compaction，則開啟了日誌壓縮策略，但是需要注意的是，開啟了日誌壓縮策略並不代表kafka會清理歷史數據，只有將log.cleaner.enable設置為true才會定時清理歷史數據。
在kafka中，其本身也在使用日誌壓縮策略，主要體現在kafka消息的偏移量存儲。在舊版本中，kafka將每個consumer分組當前消費的偏移量信息保存在zookeeper中，但是由於zookeeper是一款分佈式協調工具，其對於讀操作具有非常高的性能，但是對於寫操作性能比較低，而consumer的位移提交動作是非常頻繁的，這勢必會導致zookeeper成為kafka消息消費的瓶頸。
因而在最新版本中，kafka將分組消費的位移數據存儲在了一個特殊的topic中，即__consumer_offsets，由於每個分組group的位移信息都會提交到該topic，因而kafka默認為其設置了非常多的分區，也即50個分區。
另外，consumer在提交位移時，使用的key為groupId+topic+partition，而值則為當前提交的位移，也就是說，對於每一個分組所消費的topic的partition，其都只會保留最新的位移。如果consumer需要讀取位移，那麼只需要按照上述格式組裝key，然後在該topic中讀取最新的消息數據即可。
5. 小結
本文首先對kafka的日誌設計的優點進行了介紹，然後著重講解了日誌存儲格式、日誌目錄規劃、日誌索引文件的格式以及日誌壓縮的功能。
閱讀更多 程序員聖經 的文章
