Pulsar的消息存儲(chǔ)機(jī)制和Bookie的GC機(jī)制原理是什么

Pulsar消息存儲(chǔ)

Pulsar的消息存儲(chǔ)在BookKeeper中，BookKeeper是一個(gè)胖客戶的系統(tǒng)，客戶端部分稱為BookKeeper，服務(wù)器端集群中的每個(gè)存儲(chǔ)節(jié)點(diǎn)稱為bookie。Pulsar系統(tǒng)的broker作為BookKeeper存儲(chǔ)系統(tǒng)的客戶端，通過BookKeeper提供的客戶端SDK將Pulsar的消息存儲(chǔ)到bookies集群中。

Pulsar中的每個(gè)topic的每個(gè)分區(qū)（非分區(qū)topic，可以按照分區(qū)0理解，分區(qū)topic的編號(hào)是從0開始的），會(huì)對(duì)應(yīng)一系列的ledger，而每個(gè)ledger只會(huì)存儲(chǔ)對(duì)應(yīng)分區(qū)下的消息。對(duì)于每個(gè)分區(qū)同時(shí)只會(huì)有一個(gè)ledger處于open即可寫狀態(tài)。

Pulsar在生產(chǎn)消息，存儲(chǔ)消息時(shí)，會(huì)先找到當(dāng)前分區(qū)使用的ledger ，然后生成當(dāng)前消息對(duì)應(yīng)的entry ID，entry ID在同一個(gè)ledger內(nèi)是遞增的。非批量生產(chǎn)的情況（producer 端可以配置這個(gè)參數(shù)，默認(rèn)是批量的），一個(gè)entry 中包含一條消息。批量方式下，一個(gè)entry可能包含多條消息。而bookie中只會(huì)按照entry維度進(jìn)行寫入、查找、獲取。

因此，每個(gè)Pulsar下的消息的msgID 需要有四部分組成（老版本由三部分組成），分別為（ledgerID，entryID，partition-index，batch-index），其中，partition-index 在非分區(qū)topic的時(shí)候?yàn)?1，batch-index在非批量消息的時(shí)候?yàn)?1。

每個(gè)ledger，當(dāng)存在的時(shí)長(zhǎng)或保存的entry個(gè)數(shù)超過閾值后會(huì)進(jìn)行切換，同一個(gè)partition下的，新的消息會(huì)存儲(chǔ)到下一個(gè)ledger中。Ledger只是一個(gè)邏輯概念，是數(shù)據(jù)的一種邏輯組裝維度，并沒有對(duì)應(yīng)的實(shí)體。

BookKeeper集群中的每個(gè)bookie 節(jié)點(diǎn)收到消息后，數(shù)據(jù)會(huì)分三部分進(jìn)行存儲(chǔ)處理，分別為：journal 文件、entryLog 文件、索引文件。

其中journal文件，entry數(shù)據(jù)是按照wal方式寫入的到j(luò)ournal文件中，每個(gè)journal文件有大小限制，當(dāng)超過單個(gè)文件大小限制的時(shí)候會(huì)切換到下一個(gè)文件繼續(xù)寫，因?yàn)閖ournal文件是實(shí)時(shí)刷盤的，所以為了提高性能,避免相互之間的讀寫IO相互影響，建議存儲(chǔ)目錄與存儲(chǔ)entrylog的目錄區(qū)分開，并且給每個(gè)journal文件的存儲(chǔ)目錄單獨(dú)掛載一塊硬盤（建議使用ssd硬盤）。journal文件只會(huì)保存保存幾個(gè)，超過配置個(gè)數(shù)的文件將會(huì)被刪除。entry 存儲(chǔ)到j(luò)ournal文件完全是隨機(jī)的，先到先寫入，journal文件是為了保證消息不丟失而設(shè)計(jì)的。

