一掏熬、Kafka

存儲(chǔ)結(jié)構(gòu).png

Kafka 日志對(duì)象由多個(gè)日志段對(duì)象組成捻勉，而每個(gè)日志段對(duì)象會(huì)在磁盤上創(chuàng)建一組文件骚勘，包括消息日志文件（.log）、位移索引文件（.index）疑故、時(shí)間戳索引文件（.timeindex）以及已中止（Aborted）事務(wù)的索引文件（.txnindex）(如果沒有使用 Kafka 事務(wù)杠览，已中止事務(wù)的索引文件是不會(huì)被創(chuàng)建出來的)。圖中的一串?dāng)?shù)字 0 是該日志段的起始位移值纵势，也就是該日志段中所存的第一條消息的位移值踱阿。

一般情況下，一個(gè) Kafka 主題有很多分區(qū)钦铁，每個(gè)分區(qū)就對(duì)應(yīng)一個(gè) Log 對(duì)象软舌，在物理磁盤上則對(duì)應(yīng)于一個(gè)子目錄。比如你創(chuàng)建了一個(gè)雙分區(qū)的主題 test-topic育瓜，那么葫隙，Kafka 在磁盤上會(huì)創(chuàng)建兩個(gè)子目錄：test-topic-0 和 test-topic-1栽烂。而在服務(wù)器端(Broker)躏仇，這就是兩個(gè) Log 對(duì)象。每個(gè)子目錄下存在多組日志段腺办，也就是多組.log焰手、.index、.timeindex 文件組合怀喉，只不過文件名不同书妻，因?yàn)槊總€(gè)日志段的起始位移不同。

文件內(nèi)容

# 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出來的時(shí)間戳索引文件內(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
......
......

1. 消息日志文件(.log)

消息日志文件就是存儲(chǔ)具體的消息內(nèi)容，默認(rèn)1G菱蔬；其主要內(nèi)容為：

• offset：唯一確定了同一分區(qū)中(不是同一消息日志文件中篷帅，也不是同一Broker上)一條Message的邏輯位置，同一個(gè)分區(qū)下的消息偏移量按照順序遞增
• position：表示該條Message在磁盤上消息日志文件中的絕對(duì)位置拴泌。只要打開文件并移動(dòng)文件指針到這個(gè)position就可以讀取對(duì)應(yīng)的Message魏身；
• key：消息Key實(shí)際數(shù)據(jù)
• payload：消息內(nèi)容實(shí)際數(shù)據(jù)

2. 偏移量索引文件(.index)

偏移量索引文件中存儲(chǔ)的 offset -> position 的映射關(guān)系；

• 為了減少存儲(chǔ)空間的使用蚪腐，Kafka采用稀疏索引存儲(chǔ)的方式箭昵，每隔一定的字節(jié)數(shù)建立了一條索引，可以通過“index.interval.bytes”設(shè)置索引的跨度回季；
• 根據(jù)指定的偏移量家制，
  首先使用二分法找到該消息所在的.index文件和.log文件掉房；
  然后在.index文件中再通過二分查找法，繼續(xù)查找出小于等于指定偏移量的最大偏移量慰丛，同時(shí)也得出了對(duì)應(yīng)的position卓囚；
  最后根據(jù)該position在.log文件中順序掃描查找偏移量與指定偏移量相等的消息；

為了解決傳統(tǒng)二分查找算法導(dǎo)致的不必要的Page Fault的問題诅病，Kafka進(jìn)行了改進(jìn)哪亿，把所有索引項(xiàng)分成兩部分：熱區(qū)和冷區(qū)，然后分別在這兩個(gè)區(qū)域內(nèi)執(zhí)行二分查找算法贤笆；

3. 時(shí)間戳索引文件(.timeindex)

時(shí)間戳索引文件存儲(chǔ) 時(shí)間戳 -> offset映射關(guān)系蝇棉；

