對(duì)標(biāo) Kafka，消息中間件新秀 Pulsar 了解一下！

點(diǎn)擊關(guān)注公眾號(hào)，Java干貨及時(shí)送達(dá)

下圖是幾款消息中間件的歷史：

2012年pulsar在Yahoo內(nèi)部開發(fā)，2016年開源并捐獻(xiàn)給Apache，2018成為Apache頂級(jí)項(xiàng)目。

1架構(gòu)

pulsar的架構(gòu)圖如下：

總結(jié)一下，pulsar有下面的幾個(gè)特性。

1.1 計(jì)算存儲(chǔ)分離

pulsar采用計(jì)算和存儲(chǔ)相分離的架構(gòu)，Broker集群負(fù)責(zé)把producer發(fā)出的消息發(fā)送給consumer，同時(shí)承擔(dān)負(fù)載均衡的作用。

Pulsar用 Apache BookKeeper作為持久化存儲(chǔ)，Broker持有BookKeeper client，把未確認(rèn)的消息發(fā)送到BookKeeper進(jìn)行保存。

BookKeeper是一個(gè)分布式的WAL(Write Ahead Log)系統(tǒng)，pulsar使用BookKeeper有下面幾個(gè)便利：

• 可以為Topic創(chuàng)建多個(gè)ledgers

Ledger是一個(gè)只追加的數(shù)據(jù)結(jié)構(gòu)，并且只有一個(gè)writer，這個(gè)writer負(fù)責(zé)多個(gè)BookKeeper存儲(chǔ)節(jié)點(diǎn)(就是Bookies)的寫入。Ledger的條目會(huì)被復(fù)制到多個(gè)bookies。

• Broker可以創(chuàng)建、關(guān)閉和刪除Ledger，也可以追加內(nèi)容到Ledger。
• Ledger被關(guān)閉后，只能以只讀狀態(tài)打開，除非要明確地寫數(shù)據(jù)或者是因?yàn)閣riter掛掉導(dǎo)致的關(guān)閉。
• Ledger只能有writer這一個(gè)進(jìn)程寫入，這樣寫入不會(huì)有沖突，所以寫入效率很高。如果writer掛了，Ledger會(huì)啟動(dòng)恢復(fù)進(jìn)程來確定Ledger最終狀態(tài)和最后提交的日志，保證之后所有Ledger進(jìn)程讀取到相同的內(nèi)容。
• 除了保存消息數(shù)據(jù)外，還會(huì)保存cursors，也就是消費(fèi)端訂閱消費(fèi)的位置。這樣所有cursors消費(fèi)完一個(gè)Ledger的消息后這個(gè)Ledger就可以被刪除，這樣可以實(shí)現(xiàn)ledgers的定期翻滾從頭寫。

1.2 節(jié)點(diǎn)對(duì)等

從架構(gòu)圖可以看出，broker節(jié)點(diǎn)不保存數(shù)據(jù)，所有broker節(jié)點(diǎn)都是對(duì)等的。如果一個(gè)broker宕機(jī)了，不會(huì)丟失任何數(shù)據(jù)，只需要把它服務(wù)的topic遷移到一個(gè)新的broker上就行。

Broker的topic擁有多個(gè)邏輯分區(qū)，同時(shí)每個(gè)分區(qū)又有多個(gè)segment，writer寫數(shù)據(jù)時(shí)，首先會(huì)選擇Bookies，比如圖中的segment1，選擇了Bookie1、Bookie2、Bookie4，然后并發(fā)地寫下去。這樣這三個(gè)節(jié)點(diǎn)并沒有主從關(guān)系，協(xié)調(diào)完全依賴于writer，因此它們也是對(duì)等的。

1.3 擴(kuò)展和擴(kuò)容

在遇到雙十一等大流量的場(chǎng)景時(shí)，必須增加consumer，這時(shí)因?yàn)锽roker不存儲(chǔ)任何數(shù)據(jù)，可以方便的增加broker。broker集群會(huì)有一個(gè)或多個(gè)broker做消息負(fù)載均衡，當(dāng)新的broker加入后，流量會(huì)自動(dòng)從壓力大的broker上遷移過來。

對(duì)于BookKeeper，如果對(duì)存儲(chǔ)要求變高，比如之前存儲(chǔ)2個(gè)副本，現(xiàn)在需要存儲(chǔ)4個(gè)副本，這時(shí)可以單獨(dú)擴(kuò)展bookies而不用考慮broker。因?yàn)楣?jié)點(diǎn)對(duì)等，之前節(jié)點(diǎn)的segment又堆放整齊，加入新節(jié)點(diǎn)并不用搬移數(shù)據(jù)。writer會(huì)感知新的節(jié)點(diǎn)并優(yōu)先選擇使用。

1.4 容錯(cuò)機(jī)制