如下圖所示，每個(gè)bookie收到增加entry的請(qǐng)求后，會(huì)根據(jù)ledger id映射到存儲(chǔ)到那個(gè)journal目錄和entry log目錄，entry數(shù)據(jù)會(huì)存儲(chǔ)在對(duì)應(yīng)的目錄下。目前bookie不支持在運(yùn)行過程中變更存儲(chǔ)目錄（使用過程中，增加或減少目錄會(huì)導(dǎo)致部分的數(shù)據(jù)查找不到）。

如下圖所示，bookie收到entry寫入請(qǐng)求后，寫入journal文件的同時(shí)，也會(huì)保存到write cache中，write cache分為兩部分，一部分是正在寫入的write cache， 一部分是正在正在刷盤的部分，兩部分交替使用。

write cache中有索引數(shù)據(jù)結(jié)構(gòu)，可以通過索引查找到對(duì)應(yīng)的entry，write cache中的索引是內(nèi)存級(jí)別的，基于bookie自己定義的ConcurrentLongLongPairHashMap結(jié)構(gòu)實(shí)現(xiàn)。

另外，每個(gè)entorylog的存儲(chǔ)目錄，會(huì)對(duì)應(yīng)一個(gè)SingleDirectoryDbLedgerStorage類實(shí)例對(duì)象，而每個(gè)SingleDirectoryDbLedgerStorage對(duì)象里面會(huì)有一個(gè)基于RockDB實(shí)現(xiàn)的索引結(jié)構(gòu)，通過這個(gè)索引可以快速的查到每個(gè)entry存儲(chǔ)在哪個(gè)entrylog文件中。每個(gè)write cache在增加entry的時(shí)候會(huì)進(jìn)行排序處理，在同一個(gè)write cache，同一個(gè)ledger下的數(shù)據(jù)是相鄰有序的，這樣在write cache中的數(shù)據(jù)flush到entrylog文件時(shí)，使得寫入到entrylog文件中的數(shù)據(jù)是局部有序的，這樣的設(shè)計(jì)能夠極大的提高后續(xù)的讀取效率。

SingleDirectoryDbLedgerStorage中的索引數(shù)據(jù)也會(huì)隨著entry的刷盤而刷盤到索引文件中。在bookie宕機(jī)重啟時(shí)，可以通過journal文件和entry log文件還原數(shù)據(jù)，保證數(shù)據(jù)不丟失。

Pulsar consumer 在消費(fèi)數(shù)據(jù)的時(shí)候，做了多層的緩存加速處理，如下圖所示：

獲取數(shù)據(jù)的順序如下：

• 在broker端的entry cache中獲取，如果沒有在繼續(xù)；
• 在bookie的write cache正在寫的這部分中獲取，如果沒有則繼續(xù)；
• 在bookie的write cache正在刷盤的這部分中獲取，如果沒有則繼續(xù)；
• 從bookie的read cache中獲取，如果沒有則繼續(xù)；
• 通過索引讀取磁盤上的entry log文件。

上面每一步，如果能獲取到數(shù)據(jù)，都會(huì)直接返回，跳過后面的步驟。如果是從磁盤文件中獲取的數(shù)據(jù)，會(huì)在返回的時(shí)候?qū)?shù)據(jù)存儲(chǔ)到read cache中，另外如果是讀取磁盤的操作，會(huì)多讀取一部分磁盤上的時(shí)候，因?yàn)榇鎯?chǔ)的時(shí)候有局部有序的處理，獲取相鄰數(shù)據(jù)的概率非常大，這種處理的話會(huì)極大的提高后續(xù)獲取數(shù)據(jù)的效率。

我們?cè)谑褂玫倪^程中，應(yīng)盡量避免或減少出現(xiàn)消費(fèi)過老數(shù)據(jù)即觸發(fā)讀取磁盤文件中的消息的場(chǎng)景，以免對(duì)整體系統(tǒng)的性能造成影響。

BookKeeper的GC機(jī)制