Kafka零拷貝機(jī)制：
1.Kafka索引都是基于 MappedByteBuffer 的，也就是讓用戶態(tài)和內(nèi)核態(tài)共享內(nèi)核態(tài)的數(shù)據(jù)緩沖區(qū)芥永，此時(shí)篡殷，數(shù)據(jù)不需要復(fù)制到用戶態(tài)空間。不過埋涧，mmap 雖然避免了不必要的拷貝板辽，但不一定就能保證很高的性能。在不同的操作系統(tǒng)下棘催，mmap 的創(chuàng)建和銷毀成本可能是不一樣的劲弦。很高的創(chuàng)建和銷毀開銷會(huì)抵消 Zero Copy 帶來的性能優(yōu)勢(shì)。由于這種不確定性醇坝，在 Kafka 中邑跪，只有索引應(yīng)用了 mmap，最核心的日志并未使用 mmap 機(jī)制呼猪。
2.TransportLayer 是 Kafka 傳輸層的接口画畅。它的某個(gè)實(shí)現(xiàn)類使用了
FileChannel 的 transferTo 方法。該方法底層使用 sendfile 實(shí)現(xiàn)了 Zero Copy宋距。對(duì) Kafka而言轴踱，如果 I/O 通道使用普通的 PLAINTEXT，那么乡革，Kafka 就可以利用 Zero Copy 特性寇僧，直接將頁緩存中的數(shù)據(jù)發(fā)送到網(wǎng)卡的 Buffer 中，避免中間的多次拷貝沸版。相反嘁傀，如果I/O 通道啟用了 SSL，那么视粮，Kafka 便無法利用 Zero Copy 特性了细办。

二、RocketMQ

1. 文件結(jié)構(gòu)

|-- store
    |-- commitlog
        |-- 00000003384434229248
        |-- 00000003385507971072
        |-- 00000003386581712896 
    |-- consumequeue
        |-- TopicA
            |-- 0
                |-- 00000000002604000000
                |-- 00000000002610000000
            |-- 1
                |-- 00000000002610000000
                |-- 00000000002616000000
        |-- TopicB
            |-- 0
                |-- 00000000000732000000
            |-- 1
                |-- 00000000004610000000
            |-- 3
                |-- 00000000005610000000
            |-- 4
                |-- 00000000006610000000
    |-- index
        |-- 20210620170423012

RocketMQ主要有三中日志文件：

• commitlog：存儲(chǔ)原始消息，單個(gè)文件默認(rèn)1G笑撞，寫滿后生成新文件岛啸；Broker上所有Topic共享commitlog文件；
• consumequeue：創(chuàng)建Topic時(shí)可以指定queue數(shù)量茴肥，queue平均分配(或者指定具體Broker)到Broker上坚踩，每個(gè)queue對(duì)應(yīng)一個(gè)consumequeue文件；
• index：存儲(chǔ)key -> 消息在CommitLog文件中的物理偏移量 的映射關(guān)系瓤狐，方便通過key或時(shí)間區(qū)間進(jìn)行查找瞬铸；Broker上所有Topic共享index文件；

2. 消息存儲(chǔ)

消息存儲(chǔ).png

RocketMQ的混合型存儲(chǔ)結(jié)構(gòu)采用了數(shù)據(jù)(commitlog)和索引(index/consumequeue)部分相分離的存儲(chǔ)結(jié)構(gòu)础锐，Producer發(fā)送消息至Broker端嗓节，然后Broker端使用同步或者異步的方式對(duì)消息刷盤持久化，保存至CommitLog中皆警。只要消息被刷盤持久化至磁盤文件CommitLog中拦宣，那么Producer發(fā)送的消息就不會(huì)丟失，因此Consumer也就肯定有機(jī)會(huì)去消費(fèi)這條消息信姓，至于消費(fèi)的時(shí)間可以稍微滯后一些也沒有太大的關(guān)系鸵隧。
這里，RocketMQ的具體做法是财破，使用Broker端的后臺(tái)服務(wù)線程ReputMessageService不停地分發(fā)請(qǐng)求并異步構(gòu)建ConsumeQueue和IndexFile掰派。

2.1 commitlog

commitlog.png

RocketMQ的commitLog是WAL的一種从诲，通過對(duì)文件的順序追加寫入左痢，提高了文件的寫入性能；

2.2 consumequeue

ConsumeQueue.png

• consumequeue文件中主要是offset系洛，即在commitLog中的位置信息俊性，通過這個(gè)offset偏移量保證消息讀取階段能夠定位到消息的物理位置；
• 單條數(shù)據(jù)大小為20字節(jié)描扯，單個(gè)ConsumeQueue文件能夠保存30萬條數(shù)據(jù)定页，每個(gè)文件大約占用5.7MB；
• Topic下每個(gè)MessageQueue對(duì)應(yīng)了Broker上多個(gè)ConsumeQueue文件绽诚，這些ConsumeQueue文件保存了該MessageQueue的所有消息在CommitLog文件中的物理位置典徊，即offset偏移量；
• ConsumeQueue能夠區(qū)分不同Topic下的不同MessageQueue的消息恩够，同時(shí)能夠?yàn)橄M(fèi)消息起到一定的緩沖作用（當(dāng)只有ReputMessageService異步服務(wù)線程通過doDispatch異步生成了ConsumeQueue隊(duì)列的元素后卒落，Consumer端才能進(jìn)行消費(fèi)）；

