Pulsar工作原理

參考文檔：

Pulsar：

https://jack-vanlightly.com/blog/2018/10/2/understanding-how-apache-pulsar-works

https://jack-vanlightly.com/blog/2018/10/21/how-to-not-lose-messages-on-an-apache-pulsar-cluster

https://www.splunk.com/en_us/blog/it/effectively-once-semantics-in-apache-pulsar.html

BookKeeper：

https://medium.com/splunk-maas/apache-bookkeeper-insights-part-1-external-consensus-and-dynamic-membership-c259f388da21

https://medium.com/splunk-maas/apache-bookkeeper-internals-part-1-high-level-6dce62269125

https://medium.com/splunk-maas/apache-bookkeeper-internals-part-2-writes-359ffc17c497

https://medium.com/splunk-maas/apache-bookkeeper-internals-part-3-reads-31637b118bf

https://medium.com/splunk-maas/apache-bookkeeper-internals-part-4-back-pressure-7847bd6d1257

https://medium.com/splunk-maas/a-guide-to-the-bookkeeper-replication-protocol-tla-series-part-2-29f3371fe395

中文（翻譯的不太行，建議看英文）：

https://blog.51cto.com/u_13491808/3124971

https://juejin.cn/post/7034135607869702157

https://juejin.cn/post/6948044887941971998

https://blog.csdn.net/zhaijia03/article/details/112691063

https://xie.infoq.cn/article/5dc2c3a2afa62d1eec3061fec

https://xie.infoq.cn/article/35b764ea84155a0f41542df6f

本文暫不關(guān)心Pulsar的消息消費（如批量消費）造虎，Topic分區(qū)傅蹂，和消息存儲時間（TTL和Retention）等的具體細節(jié)，而主要探討和總結(jié)Pulsar的工作原理算凿，部分地方會和Kafka進行對比份蝴。

Pulsar的總體架構(gòu)

Pulsar的總體架構(gòu)如下，

20220112-Pulsar 工作原理-2022-01-17-10-10-24.png

總共有3個部分澎媒，Broker搞乏、BookKeeper、Zookeeper戒努。通常，我們講Pulsar的話，主要都是指Broker储玫，另外兩個模塊也是獨立的Apache項目侍筛。

Broker本身沒有存儲，因此也不會丟失數(shù)據(jù)撒穷，Broker運行需要的數(shù)據(jù)都來自BookKeeper和Zookeeper匣椰。Broker對client提供服務，處理client的請求端礼。

BookKeeper主要存儲消息數(shù)據(jù)禽笑，因而也是需要存儲最大的地方，單個BookKeeper也叫Bookie蛤奥。BookKeeper本身是有WAL的消息存儲引擎佳镜。

Zookeeper主要存儲元數(shù)據(jù)，包括Broker和BookKeeper的元數(shù)據(jù)凡桥。

Pulsar提供的消息讀寫模式與Kafka比較類似蟀伸，都是按Topic來關(guān)系消息，都有消息順序性保證缅刽，都有Offset機制啊掏，在Pulsar中叫Cursor。

Client發(fā)起請求衰猛，包括讀/寫請求迟蜜，會先發(fā)送到Broker，Broker來判定啡省，當前請求消息的Cursor在哪個Bookie上小泉，然后訪問對應的Bookie讀取/寫入消息并返回。這個只是最簡單的流程冕杠，Pulsar中有Cache機制微姊，實際的流程會比這個更復雜一些。
出于消息可靠性的考慮分预，Pulsar會將消息存儲多份兢交，也就是說，相同的消息會存在多個Bookie上笼痹。

數(shù)據(jù)存儲模型

關(guān)于消息數(shù)據(jù)的存儲配喳，首先要理解下面這張圖，

20220112-Pulsar 工作原理-2022-01-17-10-10-54.png

上圖中凳干，從上到下晴裹，每個層次的名稱是：Topic（主題），Ledger（賬本）救赐，F(xiàn)ragment（片段）涧团，Entry（條目）。逐一解釋一下，