對(duì)于broker，因?yàn)椴槐４嫒魏螖?shù)據(jù)，如果節(jié)點(diǎn)宕機(jī)了，就相當(dāng)于客戶端斷開，重新連接其他的broker就可以了。

對(duì)于BookKeeper，因?yàn)楸４媪硕喾莞北荆⑶疫@些副本都是對(duì)等的，沒有主從關(guān)系，所以當(dāng)一個(gè)節(jié)點(diǎn)宕機(jī)后，不用立即恢復(fù)，后臺(tái)有一個(gè)線程會(huì)檢查宕機(jī)節(jié)點(diǎn)的數(shù)據(jù)備份進(jìn)行恢復(fù)。

2 BookKeeper簡(jiǎn)介

從上一節(jié)的講解看出，Apache Bookkeeper是一個(gè)易擴(kuò)展、高可用、運(yùn)維簡(jiǎn)單的分布式存儲(chǔ)系統(tǒng)。這節(jié)再看一下Bookkeeper的其他三個(gè)特性。

2.1 客戶端數(shù)量

我們知道，在Kafka中，客戶端只能從leader節(jié)點(diǎn)讀取數(shù)據(jù)。但在BookKeeper中，客戶端可以從任何一個(gè)bookie副本讀取數(shù)據(jù)，這有三個(gè)好處：

• 增加了讀高可用
• 把客戶端流量平均分配到了不同的bookie
• 可以通過增加客戶端數(shù)量來提高讀取效率

客戶端和服務(wù)器通信采用Netty實(shí)現(xiàn)異步I/O。網(wǎng)絡(luò)I/O使用單個(gè)TCP連接進(jìn)行多路復(fù)用，這就以很少的資源消耗實(shí)現(xiàn)了非常高的吞吐量。

2.3 I/O隔離

為什么要做I/O隔離？在大多數(shù)消息系統(tǒng)中，如果consumer處理慢，可能會(huì)導(dǎo)致消息積壓。這迫使存儲(chǔ)系統(tǒng)從持久存儲(chǔ)介質(zhì)中讀取數(shù)據(jù)。當(dāng)存儲(chǔ)系統(tǒng)I/O組件共享寫入、追尾讀、追趕讀的單一路徑時(shí)，就會(huì)出現(xiàn)I/O抖動(dòng)及頁(yè)面緩存的換入換出。

寫入和追尾讀對(duì)可預(yù)測(cè)的低延遲有較高要求，而追趕讀則對(duì)吞吐量的要求比較高，分離這三個(gè)路徑很重要。

在BookKeeper中，bookie使用三條獨(dú)立的I/O路徑，分別用于寫入、追尾讀、追趕讀。如下圖：

3 多租戶

Pulsar可以使用多租戶來管理大集群。Pulsar的租戶可以跨集群分布，每個(gè)租戶都可以有單獨(dú)的認(rèn)證和授權(quán)機(jī)制。租戶也是存儲(chǔ)配額、消息TTL和隔離策略的管理單元。

Pulsar的多租戶性質(zhì)主要體現(xiàn)在topic的URL中，其結(jié)構(gòu)如下：

persistent://tenant/namespace/topic

可以看到，租戶是topic的最基本單位。

假如一個(gè)公司有三個(gè)部門，tenant1、tenant2、tenant3，可以分配三個(gè)租戶，這三個(gè)租戶互不干擾，如下圖：

如果消息平臺(tái)不支持租戶，那部門之間想要隔離，就要給每個(gè)部門部署一套集群，運(yùn)維成本非常高。

4 消息模型

4.1 消息結(jié)構(gòu)

首先看一下Pulsar的消息結(jié)構(gòu)，如下圖：

消息流由多個(gè)獨(dú)立的segment組成，(這里的segment就是上面講的ledger),segment又包含獨(dú)立的entry，entry又由獨(dú)立的message組成。這里的message就是consumer發(fā)來的消息。

可以看到，一個(gè)message的id組成包括Ledger-id，entry-id，batch-index，partition-index。

需要注意兩點(diǎn)：

• segment和entry都是BookKeeper里面的概念。
• pulsar作為消息平臺(tái)時(shí)，一個(gè)message就是一個(gè)entry。當(dāng)pulsar作為流平臺(tái)時(shí)，為了提高吞吐量，會(huì)開啟batch，這樣多個(gè)message組成一個(gè)entry。

4.2 創(chuàng)建過程

消息的創(chuàng)建過程如下圖：

消息創(chuàng)建后主要經(jīng)歷下面幾步：