2.3 index

IndexFile.png

為了滿足根據(jù)msgId以及消息key查詢消息的需求蜂桶，每個(gè)Broker對(duì)應(yīng)一組indexFile儡毕；

• index文件分為三部分：文件頭indexHeader，一系列槽位slots扑媚，真正的索引數(shù)據(jù)index腰湾；
• index文件最大大小為40+50000004+5000000 * 420byte雷恃，寫完后繼續(xù)寫下一個(gè)，indexFile文件名比較特殊费坊，是一串時(shí)間戳倒槐；
• indexFile結(jié)構(gòu)與hash表很相似，固定數(shù)量的slot組成數(shù)組附井，每個(gè)slot對(duì)應(yīng)一條index鏈导犹，index之間通過鏈表方式組織在一起。slot的值對(duì)應(yīng)當(dāng)前slot下最新的那個(gè)index的序號(hào)羡忘，index中存儲(chǔ)了當(dāng)前slot下谎痢、當(dāng)前index的前一個(gè)index序號(hào)，這就把slot下的所有index鏈起來了卷雕；·
  由于indexHeader节猿，slot，index都是固定大小漫雕，所以：
  公式1：第n個(gè)slot在indexFile中的起始位置是這樣:40+(n-1)4
  公式2： 第s個(gè)index在indexFile中的起始位置是這樣:40+50000004+(s-1)*20
• 查詢流程：
  key-->計(jì)算hash值-->hash值對(duì)500萬取余算出對(duì)應(yīng)的slot序號(hào)-->根據(jù)40+(n-1)4(公式1)算出該slot在文件中的位置-->讀取slot值滨嘱，也就是index序號(hào)-->根據(jù)40+50000004+(s-1)*20(公式2)算出該index在文件中的位置-->讀取該index-->將key的hash值與index的keyHash值進(jìn)行比對(duì)；
  -->不滿足則根據(jù)index中的preIndexNo找到上一個(gè)index浸间，繼續(xù)上一步
  -->滿足則根據(jù)index中的phyOffset拿到commitLog中的消息

三太雨、對(duì)比

• 1. Kafka針對(duì)Producer和Consumer使用了同一份存儲(chǔ)結(jié)構(gòu)；
     RocketMQ卻為Producer和Consumer分別設(shè)計(jì)了不同的存儲(chǔ)結(jié)構(gòu)魁蒜，Producer對(duì)應(yīng)CommitLog, Consumer對(duì)應(yīng)ConsumeQueue囊扳，CommitLog和ConsumeQueue采用“最終一致性”的方案保證一致性；
• 2. Kafka每個(gè)partition對(duì)應(yīng)一個(gè)日志文件兜看，Producer對(duì)該日志文件進(jìn)行“順序?qū)憽弊断蹋珻onsumer對(duì)該文件進(jìn)行“順序讀”，這種存儲(chǔ)方式细移，對(duì)于每個(gè)文件來說是順序IO搏予，但是當(dāng)并發(fā)的讀寫多個(gè)partition的時(shí)候，對(duì)應(yīng)多個(gè)文件的順序IO弧轧，表現(xiàn)在文件系統(tǒng)的磁盤層面雪侥，還是隨機(jī)IO精绎。因此當(dāng)partition或者topic個(gè)數(shù)過多時(shí)速缨，Kafka的性能急劇下降；
     RocketMQ采用了單一的日志文件捺典，即把同一臺(tái)機(jī)器上面所有topic的所有queue的消息鸟廓，存放在一個(gè)文件里面，從而避免了隨機(jī)的磁盤寫入；
• 3. Kafka：消息的讀寫都是基于 FileChannel引谜；索引讀寫基于 MMAP牍陌；
     RocketMQ：讀盤基于 MMAP，寫盤默認(rèn)使用 MMAP员咽，可通過修改配置毒涧，配置成 FileChannel，原因是可以避免 PageCache 的鎖競(jìng)爭(zhēng)贝室，通過兩層架構(gòu)實(shí)現(xiàn)讀寫分離契讲；

• 參考《消息中間件—Kafka數(shù)據(jù)存儲(chǔ)（一）》