1. Entry是存儲的最小單位泌绣，在Pulsar中钮追，Entry可以是一條消息或一組消息；
2. Fragment是多個Entry的組合阿迈，是BookKeeper上最小的分布單位元媚，以Fragment為單位在多個Bookie上做數(shù)據(jù)冗余復制；
3. Ledger包含1個或多個Fragment苗沧，是BookKeeper的管理單元刊棕，進行可用性管理，Ledger一旦關(guān)閉就是不可變的（immutable）待逞，以Ledger為單位進行刪除甥角，無法刪除單個Entry；
4. Topic包括多個Ledger飒焦，是最上層的邏輯概念蜈膨，消費者可以對Topic進行訂閱；

其中牺荠，Ledger翁巍、Fragment、Entry是BookKeeper中的概念休雌，Topic是Pulsar中的概念灶壶。在Pulsar官方文檔中還出現(xiàn)了Managed Ledger概念，沒有特別看出和Ledger的差別杈曲，應該是和BookKeeper的Ledger等價的Pulsar概念驰凛。

在實際的物理結(jié)構(gòu)上，數(shù)據(jù)存儲的分布如下圖担扑，

20220112-Pulsar 工作原理-2022-01-17-10-11-20.png

可以看到恰响，一個Topic對應多個Ledger，一個Ledger有1個或多個Fragment涌献，不同的Fragment分布在不同的Bookie上胚宦，存儲了多份。Ledger燕垃、Fragment如何切分枢劝，如何分布在Bookie的元數(shù)據(jù)，統(tǒng)一存儲在Zookeeper卜壕。

存儲結(jié)構(gòu)與Kafka的最大差別是：一個Topic包含多個Ledger您旁，Kafka是1個Ledger；以Fragment為單位進行分布式存儲轴捎，Kafka是以Ledger為單位分布式存儲鹤盒〔显啵可以很明顯的感覺到，Pulsar的存儲分的很細昨悼，而且做到了物理存儲結(jié)構(gòu)與邏輯結(jié)構(gòu)相隔離蝗锥，最終達到跃洛，只要擴展Bookie集群就能提升整體可用性和性能率触。

什么時候切分Ledger和Fragment？
切分Ledger的時機有以下幾個地方：

1. 新建Topic汇竭；
2. 當前Ledger達到大小上限葱蝗，或時間上限；
3. Topic的Broker所有權(quán)發(fā)生變化细燎；

切分Fragment的時機有以下幾個地方：

1. 新建Ledger两曼；
2. 寫入Bookie失敗玻驻；
   也就是說悼凑，如果Bookie沒有發(fā)生停機的情況下，Ledger和Fragment會是一對一的璧瞬。

存儲的高可用

存儲的配置是以Ledger為單位來管理的户辫，最重要的配置有三個，

• Ensemble size (E)嗤锉，全體數(shù)量
• Write quorum size (Qw)渔欢，寫入數(shù)量
• Ack quorum size (Qa)，響應數(shù)量

全體數(shù)量瘟忱，E奥额，表示Ledger可以寫入的總體Bookie池的Bookie數(shù)量；寫入數(shù)量访诱，Qw垫挨，表示對于每個Entry，Ledger需要寫入的份數(shù)触菜；響應數(shù)量九榔，Qa，表示當寫入返回多少個Ack時玫氢，返回給客戶端帚屉，即寫入成功。通常情況下漾峡，E >= Qw >= Qa攻旦。

Qa和Qw

先來看Qa和Qw，舉例生逸，Qa=2牢屋，Qw=3且预。也就是說，對于每個寫入的Entry烙无，需要復制3份锋谐，也就是存儲在3個Bookie上；但只要已經(jīng)收到成功寫入2份的Ack截酷，就表示成功寫入涮拗，返回給客戶端。在這個配置下迂苛，如果宕機了1個Bookie三热，數(shù)據(jù)是完全可以恢復回來的，但是宕機2個Bookie的話三幻，數(shù)據(jù)就可能出現(xiàn)丟失就漾。如果想宕機2個Bookie數(shù)據(jù)仍然可以不丟失，那么至少需要配置Qa=3念搬。

也就是說抑堡，Qa是保證數(shù)據(jù)不丟失的最小數(shù)據(jù)復制份數(shù)，這個取決于應用場景朗徊，需要恢復何種宕機程度的數(shù)據(jù)首妖。這個概念和Kafka的in sync replica很相似。

