Pulsar消息存儲

Pulsar的消息存儲在BookKeeper中，BookKeeper是一個胖客戶的系統，客戶端部分稱為BookKeeper，服務器端集群中的每個存儲節點稱為bookie。Pulsar系統的broker作為BookKeeper存儲系統的客戶端，通過BookKeeper提供的客戶端SDK將Pulsar的消息存儲到bookies集群中。

Pulsar中的每個topic的每個分區（非分區topic，可以按照分區0理解，分區topic的編號是從0開始的），會對應一系列的ledger，而每個ledger只會存儲對應分區下的消息。對于每個分區同時只會有一個ledger處于open即可寫狀態。

Pulsar在生產消息，存儲消息時，會先找到當前分區使用的ledger ，然后生成當前消息對應的entry ID，entry ID在同一個ledger內是遞增的。非批量生產的情況（producer 端可以配置這個參數，默認是批量的），一個entry 中包含一條消息。批量方式下，一個entry可能包含多條消息。而bookie中只會按照entry維度進行寫入、查找、獲取。

因此，每個Pulsar下的消息的msgID 需要有四部分組成（老版本由三部分組成），分別為（ledgerID，entryID，partition-index，batch-index），其中，partition-index 在非分區topic的時候為-1，batch-index在非批量消息的時候為-1。

每個ledger，當存在的時長或保存的entry個數超過閾值后會進行切換，同一個partition下的，新的消息會存儲到下一個ledger中。Ledger只是一個邏輯概念，是數據的一種邏輯組裝維度，并沒有對應的實體。

BookKeeper集群中的每個bookie 節點收到消息后，數據會分三部分進行存儲處理，分別為：journal 文件、entryLog 文件、索引文件。

其中journal文件，entry數據是按照wal方式寫入的到journal文件中，每個journal文件有大小限制，當超過單個文件大小限制的時候會切換到下一個文件繼續寫，因為journal文件是實時刷盤的，所以為了提高性能,避免相互之間的讀寫IO相互影響，建議存儲目錄與存儲entrylog的目錄區分開，并且給每個journal文件的存儲目錄單獨掛載一塊硬盤（建議使用ssd硬盤）。journal文件只會保存保存幾個，超過配置個數的文件將會被刪除。entry 存儲到journal文件完全是隨機的，先到先寫入，journal文件是為了保證消息不丟失而設計的。

如下圖所示，每個bookie收到增加entry的請求后，會根據ledger id映射到存儲到那個journal目錄和entry log目錄，entry數據會存儲在對應的目錄下。目前bookie不支持在運行過程中變更存儲目錄（使用過程中，增加或減少目錄會導致部分的數據查找不到）。

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

write cache中有索引數據結構，可以通過索引查找到對應的entry，write cache中的索引是內存級別的，基于bookie自己定義的ConcurrentLongLongPairHashMap結構實現。

另外，每個entorylog的存儲目錄，會對應一個SingleDirectoryDbLedgerStorage類實例對象，而每個SingleDirectoryDbLedgerStorage對象里面會有一個基于RockDB實現的索引結構，通過這個索引可以快速的查到每個entry存儲在哪個entrylog文件中。每個write cache在增加entry的時候會進行排序處理，在同一個write cache，同一個ledger下的數據是相鄰有序的，這樣在write cache中的數據flush到entrylog文件時，使得寫入到entrylog文件中的數據是局部有序的，這樣的設計能夠極大的提高后續的讀取效率。

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

Pulsar consumer 在消費數據的時候，做了多層的緩存加速處理，如下圖所示：

獲取數據的順序如下：

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

上面每一步，如果能獲取到數據，都會直接返回，跳過后面的步驟。如果是從磁盤文件中獲取的數據，會在返回的時候將數據存儲到read cache中，另外如果是讀取磁盤的操作，會多讀取一部分磁盤上的時候，因為存儲的時候有局部有序的處理，獲取相鄰數據的概率非常大，這種處理的話會極大的提高后續獲取數據的效率。

我們在使用的過程中，應盡量避免或減少出現消費過老數據即觸發讀取磁盤文件中的消息的場景，以免對整體系統的性能造成影響。

BookKeeper的GC機制

BookKeeper中的每個bookie都會周期的進行數據清理操作，默認15分鐘檢查處理一次，清理的主要流程如下

1. 清理bookie存儲的ledger id（bookie內存儲的ledger id與zk上面存儲的 ledger id做比較，如果zk上面沒有則刪除bookie中存儲的ledger id）；
2. 統計每個entry log中存活的entry占比，當前entry log 存活的ledger個數為0時刪除這個entry log；
3. 根據entry log的元數據信息，清理entry log 文件（當entry log包含的所有ledger id全部失效時刪除）；

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

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

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

broker端有周期的處理線程（默認2分鐘），清理已經消費過的消息所在的ledger機制，獲取topic中包含的cursor最后確認的消息，將這個topic包含的ledger列表中，在這個id之前的（注意不包含當前的ledger id）全部刪除（包括zk中的元數據，同時通知bookie刪除對應的ledger）。

運營中遇到的問題分析

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

原因一：

生產消息過于分散，例如，舉個極端的場景，1w個topic，每個topic生產一條，1w個topic順序生產。這樣每個topic 對應的ledger短時間內不會因為時長或者存儲大小進行切換，active狀態的ledger id分散在大量的entry log文件中。這些entry log文件是不能刪除或者及時壓縮的。

如果遇到這種場景，可以通過重啟，強制ledger進行切換進行處理。當然如果這個時候消費進行沒有跟上，消費的last ack位置所在的ledger也是處于active狀態的，不能進行刪除。

原因二：

GC時間過程，如果現存的enrylog文件比較多，且大量符合minor或major gc閾值，這樣，單次的minor gc或者major gc時間過長，在這段時間內是不能清理過期的entry log文件。

這是由于單次清理流程的順序執行導致的，只有上次一輪執行完，才會執行下一次。目前，這塊也在提優化流程，避免子流程執行實現過長，對整體產生影響。