柚子快報激活碼778899分享：后端 Kafka 物理存儲機制

http://yzkb.51969.com/

優(yōu)質(zhì)博文：IT-BLOG-CN

一個商業(yè)化消息隊列的性能好壞，其文件存儲機制設(shè)計是衡量一個消息隊列服務(wù)技術(shù)水平和最關(guān)鍵指標(biāo)之一。下面將從Kafka文件存儲機制和物理結(jié)構(gòu)角度，分析Kafka是如何實現(xiàn)高效文件存儲，及實際應(yīng)用效果。Kafka的基本存儲單位是分區(qū)。在配置Kafka的時候，管理員指定了一個用于存儲分區(qū)的目錄清單log.dirs參數(shù)的值。

一、分區(qū)分配

創(chuàng)建主題時，Kafka首先決定如何在broker之間分配分區(qū)。假設(shè)有6個broker，打算創(chuàng)建一個包含10個分區(qū)的主題。并且復(fù)制系數(shù)是3，相當(dāng)于30個分區(qū)副本。在被分配到6個broker上時，要達到如下的目標(biāo)： 【1】在broker間平均分配分區(qū)副本。對于上述例子來說，就是要保證每個broker可以分到5個副本。 【2】確保每個分區(qū)的每個副本分布在不同的broker上。 【3】如果為broker指定了機架信息，那么盡可能把每個分區(qū)的副本分配到不同機架的broker上。這樣做是為了保證一個機架不可用不會導(dǎo)致整個分區(qū)不可用。

為了實現(xiàn)這個目標(biāo)，我們先隨機選擇一個broker（假設(shè)是2），然后通過輪詢給每個broker分配分區(qū)來確定首領(lǐng)的位置。如果分區(qū)0的首領(lǐng)在broker2上，那么分區(qū)1的首領(lǐng)就在broker3上，以此類推。然后，從分區(qū)首領(lǐng)開始，以此分配跟隨者副本。如分區(qū)0首領(lǐng)在broker2上，那么它的第一個副本會出現(xiàn)在broker3上，第二個出現(xiàn)在 broker4上。如果配置了機架信息，那么就不是按照數(shù)字順序來選擇broker了，而是按照交替機架的方式來選擇broker。假設(shè)broker0、broker1、broker2放在同一個機架，broker3、broker4、broker5放在其他不同的機架。此時就不是按照0到5的順序來選擇broker，而是按照0,3,1,4,2,5的順序進行選擇的。

二、文件管理

保留數(shù)據(jù)時Kafka的一個基本特性，Kafka不會一直保留數(shù)據(jù)，也不會等到所有消費者都讀取了消息之后才刪除消息。相反，Kafka管理員為每個主題配置了數(shù)據(jù)保留期限，規(guī)定數(shù)據(jù)被刪除之前可以保留多長時間，或者清理數(shù)據(jù)之前可以保留的數(shù)據(jù)量大小。 因為在一個大文件里查找和刪除消息是很費時的，也很容易出錯，所以我們把分區(qū)分成若干個片段。默認(rèn)情況下，index大小為10M，每個片段log包含1GB或一周數(shù)據(jù)，以較小的那個為準(zhǔn)。當(dāng)前正在寫入數(shù)據(jù)的片段叫做活躍片段，活躍片段永遠不會被刪除。

三、文件格式

我們把Kafka的消息和偏移量保存在文件里。保存在磁盤上的數(shù)據(jù)格式與從生產(chǎn)者發(fā)送過來或者發(fā)送給消費者的消息格式是一樣的。因為使用相同的消息格式進行磁盤存儲和網(wǎng)絡(luò)傳輸，Kafka可以使用零復(fù)制技術(shù)給消費者發(fā)送消息，同時避免了對生產(chǎn)者已經(jīng)壓縮過的消息進行解壓縮。除了鍵、值和偏移量外，消息里還包含了消息大小、校驗和、消息格式版本號、壓縮算法和時間戳。時間戳可以是生產(chǎn)者發(fā)送消息的時間，也可以是消息到達broker的時間，這個是可配置的。如果生產(chǎn)者發(fā)送的是壓縮過的消息，那么同一個批次的消息會被再壓縮一次，被當(dāng)做包裝消息進行發(fā)送。下面是普通消息和包裝消息圖：