Qw和E

再來看Qw和E荣倾，舉例悯搔，Qw=3，E=3舌仍。這個情況下妒貌，對于每個寫入的Entry，需要復制到當前Fragment每個Bookie上铸豁，如下圖灌曙，

20220112-Pulsar 工作原理-2022-01-17-10-12-46.png

可以看到，Entry按寫入的順序緊密排列节芥，如果是Qw=3在刺，E=5的情況下，可用Bookie的數(shù)量比寫入Bookie的數(shù)量要多头镊，寫入的Entry的排列會出現(xiàn)空洞蚣驼，如下圖，

20220112-Pulsar 工作原理-2022-01-17-10-13-02.png

這樣的現(xiàn)象相艇，Pulsar稱為Striping颖杏。這種情況下，寫入的tps會提高坛芽，但是讀取的性能會下降留储，最終增大整體的延遲翼抠。在這種情況下，BookKeeper的順序讀取被打破获讳，降低整體性能阴颖，因此不建議使用。

因此丐膝，通常情況下量愧，取E = Qw >= Qa，例如尤误，E=3侠畔，Qw=3结缚，Qa=2损晤。

同樣不建議取Qa=1。這是一個危險的設置红竭，如果唯一的Bookie宕機尤勋，那么就不知道Entry是否已寫入。Bookie的恢復會因無法進行而停止茵宪。

Brookie也可以配置機架感知（rack-awareness）最冰，當配置了機架感知策略時，Broker會嘗試選取不同機架的Bookie節(jié)點稀火。當然也可以自定義其他選取策略暖哨。

Broker和Topic所有權(quán)

Pulsar的Broker不存儲數(shù)據(jù)，因此也不會丟失凰狞。Jack Vanlightly的博客原文是這樣篇裁，

Pulsar brokers have no persistent state that cannot be lost.

這里并沒有無狀態(tài)的意思，很多中文翻譯博客把這里翻譯成赡若，Broker是無狀態(tài)的达布，甚至把這一句放在非常開頭的地方，但其實是不對的逾冬。Broker只是不存儲有狀態(tài)的數(shù)據(jù)而已黍聂，本身在內(nèi)存中是有狀態(tài)的。Broker和其他Broker并不對等身腻。

每個Topic都歸一個Broker所有产还，所有的讀寫都需要通過這個Broker進行。寫入過程如下圖嘀趟，

20220112-Pulsar 工作原理-2022-01-17-10-13-57.png

可以看到脐区，上圖例子中，Qw=3去件，Broker收到寫入請求的時候坡椒，先寫入Bookie扰路，Bookie完成寫入請求后返回Ack，Broker收到Qa個Ack后返回Ack給客戶端倔叼。如果Bookie返回失敗或者無返回汗唱，那么Broker會發(fā)起創(chuàng)建新Fragment。

讀取過程如下圖丈攒，

20220112-Pulsar 工作原理-2022-01-17-10-14-11.png

因為Topic所有的請求都需要通過所有者Broker哩罪，那么，我們可以在這個Broker上引入Cache機制巡验，提升讀的QPS际插。

這樣的緩存機制會對不同消費者有比較大的性能差異，如果是追尾消費者（tail reader）显设，即一直追蹤Entry最新變化的消費者框弛，當有Entry寫入時，會更新Cache捕捂，于是直接從緩存中取走Entry瑟枫；但如果是追趕消費者（catch-up reader），即讀取的是老的Entry指攒，例如消費者宕機后重啟慷妙，中間堆積了一段時間的消息的情況，此時允悦，緩存中沒有數(shù)據(jù)膝擂，必須去Bookie上讀取，再返回給客戶端隙弛。由于無法直接從緩存獲取Entry架馋，追趕消費者獲取消息的性能是要比追尾消費者差很多的。

Broker故障恢復

Broker因為是有狀態(tài)的驶鹉，無法做到非常完美的災備切換绩蜻，只能在故障后盡快恢復Broker的工作場景。