BookKeeper中的每個(gè)bookie都會(huì)周期的進(jìn)行數(shù)據(jù)清理操作，默認(rèn)15分鐘檢查處理一次，清理的主要流程如下

1. 清理bookie存儲(chǔ)的ledger id（bookie內(nèi)存儲(chǔ)的ledger id與zk上面存儲(chǔ)的 ledger id做比較，如果zk上面沒有則刪除bookie中存儲(chǔ)的ledger id）；
2. 統(tǒng)計(jì)每個(gè)entry log中存活的entry占比，當(dāng)前entry log 存活的ledger個(gè)數(shù)為0時(shí)刪除這個(gè)entry log；
3. 根據(jù)entry log的元數(shù)據(jù)信息，清理entry log 文件（當(dāng)entry log包含的所有l(wèi)edger id全部失效時(shí)刪除）；

4. 壓縮entry log文件 ,分別在當(dāng)前entry log文件下存活的entry比例在0.5-默認(rèn)周期1天(major gc) 或比例0.2-默認(rèn)周期1個(gè)小時(shí)(minor gc) 的時(shí)候，Compaction entry log文件，將老的文件中存活的entry轉(zhuǎn)移新的文件中，然后將老的entry log文件刪除，單次的GC如果處理的entry log文件比較大的時(shí)候可能耗時(shí)比較長(zhǎng)。

通過上面的流程，我們可以了解bookie在清理entrylog文件時(shí)的大體流程。

需要特別說明的是，ledger是否是可以刪除的，完全是客戶端的觸發(fā)的，在Pulsar中是broker觸發(fā)的。

broker端有周期的處理線程（默認(rèn)2分鐘），清理已經(jīng)消費(fèi)過的消息所在的ledger機(jī)制，獲取topic中包含的cursor最后確認(rèn)的消息，將這個(gè)topic包含的ledger列表中，在這個(gè)id之前的（注意不包含當(dāng)前的ledger id）全部刪除（包括zk中的元數(shù)據(jù)，同時(shí)通知bookie刪除對(duì)應(yīng)的ledger）。

運(yùn)營(yíng)中遇到的問題分析

在運(yùn)用的過程中我們多次遇到了bookie磁盤空間不足的場(chǎng)景，bookie中存儲(chǔ)了大量的entry log文件。比較典型的原因主要有如下兩個(gè)。

原因一：

生產(chǎn)消息過于分散，例如，舉個(gè)極端的場(chǎng)景，1w個(gè)topic，每個(gè)topic生產(chǎn)一條，1w個(gè)topic順序生產(chǎn)。這樣每個(gè)topic 對(duì)應(yīng)的ledger短時(shí)間內(nèi)不會(huì)因?yàn)闀r(shí)長(zhǎng)或者存儲(chǔ)大小進(jìn)行切換，active狀態(tài)的ledger id分散在大量的entry log文件中。這些entry log文件是不能刪除或者及時(shí)壓縮的。

如果遇到這種場(chǎng)景，可以通過重啟，強(qiáng)制ledger進(jìn)行切換進(jìn)行處理。當(dāng)然如果這個(gè)時(shí)候消費(fèi)進(jìn)行沒有跟上，消費(fèi)的last ack位置所在的ledger也是處于active狀態(tài)的，不能進(jìn)行刪除。

原因二：

GC時(shí)間過程，如果現(xiàn)存的enrylog文件比較多，且大量符合minor或major gc閾值，這樣，單次的minor gc或者major gc時(shí)間過長(zhǎng)，在這段時(shí)間內(nèi)是不能清理過期的entry log文件。

這是由于單次清理流程的順序執(zhí)行導(dǎo)致的，只有上次一輪執(zhí)行完，才會(huì)執(zhí)行下一次。目前，這塊也在提優(yōu)化流程，避免子流程執(zhí)行實(shí)現(xiàn)過長(zhǎng)，對(duì)整體產(chǎn)生影響。