1. 選擇一個(gè)partition
2. 發(fā)送到管理這個(gè)partition的broker
3. broker將消息并發(fā)的發(fā)送給N個(gè)bookie，這個(gè)N是可以配置的。broker持有BookKeeper的客戶端，也就是writer，writer收到寫請(qǐng)求后，會(huì)并發(fā)的寫入N個(gè)bookie。上圖中N=3。
4. bookie寫完消息后會(huì)給broker一個(gè)回復(fù)，broker收到指定數(shù)量的確認(rèn)消息后就會(huì)認(rèn)為寫B(tài)ookKeeper成功。這個(gè)數(shù)量是這個(gè)配置的，比如M，M越大，寫B(tài)ookKeeper延遲越大，數(shù)據(jù)一致性越高。因此這個(gè)配置要對(duì)一致性和延遲到進(jìn)行。

5 消費(fèi)模型

5.1 概要

Pulsar的消費(fèi)模型如下圖：

producer將消息發(fā)送給topic，topic下有多個(gè)partition，partition下面又有多個(gè)broker。

broker負(fù)責(zé)接收消息并把消息分配給給consumer，并把消息寫到BookKeeper。

broker還具有限流功能，可以根據(jù)限流閾值對(duì)producer的消息進(jìn)行限流。

consumer并不能直接從broker中獲取消息，consumer和broker之間有一個(gè)Subscription，Consumer通過Subscription獲取消息。

5.2 subscription

subscription有四種類型：

• 獨(dú)占模式(Exclusive):同一個(gè)topic只能有一個(gè)消費(fèi)者，如果多個(gè)消費(fèi)者，就會(huì)出錯(cuò)。
• 災(zāi)備模式(Failover)：同一個(gè)topic可以有多個(gè)消費(fèi)者，但是只能有一個(gè)消費(fèi)者消費(fèi)，其他消費(fèi)者作為故障轉(zhuǎn)移備用，如果當(dāng)前消費(fèi)者出了故障，就從備用消費(fèi)者中選擇一個(gè)進(jìn)行消費(fèi)。如下圖：

• 共享模式(Shared):同一個(gè)topic可以由多個(gè)消費(fèi)者訂閱和消費(fèi)。消息通過round robin輪詢機(jī)制分發(fā)給不同的消費(fèi)者，并且每個(gè)消息僅會(huì)被分發(fā)給一個(gè)消費(fèi)者。當(dāng)消費(fèi)者斷開，發(fā)送給它的沒有被消費(fèi)的消息還會(huì)被重新分發(fā)給其它存活的消費(fèi)者。如下圖：

• Key_Shared：消息和消費(fèi)者都會(huì)綁定一個(gè)key，消息只會(huì)發(fā)送給綁定同一個(gè)key的消費(fèi)者。如果有新消費(fèi)者建立連接或者有消費(fèi)者斷開連接，就需要更新一些消息的key。如下圖：

跟Shared模式相比，Key_Shared的好處是既可以讓消費(fèi)者并發(fā)地消費(fèi)消息，又能保證同一Key下的消息順序。

5.3 Cursor

當(dāng)多個(gè)consumer訂閱同一個(gè)topic時(shí)，Subscription為每一個(gè)consumer分配一個(gè)Cursor，這樣多個(gè)Consumer之間就不會(huì)相互影響了。如下圖：

Subscription會(huì)維護(hù)一個(gè)消息的ACK狀態(tài)，consumer處理完消息后，會(huì)給broker返回ACK，表示消息已經(jīng)處理完成。如果broker一直沒有收到ACK，就會(huì)把消息發(fā)送到其他consumer。

如果客戶端想要重新消費(fèi)Cursor以前的消息，Cursor是支持reset的，reset之后，Cursor就回退回去了，這時(shí)consumer可以從新的Cursor位置進(jìn)行消費(fèi)。

Cursor的位置是會(huì)實(shí)時(shí)寫入BookKeeper的，這必定會(huì)有一定的性能損耗。因此，Pulsar提供了一種非持久化的Subscription(Non-durable Exclusive)。Pulsar的Reader接口內(nèi)嵌了Non—durable Exclusive Cursor，它讀取消息不會(huì)返回ACK。

6 broker代理

通過前面的講解可以看到，consumer和producer只需要跟broker進(jìn)行交互，而不用跟底層的BookKeeper交互，事實(shí)上，broker還有一層代理，consumer和producer直接跟代理進(jìn)行交互。如下圖：

7 Zookeeper

Pulsar提供了System topic用來保存策略之類的元數(shù)據(jù)，盡量減少對(duì)Zookeeper的依賴。

Zookeeper也保存一些策略相關(guān)的元數(shù)據(jù)，還保存了broker和BookKeeper集群相關(guān)的配置元數(shù)據(jù)，比如服務(wù)發(fā)現(xiàn)相關(guān)的元數(shù)據(jù)。

8 總結(jié)

Pulsar是一款非常優(yōu)秀的中間件，實(shí)現(xiàn)了計(jì)算和存儲(chǔ)相分離，支持多租戶，擴(kuò)展和擴(kuò)容、容錯(cuò)都是非常容易的。