Broker故障恢復中有一個非常重要的概念室埋，最新確認序號（Last Added Confirmed 办绝，LAC）。這個表示當前Ledger最后commit的序號姚淆，也就是收到Qa個Ack的Entry的序號孕蝉。Pulsar約定，讀取數(shù)據(jù)不可以讀取LAC之后的數(shù)據(jù)腌逢，讀取LAC之后的數(shù)據(jù)是沒用一致性和正確性保障的降淮，視為臟讀。

理解LAC之后，就可以理解Broker故障恢復的柵欄阻擋機制（Fencing）了佳鳖。步驟如下霍殴，

1. 當前Broker B1，擁有Topic X系吩，被Zookeeper確認為不可用来庭；
2. 另一個Broker B2 將Topic X 的當前Ledger狀態(tài)從OPEN修改為IN_RECOVERY（應該是修改Zookeeper的狀態(tài)）；
3. B2發(fā)起柵欄阻擋LAC請求（Fencing LAC Request）給當前Fragment/Ledger的所有Bookie穿挨，并等待(Qw-Qa)+1個響應月弛。一旦收集齊，Ledger就被阻擋了科盛。即使B1仍然存活（如B1網(wǎng)絡斷線重連場景）帽衙，也無法寫入消息，因為無法獲得Qa個Ack贞绵，會返回Fencing異常厉萝；
4. B2得到最大的LAC，然后從LAC+1開始對Bookie上的數(shù)據(jù)進行恢復讀取但壮，確保從LAC+1開始冀泻，每個Entry都被復制到Qw個Bookie。這主要是由于一些Bookie的Ack之前沒有傳輸?shù)紹1蜡饵。一旦B2處理完Bookie上所有Entry，無法讀取到新的Entry了胳施，數(shù)據(jù)恢復就完成了溯祸；
5. B2將Ledger的狀態(tài)設置為CLOSED；
6. B2創(chuàng)建一個新Ledger舞肆，并開始接收Topic的消息寫入和讀冉垢ā；

Fencing解決方案解決了腦裂問題椿胯，也沒有數(shù)據(jù)丟失筷登。故障恢復，Ledger的狀態(tài)流程圖哩盲，

20220112-Pulsar 工作原理-2022-01-17-10-15-01.png

這個方案和 Raft Leader 的故障恢復機制實際上是沒有什么差別的前方，應該是有所借鑒。至于解決了腦裂廉油，這個也不是真正解決惠险，也是由于消息系統(tǒng)的特性導致的直接結(jié)果。為什么這么說呢抒线？

Raft 中也有類似的概念班巩，叫 committed index（Pulsar與之對應的是LAC），只有在收到多數(shù)節(jié)點寫入 Entry 返回成功之后嘶炭，才可以更新 committed index抱慌，再更新 Entry 到狀態(tài)機中逊桦，并返回給客戶端。對尚未更新 committed index 的 Entry抑进，Raft 也是不可讀的卫袒。可以發(fā)現(xiàn)单匣，Pulsar 的 Fencing 和 Raft 的機制幾乎一致夕凝，但是 Raft 有腦裂問題。

先回顧一下 Raft 的腦裂問題户秤。當 Raft Leader 節(jié)點故障發(fā)生時码秉，例如 Raft Leader 網(wǎng)絡斷開，其他節(jié)點已經(jīng)發(fā)現(xiàn)當前 Leader 超時鸡号，并發(fā)起下一輪選舉投票转砖，快速選舉出新的 Leader，但是老 Raft Leader 的 Follower 無響應超時時間尚未到達鲸伴，導致老 Leader 仍然認為自己是真正的 Leader府蔗，并響應客戶端的請求，因此導致客戶端讀取到了舊的數(shù)據(jù)汞窗。而與此同時姓赤，部分客戶端連接到了新 Raft Leader，寫入并讀取到新的數(shù)據(jù)仲吏，造成不一致不铆，這是 Raft 發(fā)生腦裂的原因。Raft 發(fā)生腦裂不會持續(xù)很長時間，當老 Leader 發(fā)現(xiàn)長時間沒有收到 Follower 響應而超時（主要取決于超時參數(shù)的配置），或者發(fā)現(xiàn)有新 Leader 產(chǎn)生時竿裂，老 Leader 就會將自己重置為 Follower。

