kafka中間件-數(shù)據(jù)存儲（1）

摘要：

Kafka這款分布式消息隊列使用文件系統(tǒng)和操作系統(tǒng)的頁緩存（page cache）分別存儲和緩存消息，摒棄了Java的堆緩存機制，同時將隨機寫操作改為順序?qū)?，再結(jié)合Zero-Copy的特性極大地改善了IO性能?！耙粔KSATA RAID-5陣列磁盤的線性寫速度可以達到幾百M/s，而隨機寫的速度只能是100多KB/s，線性寫的速度是隨機寫的上千倍”，由此可以看出對磁盤寫消息的速度快慢關(guān)鍵還是取決于我們的使用方法。
Kafka也能夠通過配置讓用戶自己決定已經(jīng)落盤的持久化消息保存的時間，提供消息處理更為靈活的方式。本文將主要介紹Kafka中數(shù)據(jù)的存儲消息結(jié)構(gòu)、存儲方式以及如何通過offset來查找消息等內(nèi)容。

-

-

1、Kafka中幾個重要概念介紹

（1）Broker：消息中間件處理節(jié)點，一個Kafka節(jié)點就是一個broker，一個或者多個Broker可以組成一個Kafka集群；
（2）Topic：主題是對一組消息的抽象分類，比如例如page view日志、click日志等都可以以topic的形式進行抽象劃分類別。在物理上，不同Topic的消息分開存儲，邏輯上一個Topic的消息雖然保存于一個或多個broker上但用戶只需指定消息的Topic即可使得數(shù)據(jù)的生產(chǎn)者或消費者不必關(guān)心數(shù)據(jù)存于何處；
（3）Partition：每個主題又被分成一個或者若干個分區(qū)（Partition）。每個分區(qū)在本地磁盤上對應(yīng)一個文件夾，分區(qū)命名規(guī)則為主題名稱后接“—”連接符，之后再接分區(qū)編號，分區(qū)編號從0開始至分區(qū)總數(shù)減-1；
（4）LogSegment：每個分區(qū)又被劃分為多個日志分段（LogSegment）組成，日志段是Kafka日志對象分片的最小單位；LogSegment算是一個邏輯概念，對應(yīng)一個具體的日志文件（“.log”的數(shù)據(jù)文件）和兩個索引文件（“.index”和“.timeindex”，分別表示偏移量索引文件和消息時間戳索引文件）組成；
（5）Offset：每個partition中都由一系列有序的、不可變的消息組成，這些消息被順序地追加到partition中。每個消息都有一個連續(xù)的序列號稱之為offset—偏移量，用于在partition內(nèi)唯一標(biāo)識消息（并不表示消息在磁盤上的物理位置）；
（6）Message：消息是Kafka中存儲的最小最基本的單位，即為一個commit log，由一個固定長度的消息頭和一個可變長度的消息體組成；

2、 Kafka的日志結(jié)構(gòu)與數(shù)據(jù)存儲

Kafka中的消息是以主題（Topic）為基本單位進行組織的，各個主題之間相互獨立。在這里主題只是一個邏輯上的抽象概念，而在實際數(shù)據(jù)文件的存儲中，Kafka中的消息存儲在物理上是以一個或多個分區(qū)（Partition）構(gòu)成，每個分區(qū)對應(yīng)本地磁盤上的一個文件夾，每個文件夾內(nèi)包含了日志索引文件（“.index”和“.timeindex”）和日志數(shù)據(jù)文件（“.log”）兩部分。分區(qū)數(shù)量可以在創(chuàng)建主題時指定，也可以在創(chuàng)建Topic后進行修改。（ps：Topic的Partition數(shù)量只能增加而不能減少，這點內(nèi)容超出本篇幅的減少范圍，大家可以先思考下）。
在Kafka中正是因為使用了分區(qū)（Partition）的設(shè)計模型，通過將主題（Topic）的消息打散到多個分區(qū)，并分布保存在不同的Kafka Broker節(jié)點上實現(xiàn)了消息處理的高吞吐量。其生產(chǎn)者和消費者都可以多線程地并行操作，而每個線程處理的是一個分區(qū)的數(shù)據(jù)。

同時，Kafka為了實現(xiàn)集群的高可用性，在每個Partition中可以設(shè)置有一個或者多個副本（Replica），分區(qū)的副本分布在不同的Broker節(jié)點上。同時，從副本中會選出一個副本作為Leader，Leader副本負(fù)責(zé)與客戶端進行讀寫操作。而其他副本作為Follower會從Leader副本上進行數(shù)據(jù)同步。

2.1、Kafka中分區(qū)/副本的日志文件存儲分析

創(chuàng)建3副本的topic

./kafka-topics.sh --create --zookeeper 10.154.0.73:2181 --replication-factor 3 --partitions  3 --topic kafka-topic-01

./kafka-topics.sh --describe --zookeeper 10.154.0.73:2181 --topic kafka-topic-01

每個分區(qū)在物理上對應(yīng)一個文件夾，分區(qū)的命名規(guī)則為主題名后接“—”連接符，之后再接分區(qū)編號，分區(qū)編號從0開始，編號的最大值為分區(qū)總數(shù)減1。每個分區(qū)又有1至多個副本，分區(qū)的副本分布在集群的不同代理上，以提高可用性。從存儲的角度上來說，分區(qū)的每個副本在邏輯上可以抽象為一個日志（Log）對象，即分區(qū)副本與日志對象是相對應(yīng)的。下圖是在三個Kafka Broker節(jié)點所組成的集群中分區(qū)的主/備份副本的物理分布情況圖：

2.2、Kafka中日志索引和數(shù)據(jù)文件的存儲結(jié)構(gòu)