四、文件存儲機制

【1】Broker： 消息中間件處理結(jié)點，一個Kafka節(jié)點就是一個Broker，多個Broker可以組成一個Kafka集群。 【2】Topic： 主題，如page view日志、click日志等都可以以Topic的形式存在，Kafka集群能夠同時負(fù)責(zé)多個Topic的分發(fā)。 【3】Partition： Topic物理上的分組，一個Topic可以分為多個Partition，每個Partition是一個有序的隊列。 【4】Segment： Partition物理上由多個Segment組成。 【5】offset： 每個 Partition都由一系列有序的、不可變的消息組成，這些消息被連續(xù)的追加到Partition中。Partition中的每個消息都有一個連續(xù)的序列號叫做offset，用于Partition唯一標(biāo)識一條消息。

分析過程分為以下4個步驟：

【1】Topic中Partition存儲分布： 假設(shè)Kafka集群只有一個Broker，xxx/message-folder為數(shù)據(jù)文件存儲根目錄，在Kafka Broker中server.properties文件配置(參數(shù)log.dirs=xxx/message-folder)，例如創(chuàng)建2個Topic名稱分別為report_push、launch_info，Partitions數(shù)量都為partitions=4（將一個Topic分為4個部分存儲）存儲路徑和目錄規(guī)則為：

xxx/message-folder

|--report_push-0

|--report_push-1

|--report_push-2

|--report_push-3

|--launch_info-0

|--launch_info-1

|--launch_info-2

|--launch_info-3

【2】Partiton中文件存儲方式： 每個Partion（目錄）相當(dāng)于一個巨型文件被平均分配到多個大小相等Segment（段）數(shù)據(jù)文件中。但每個段Segment file消息數(shù)量不一定相等，這種特性方便old segment file快速被刪除。每個Partiton只需要支持順序讀寫就行了，Segment文件生命周期由服務(wù)端配置參數(shù)決定。這樣做的好處就是能快速刪除無用文件，有效提高磁盤利用率。 ?

【3】partiton中segment文件存儲結(jié)構(gòu)：segment file由2大部分組成，分別為index file和data file，此2個文件成對出現(xiàn)，后綴".index"和“.log”分別表示為segment索引文件、數(shù)據(jù)文件。segment文件命名規(guī)則：partion全局的第一個segment從0開始，后續(xù)每個segment文件名為上一個segment文件最后一條消息的offset值。數(shù)值最大為64位long大小，19位數(shù)字字符長度，沒有數(shù)字用0填充。下面文件列表是筆者在Kafka broker上做的一個實驗，創(chuàng)建一個topicXXX包含1 partition，設(shè)置每個segment大小為500MB，并啟動producer向Kafka broker寫入大量數(shù)據(jù)，如下圖所示segment文件列表形象說明了上述2個規(guī)則以及segment中index<—->data file對應(yīng)關(guān)系物理結(jié)構(gòu)如下：

索引文件存儲大量元數(shù)據(jù)，數(shù)據(jù)文件存儲大量消息，索引文件中元數(shù)據(jù)指向?qū)?yīng)數(shù)據(jù)文件中message的物理偏移地址。 其中以索引文件中元數(shù)據(jù)3,497為例，依次在數(shù)據(jù)文件中表示第3個message（在全局partiton表示第368772個message）、以及該消息的物理偏移地址為497。從上述圖3了解到segment data file由許多 message組成，下面詳細說明message物理結(jié)構(gòu)如下：

【參數(shù)說明】： 8 byte offset：在Parition（分區(qū)）內(nèi)的每條消息都有一個有序的id號，這個id號被稱為偏移offset，它可以唯一確定每條消息在Parition內(nèi)的位置。即offset表示Partiion的第多少message。 4 byte message size：message大??； 4 byte CRC32：用crc32校驗message； 1 byte “magic"：表示本次發(fā)布Kafka服務(wù)程序協(xié)議版本號； 1 byte “attributes"：表示為獨立版本、或標(biāo)識壓縮類型、或編碼類型； 4 byte key length：表示key的長度，當(dāng)key為-1時，K byte key字段不填； value bytes payload：表示實際消息數(shù)據(jù)；