那使用了同樣機制的 Pulsar 為什么就沒有腦裂問題呢劳坑？那是因為，Pulsar 是個消息系統(tǒng)成畦，寫入的消息類似 WAL距芬，是不可變的（immutable），追加的羡鸥。當發(fā)生和 Raft 一樣的故障的時候，老的 Pulsar Broker 也會讀到老的數(shù)據(jù)存和，但老的數(shù)據(jù)仍然合法，因為對同樣的 Cursor，在新的 Broker 上也是讀到相同的數(shù)據(jù)，只要讀取 Entry 不超過 LAC 就沒問題，最多只是無法獲取到最新的消息而已，獲取的消息并不會錯蓄愁。而 Raft 的存儲是偏向于通用存儲場景胀滚，因此就會有新舊數(shù)據(jù)版本不一致的問題戚炫。

腦裂一般都是指讀取數(shù)據(jù)發(fā)生的不一致钮惠，如果是寫入數(shù)據(jù)的腦裂，那可能是分布式算法有問題耙箍，成熟的算法一般不會有這個問題旨袒。

Bookie存儲

BookKeeper的存儲引擎是可插拔的，默認是DbLedgerStorage帆精，整體架構(gòu)如下襟企，

20220112-Pulsar 工作原理-2022-01-17-10-15-27.png

Bookie寫入的流程如下狮含，

20220112-Pulsar 工作原理-2022-01-17-10-15-39.png

Bookie是一個有WAL的消息存儲顽悼，寫入時，會先寫入WAL（Journal）几迄，再寫入Write Cache蔚龙。Write Cache會定期的將數(shù)據(jù)排序并寫入磁盤的Entry Log文件中。排序過程映胁，將不同Ledger的消息聚合在一起木羹，這樣，在讀取Ledger的時候解孙，就是完全的磁盤順序讀坑填。如果沒有排序聚合的話，就無法獲得順序讀的性能弛姜。

寫入Write Cache的時候脐瑰，也會把索引信息寫入RocksDB，索引信息很簡單廷臼，就是 (ledgerId, entryId) 到 (entryLogId, 文件偏移量) 的映射苍在。

Bookie可以緩存最近寫入的Entry和最近讀取的Entry绝页，讀取的順序是： Write Cache -> Read Cache -> Bookie上的Entry。當兩個緩存都沒有命中的時候忌穿，會到RocksDB中查找該Entry所在的文件和偏移量抒寂，并讀取該Entry，然后緩存再Read Cache中掠剑，以期之后可以命中屈芜。

BookKeeper可以支持磁盤IO分離，將寫入WAL的放在一個高速磁盤上朴译，其他數(shù)據(jù)放在低速磁盤上井佑。當有寫入Entry請求時，只會發(fā)生寫WAL的磁盤同步IO操作眠寿，其他都是寫入內(nèi)存緩存躬翁。同時，以異步的方式盯拱，將Write Cache中的數(shù)據(jù)以批量寫的方式寫入到Entry Log文件和RocksDB中盒发。

Bookie的Journal可以有多個，但和Ledger并不是一一對應的狡逢。4.5.0之后的BookKeeper可以配置journalDirectories參數(shù)宁舰，如，journalDirectories=/tmp/bk-journal1,/tmp/bk-journal2奢浑，配置多個目錄蛮艰，由Bookie統(tǒng)一管理。

Bookie故障恢復

當Bookie故障的時候雀彼，所有在這個Bookie上有Fragment的Ledger都需要復制壤蚜。恢復過程是重復制Fragment徊哑，來確保每個Ledger滿足Qw個復制因子袜刷。

有兩種恢復方法：自動和手動，主要討論自動方案莺丑。自動方案包括內(nèi)置的故障節(jié)點檢測機制水泉，手動就需要人為干預。具體的復制過程窒盐，兩者是一致的。

恢復過程可以通過在Bookie集群上運行AutoRecoveryMain來完成钢拧。其中一個自動恢復進程被選舉為Auditor蟹漓，Auditor來檢測故障的Bookie，然后源内，

1. 從Zookeeper讀取所有Ledger列表葡粒，找到故障Bookie上的Ledger份殿；
2. 對上述每個Ledger創(chuàng)建一個重復制任務，并記錄在Zookeeper的/underreplicated Znode上嗽交；