在Kafka中，每個Log對象又可以劃分為多個LogSegment文件，每個LogSegment文件包括一個日志數(shù)據(jù)文件和兩個索引文件（偏移量索引文件和消息時間戳索引文件）。其中，每個LogSegment中的日志數(shù)據(jù)文件大小均相等（該日志數(shù)據(jù)文件的大小可以通過在Kafka Broker的config/server.properties配置文件的中的“l(fā)og.segment.bytes”進行設(shè)置，默認(rèn)為1G大?。?073741824字節(jié)），在順序?qū)懭胂r如果超出該設(shè)定的閾值，將會創(chuàng)建一組新的日志數(shù)據(jù)和索引文件）。
Kafka將日志文件封裝成一個FileMessageSet對象，將偏移量索引文件和消息時間戳索引文件分別封裝成OffsetIndex和TimerIndex對象。Log和LogSegment均為邏輯概念，Log是對副本在Broker上存儲文件的抽象，而LogSegment是對副本存儲下每個日志分段的抽象，日志與索引文件才與磁盤上的物理存儲相對應(yīng)；下圖為Kafka日志存儲結(jié)構(gòu)中的對象之間的對應(yīng)關(guān)系圖：

為了進一步查看“.index”偏移量索引文件、“.timeindex”時間戳索引文件和“.log”日志數(shù)據(jù)文件，可以執(zhí)行下面的命令將二進制分段的索引和日志數(shù)據(jù)文件內(nèi)容轉(zhuǎn)換為字符型文件：

# 1、執(zhí)行下面命令即可將日志數(shù)據(jù)文件內(nèi)容dump出來
./kafka-run-class.sh kafka.tools.DumpLogSegments --files /apps/svr/Kafka/kafkalogs/kafka-topic-01-0/00000000000022372103.log --print-data-log > 00000000000022372103_txt.log

#2、dump出來的具體日志數(shù)據(jù)內(nèi)容
Dumping /apps/svr/Kafka/kafkalogs/kafka-topic-01-0/00000000000022372103.log
Starting offset: 22372103
offset: 22372103 position: 0 CreateTime: 1532433067157 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 5d2697c5-d04a-4018-941d-881ac72ed9fd
offset: 22372104 position: 0 CreateTime: 1532433067159 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 0ecaae7d-aba5-4dd5-90df-597c8b426b47
offset: 22372105 position: 0 CreateTime: 1532433067159 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 87709dd9-596b-4cf4-80fa-d1609d1f2087
......
......
offset: 22372444 position: 16365 CreateTime: 1532433067166 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 8d52ec65-88cf-4afd-adf1-e940ed9a8ff9
offset: 22372445 position: 16365 CreateTime: 1532433067168 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 5f5f6646-d0f5-4ad1-a257-4e3c38c74a92
offset: 22372446 position: 16365 CreateTime: 1532433067168 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 51dd1da4-053e-4507-9ef8-68ef09d18cca
offset: 22372447 position: 16365 CreateTime: 1532433067168 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 80d50a8e-0098-4748-8171-fd22d6af3c9b
......
......
offset: 22372785 position: 32730 CreateTime: 1532433067174 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: db80eb79-8250-42e2-ad26-1b6cfccb5c00
offset: 22372786 position: 32730 CreateTime: 1532433067176 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 51d95ab0-ab0d-4530-b1d1-05eeb9a6ff00
......
......
#3、同樣地，dump出來的具體偏移量索引內(nèi)容
Dumping /apps/svr/Kafka/kafkalogs/kafka-topic-01-0/00000000000022372103.index
offset: 22372444 position: 16365
offset: 22372785 position: 32730
offset: 22373467 position: 65460
offset: 22373808 position: 81825
offset: 22374149 position: 98190
offset: 22374490 position: 114555
......
......
#4、dump出來的時間戳索引文件內(nèi)容
Dumping /apps/svr/Kafka/kafkalogs/kafka-topic-01-0/00000000000022372103.timeindex
timestamp: 1532433067174 offset: 22372784
timestamp: 1532433067191 offset: 22373466
timestamp: 1532433067206 offset: 22373807
timestamp: 1532433067214 offset: 22374148
timestamp: 1532433067222 offset: 22374489
timestamp: 1532433067230 offset: 22374830
......
......

由上面dump出來的偏移量索引文件和日志數(shù)據(jù)文件的具體內(nèi)容可以分析出來，偏移量索引文件中存儲著大量的索引元數(shù)據(jù)，日志數(shù)據(jù)文件中存儲著大量消息結(jié)構(gòu)中的各個字段內(nèi)容和消息體本身的值。索引文件中的元數(shù)據(jù)postion字段指向?qū)?yīng)日志數(shù)據(jù)文件中message的實際位置（即為物理偏移地址）。
下面的表格先列舉了Kakfa消息體結(jié)構(gòu)中幾個主要字段的說明：

總結(jié)

從全文來看，Kafka高效數(shù)據(jù)存儲設(shè)計的特點在于以下幾點：
（1）、Kafka把主題中一個分區(qū)劃分成多個分段的小文件段，通過多個小文件段，就容易根據(jù)偏移量查找消息、定期清除和刪除已經(jīng)消費完成的數(shù)據(jù)文件，減少磁盤容量的占用；
（2）、采用稀疏索引存儲的方式構(gòu)建日志的偏移量索引文件，并將其映射至內(nèi)存中，提高查找消息的效率，同時減少磁盤IO操作；
（3）、Kafka將消息追加的操作邏輯變成為日志數(shù)據(jù)文件的順序?qū)懭?，極大的提高了磁盤IO的性能；
任何一位使用Kafka的同學(xué)來說，如果能夠掌握其數(shù)據(jù)存儲機制，對于大規(guī)模Kafka集群的性能調(diào)優(yōu)和問題定位都大有裨益。