`index文件結(jié)構(gòu)：

offset: 783932 position: 69483992

offset: 784323 position: 69543233

offset: 784565 position: 69589443

offset: 784932 position: 69623433

offset: 785355 position: 69658994

offset: 785894 position: 69704355

offset: 786389 position: 69738993

offset: 786584 position: 69784345

log文件結(jié)構(gòu)： 有個眼緣即可

offset: 784932 CreateTime:1598161852389 keysize: -1 valuesize: 15 sequence: 9884 baseOffset: 7043213 lastOffset: 784932 count: 1 baseSequence: 907

【4】在partition中如何通過offset查找message： 例如讀取offset=368776的Message，需要通過下面2個步驟查找。 【第一步】查找segment file： 上圖為例，其中00000000000000000000.index表示最開始的文件，起始偏移量offset為0。第二個文件00000000000000368769.index的消息量起始偏移量為368770 = 368769 + 1。同樣，第三個文件00000000000000737337.index的起始偏移量為737338=737337 + 1，其他后續(xù)文件依次類推，以起始偏移量命名并排序這些文件，只要根據(jù)offset二分查找文件列表，就可以快速定位到具體文件。當(dāng)offset=368776時定位到00000000000000368769.index|log； 【第二步】通過segment file查找message： 通過第一步定位到segment file，當(dāng)offset=368776時，依次定位到00000000000000368769.index 的元數(shù)據(jù)物理位置和00000000000000368769.log的物理偏移地址，然后再通過00000000000000368769.log順序查找直到offset=368776為止。從上述圖可知這樣做的優(yōu)點，segment index file采取稀疏索引存儲方式，它減少索引文件大小，通過mmap可以直接內(nèi)存操作，稀疏索引為數(shù)據(jù)文件的每個對應(yīng)message設(shè)置一個元數(shù)據(jù)指針，它比稠密索引節(jié)省了更多的存儲空間，但查找起來需要消耗更多的時間。通過上述過程詳細分析，我們就可以清楚認(rèn)識到kafka文件存儲機制的奧秘。

五、Kafka文件實際運行效果

【實驗環(huán)境】：Kafka集群 = 由2臺虛擬機組成；CPU = 4核；物理內(nèi)存= 8GB；網(wǎng)卡 = 千兆網(wǎng)卡；JVM HEAP = 4GB；詳細Kafka服務(wù)端配置及其優(yōu)化請參考：Kafka server.properties配置詳解：

從上述圖可以看出，Kafka運行時很少有大量讀磁盤的操作，主要是定期批量寫磁盤操作，因此操作磁盤很高效。這跟Kafka文件存儲中讀寫message的設(shè)計是息息相關(guān)的。Kafka中讀寫message有如下特點：寫message，消息從java堆轉(zhuǎn)入page cache即物理內(nèi)存。由異步線程刷盤，消息從page cache刷入磁盤。讀message消息直接從page cache轉(zhuǎn)入socket發(fā)送出去。當(dāng)從page cache沒有找到相應(yīng)數(shù)據(jù)時，此時會產(chǎn)生磁盤IO，從磁盤Load消息到page cache，然后直接從socket發(fā)出去。

六、Kafka 中的 Partition 和 Offset

【1】Log機制： 說到分區(qū)，就要說Kafka對消息的存儲，首先，kafka是通過log(日志)來記錄消息發(fā)布的。每當(dāng)產(chǎn)生一個消息，Kafka會記錄到本地的log文件中，這個log和我們平時的log有一定的區(qū)別。這里可以參考一下The Log，不多解釋。這個log文件默認(rèn)的位置在config/server.properties中指定的，默認(rèn)的位置是log.dirs=/tmp/kafka-logs，Linux不用說，windows的話就在你對應(yīng)磁盤的根目錄下。

分區(qū)Partition： Kafka是為分布式環(huán)境設(shè)計的，因此如果日志文件，其實也可以理解成消息數(shù)據(jù)庫，放在同一個地方，那么必然會帶來可用性的下降，一掛全掛，如果全量拷貝到所有的機器上，那么數(shù)據(jù)又存在過多的冗余，而且由于每臺機器的磁盤大小是有限的，所以即使有再多的機器，可處理的消息還是被磁盤所限制，無法超越當(dāng)前磁盤大小。因此有了Partition的概念。Kafka對消息進行一定的計算，通過hash來進行分區(qū)。這樣，就把一份log文件分成了多份。如上面的分區(qū)讀寫日志圖，分成多份以后，在單臺Broker上，比如快速上手中，如果新建Topic的時候，我們選擇replication-factor 1 partitions 2，那么在log目錄里，我們會看到test-0目錄和test-1目錄就是兩個分區(qū)了。你可能會想，這沒啥區(qū)別呀。注意，當(dāng)有了多個broker之后，這個意義就存在了。這里上一張圖：

【2】Kafka分布式分區(qū)存儲： 這是一個Topic包含4個Partition，2 Replication(拷貝)，也就是說全部的消息被放在了4個分區(qū)存儲，為了高可用，將4個分區(qū)做了2份冗余，然后根據(jù)分配算法。將總共8份數(shù)據(jù)，分配到Broker集群上。結(jié)果就是每個Broker上存儲的數(shù)據(jù)比全量數(shù)據(jù)要少，但每份數(shù)據(jù)都有冗余，這樣，一旦一臺機器宕機，并不影響使用。比如圖中的Broker1，宕機了那么剩下的三臺Broker依然保留了全量的分區(qū)數(shù)據(jù)。所以還能使用，如果再宕機一臺，那么數(shù)據(jù)不完整了。當(dāng)然你可以設(shè)置更多的冗余，比如設(shè)置了冗余是4，那么每臺機器就有了0123完整的數(shù)據(jù)，宕機幾臺都行。需要在存儲占用和高可用之間做衡量。至于宕機后，zookeeper會選出新的 partition leader。

偏移offset： 上一段說了分區(qū)，分區(qū)就是一個有序的，不可變的消息隊列。新來的commit log持續(xù)往后面加數(shù)據(jù)。這些消息被分配了一個下標(biāo)(或者偏移)，就是offset，用來定位這一條消息。消費者消費到了哪條消息，是保持在消費者這一端的。消息者也可以控制，消費者可以在本地保存最后消息的offset，并間歇性的向zookeeper注冊offset。也可以重置offset。

如何通過offset算出分區(qū)：其實Partition存儲的時候，又分成了多個segment(段)，然后通過一個index索引，來標(biāo)識第幾段。這里先可以去看一下本地log目錄的分區(qū)文件夾。在我這里，test-0這個分區(qū)里面，會有一個index文件和一個log文件，對于某個指定的分區(qū)，假設(shè)每5個消息作為一個段大小，當(dāng)產(chǎn)生了10條消息的情況下，目前有會分段。 0.index(表示這里index是對0-4做的索引)、5.index (表示這里index是對5-9做的索引)、10.index (表示這里index是對10-15做的索引，目前還沒滿) 和0.log、5.log、10.log。當(dāng)消費者需要讀取offset=8的時候，首先kafka對index文件列表進行二分查找，可以算出應(yīng)該是在5.index對應(yīng)的log文件中，然后對對應(yīng)的5.log文件，進行順序查找，5->6->7->8，直到順序找到8就好了。

七、索引

消費者可以從Kafka的任意可用偏移量位置開始讀取消息，假設(shè)消費者要讀取從偏移量100開始的1MB消息，那么Broker必須立即定位到偏移量100，為了幫組broker更快地定位到指定的偏移量，Kafka為每個分區(qū)維護一個索引。索引把偏移量映射到片段文件和偏移量在文件里的位置。索引也被分成片段，所以再刪除消息時，也可以刪除相應(yīng)的索引。Kafka不維護索引的校驗和。如果索引出現(xiàn)損壞，Kafka會通過重新讀取消息并錄制偏移量和位置來重新生成索引。如果有必要，管理員是可以刪除索引的，這樣做是絕對安全的，Kafka會自動重新生成這些索引。

八、Kafka高效文件存儲設(shè)計特點

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

柚子快報激活碼778899分享：后端 Kafka 物理存儲機制

http://yzkb.51969.com/

推薦鏈接