如果Auditor失敗卿嘲，就再選舉一個Auditor。Auditor只是AutoRecoveryMain的一個線程夫壁。AutoRecoveryMain也有運行Replication Task Worker的線程拾枣，每個Worker監(jiān)聽/underreplicated Znode獲取任務。發(fā)現(xiàn)任務后盒让，就嘗試lock住這個任務梅肤，如果lock失敗，說明其他Worker已經(jīng)拿到這個任務邑茄，就去尋找下一個任務姨蝴。

如果獲取到了鎖，那么需要肺缕，

1. 掃描Ledger的Fragments左医，找到那些當前Bookie不屬于的Fragment；
2. 對那些匹配的Fragment同木，從另一個Bookie上把數(shù)據(jù)復制到本地浮梢，然后更新Zookeeper，并將此Fragment標記為完全復制泉手；

如果Ledger的所有Fragment都已經(jīng)完全復制黔寇，則刪除/underreplicated任務；如果仍然存在未完全復制的Fragment斩萌，則釋放鎖缝裤，等待其他Worker處理。

如果一個Fragment沒有結(jié)束Entry id颊郎，Worker的復制任務會等待并再次檢查憋飞。如果還是沒有，說明之前的數(shù)據(jù)副本可能沒有完全寫入姆吭，會發(fā)起Fencing任務榛做，然后再繼續(xù)重復制。

注意：自動恢復機制和Fencing機制是有差別的内狸。
Fencing機制主要是處理Broker故障的場景检眯；自動恢復機制是處理Bookie故障的場景。
雖然自動恢復機制在某些邊界情況下回調(diào)用到Fencing機制昆淡。

總結(jié)

總結(jié)部分直接抄了中文翻譯锰瘸，

• 每個Topic都有一個歸屬的Broker。
• 每個Topic在邏輯上分解為Ledgers昂灵、Fragments和Entries避凝。
• Fragments分布在Bookie集群中舞萄。Topic與Bookie并不耦合。
• Fragments可以跨多個Bookies帶狀（Striping）分布管削。
• 當Pulsar Broker不可用時倒脓，該Broker持有的Topic所有權(quán)將轉(zhuǎn)移至其他的Broker。Fencing機制避免了同一個Topic當前的Ledger同時有兩個所有者(Broker)含思。
• 當Bookie不可用時崎弃，自動恢復(如果啟用)將自動進行數(shù)據(jù)重新復制到其他的Bookies。如果禁用茸俭，則可以手動啟動此過程吊履。
• Broker緩存尾部消息日志，可以非常高效的為尾部讀取操作提供服務调鬓。
• Bookies使用Journal提供持久化保證艇炎。該日志可用于故障恢復時恢復尚未寫入Entry Log文件的數(shù)據(jù)。
• 所有Topic的的條目都保存在Entry Log文件中腾窝。查找索引保存在RocksDB中缀踪。
• Bookies讀取邏輯如下：Write Cache -> Read Cache -> Log Entry Files(RocksDB 作為索引)
• Bookies可以通過單獨的磁盤做IO讀寫分離。
• Zookeeper存儲Pulsar和BookKeeper的所有元數(shù)據(jù)虹脯。如果Zookeeper不可用整個Pulsar將不可用驴娃。
• 存儲可以單獨擴展。如果存儲是瓶頸循集，那么只需要添加更多的Bookies唇敞，他們會自動承擔負載，不需要Rebalance咒彤。

后記

Jack Vanlightly在博客中表示疆柔，Pulsar的兩個突出特點是，

• 將Broker與存儲分離镶柱，結(jié)合BookKeeper的Fencing功能旷档，優(yōu)雅的解決了腦裂問題，并防止了數(shù)據(jù)丟失歇拆；
• 將Topic分割為Ledger和Fragment鞋屈，然后將將他們分布在整個Pulsar集群上，因此擴展變的容易故觅。新的數(shù)據(jù)自然會寫到新的Bookie上厂庇，不需要再進行再平衡（Rebalancing）；

后一點應該是獨創(chuàng)输吏，前一點應該是借鑒了Raft宋列，之前提到過。