[[376829]]

本文轉(zhuǎn)載自微信公眾號(hào)「咖啡拿鐵」，作者咖啡拿鐵 。轉(zhuǎn)載本文請(qǐng)聯(lián)系咖啡拿鐵公眾號(hào)。

背景

之前琢磨了很久一直想寫一篇pulsar相關(guān)的文章，但是一直知識(shí)儲(chǔ)備不夠，對(duì)于很多細(xì)節(jié)還是不了解，于是查了很多資料，總算是可以湊出一篇文章了。

Pulsar是一個(gè)由yahoo公司于2016年開源的消息中間件，2018年成為Apache的頂級(jí)項(xiàng)目。在我之前的文章中寫過(guò)很多其他消息中間件的文章，比如kafka,rocketmq等等。

在開源的業(yè)界已經(jīng)有這么多消息隊(duì)列中間件了，pulsar作為一個(gè)新勢(shì)力到底有什么優(yōu)點(diǎn)呢?pulsar自從出身就不斷的再和其他的消息隊(duì)列(kafka,rocketmq等等)做比較，但是Pulsar的設(shè)計(jì)思想和大多數(shù)的消息隊(duì)列中間件都不同，具備了高吞吐，低延遲，計(jì)算存儲(chǔ)分離，多租戶，異地復(fù)制等功能，所以pulsar也被譽(yù)為下一代消息隊(duì)列中間件，接下來(lái)我會(huì)一一對(duì)其進(jìn)行詳細(xì)的解析。

pulsar架構(gòu)原理

整體的架構(gòu)和其他的消息隊(duì)列中間件差別不是太大，相信大家也看到了很多熟悉的名詞，接下來(lái)會(huì)給大家一一解釋這些名詞的含義。

名詞解釋

• Producer:消息生產(chǎn)者，將消息發(fā)送到broker。
• Consumer:消息消費(fèi)者，從Broker讀取消息到客戶端，進(jìn)行消費(fèi)處理。
• Broker: 可以看作是pulsar的server,Producer和Consumer都看作是client.消息處理的節(jié)點(diǎn)，pulsar的Broker和其他消息中間件的都不一樣，他是無(wú)狀態(tài)的沒有存儲(chǔ)，所以可以無(wú)限制的擴(kuò)展，這個(gè)后面也會(huì)詳解講到。
• Bookie: 負(fù)責(zé)所有消息的持久化，這里采用的是Apache Bookeeper。
• ZK: 和kafka一樣pulsar也是使用zk保存一些元數(shù)據(jù)，比如配置管理,topic分配，租戶等等。
• Service Discovery：可以理解為Pulsar中的nginx，只用一個(gè)url就可以和整個(gè)broker進(jìn)行打交道，當(dāng)然也可以使用自己的服務(wù)發(fā)現(xiàn)。客戶端發(fā)出的讀取，更新或刪除主題的初始請(qǐng)求將發(fā)送給可能不是處理該主題的 broker 。如果這個(gè) broker 不能處理該主題的請(qǐng)求，broker 將會(huì)把該請(qǐng)求重定向到可以處理主題請(qǐng)求的 broker。

不論是kafka,rocketmq還是我們的pulsar其實(shí)作為消息隊(duì)列中間件最為重要的大概就是分為三個(gè)部分：

• Producer是如何生產(chǎn)消息，發(fā)送到對(duì)應(yīng)的Broker
• Broker是如何處理消息，將高效的持久化以及查詢
• Consumer是如何進(jìn)行消費(fèi)消息

而我們后面也會(huì)圍繞著這三個(gè)部分進(jìn)行展開講解。

Producer生產(chǎn)消息

先簡(jiǎn)單看一下如何用代碼進(jìn)行消息發(fā)送：

PulsarClient client = PulsarClient.create("pulsar://pulsar.us-west.example.com:6650"); 
 
Producer producer = client.createProducer( 
                "persistent://sample/standalone/ns1/my-topic"); 
 
// Publish 10 messages to the topic 
for (int i = 0; i < 10; i++) { 
    producer.send("my-message".getBytes()); 
} 

• Step1: 首先使用我們的url創(chuàng)建一個(gè)client這個(gè)url是我們service discovery的地址，如果我們使用單機(jī)模式可以進(jìn)行直連
• Step2：我們傳入了一個(gè)類似url的參數(shù)，我們只需要傳遞這個(gè)就能指定我們到底在哪個(gè)topic或者namespace下面創(chuàng)建的：

• Step3: 調(diào)用send方法發(fā)送消息，這里也提供了sendAsync方法支持異步發(fā)送。

上面三個(gè)步驟中，步驟1，2屬于我們準(zhǔn)備階段，用于構(gòu)建客戶端，構(gòu)建Producer，我們真的核心邏輯在send中，那這里我先提幾個(gè)小問題，大家可以先想想在其他消息隊(duì)列中是怎么做的，然后再對(duì)比pulsar的看一下：

• 我們調(diào)用了send之后是會(huì)立即發(fā)送嗎?
• 如果是多partition，怎么找到我應(yīng)該發(fā)送到哪個(gè)Broker呢?

發(fā)送模式

我們上面說(shuō)了send分為async和sync兩種模式，但實(shí)際上在pulsar內(nèi)部sync模式也是采用的async模式，在sync模式下模擬回調(diào)阻塞，達(dá)到同步的效果，這個(gè)在kafka中也是采用的這個(gè)模式，但是在rocketmq中，所有的send都是真正的同步，都會(huì)直接請(qǐng)求到broker。

基于這個(gè)模式，在pulsar和kafka中都支持批量發(fā)送，在rocketmq中是直接發(fā)送，批量發(fā)送有什么好處呢?當(dāng)我們發(fā)送的TPS特別高的時(shí)候，如果每次發(fā)送都直接和broker直連，可能會(huì)做很多的重復(fù)工作，比如壓縮，鑒權(quán)，創(chuàng)建鏈接等等。比如我們發(fā)送1000條消息，那么可能會(huì)做1000次這個(gè)重復(fù)的工作，如果是批量發(fā)送的話這1000條消息合并成一次請(qǐng)求，相對(duì)來(lái)說(shuō)壓縮，鑒權(quán)這些工作就只需要做一次。

有同學(xué)可能會(huì)問，批量發(fā)送會(huì)不會(huì)導(dǎo)致發(fā)送的時(shí)間會(huì)有一定的延誤?這個(gè)其實(shí)不需要擔(dān)心，在pulsar中默認(rèn)定時(shí)每隔1ms發(fā)送一次batch,或者當(dāng)batchsize默認(rèn)到了1000都會(huì)進(jìn)行發(fā)送，這個(gè)發(fā)送的頻率都還是很快的。

發(fā)送負(fù)載均衡

在消息隊(duì)列中通常會(huì)將topic進(jìn)行水平擴(kuò)展，在pulsar和kafka中叫做partition,在rocketmq中叫做queue，本質(zhì)上都是分區(qū)，我們可以將不同分區(qū)落在不同的broker上，達(dá)到我們水平擴(kuò)展的效果。

在我們發(fā)送的時(shí)候可以自己制定選擇partition的策略，也可以使用它默認(rèn)輪訓(xùn)partition策略。當(dāng)我們選擇了partition之后，我們?cè)趺创_定哪一個(gè)partition對(duì)應(yīng)哪一個(gè)broker呢?

可以先看看下面這個(gè)圖：

• Step1: 我們所有的信息分區(qū)映射信息在zk和broker的緩存中都有進(jìn)行存儲(chǔ)。
• Step2: 我們通過(guò)查詢broker，可以獲取到分區(qū)和broker的關(guān)系，并且定時(shí)更新。
• Step3: 在pulsar中每個(gè)分區(qū)在發(fā)送端的時(shí)候都被抽象成為一個(gè)單獨(dú)的Producer,這個(gè)和kafka,rocketmq都不一樣，在kafka里面大概就是選擇了partition之后然后再去找partition對(duì)應(yīng)的broker地址，然后進(jìn)行發(fā)送。pulsar將每一個(gè)partition都封裝成Producer，在代碼實(shí)現(xiàn)上就不需要去關(guān)注他具體對(duì)應(yīng)的是哪個(gè)broker,所有的邏輯都在producer這個(gè)代碼里面，整體來(lái)說(shuō)比較干凈。

壓縮消息

消息壓縮是優(yōu)化信息傳輸?shù)氖侄沃唬覀兺ǔ？匆娨恍┐笮臀募紩?huì)是以一個(gè)壓縮包的形式提供下載，在我們消息隊(duì)列中我們也可以用這種思想，我們將一個(gè)batch的消息，比如有1000條可能有1M的傳輸大小，但是經(jīng)過(guò)壓縮之后可能就只會(huì)有幾十kb，增加了我們和broker的傳輸效率，但是與之同時(shí)我們的cpu也帶來(lái)了損耗。Pulsar客戶端支持多種壓縮類型，如 lz4、zlib、zstd、snappy 等。

client.newProducer() 
    .topic(“test-topic”) 
    .compressionType(CompressionType.LZ4) 
    .create(); 

Broker

接下來(lái)我們來(lái)說(shuō)說(shuō)第二個(gè)比較重要的部分Broker,在Broker的設(shè)計(jì)中pulsar和其他所有的消息隊(duì)列差別比較大，而正是因?yàn)檫@個(gè)差別也成為了他的特點(diǎn)。

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

首先我們來(lái)說(shuō)說(shuō)他最大的特點(diǎn)：計(jì)算和存儲(chǔ)分離。我們?cè)陂_始的說(shuō)過(guò)Pulsar是下一代消息隊(duì)列，就非常得益于他這個(gè)架構(gòu)設(shè)計(jì)，無(wú)論是kafka還是RocketMQ,所有的計(jì)算和存儲(chǔ)都放在同一個(gè)機(jī)器上，這個(gè)模式有幾個(gè)弊端：

• 擴(kuò)展困難：當(dāng)我們需要擴(kuò)展的集群的時(shí)候，我們通常是因?yàn)閏pu或者磁盤其中一個(gè)原因影響，但是我們卻要申請(qǐng)一個(gè)可能cpu和磁盤配置都很好的機(jī)器，造成了資源浪費(fèi)。并且kafka這種進(jìn)行擴(kuò)展，還需要進(jìn)行遷移數(shù)據(jù)，過(guò)程十分繁雜。
• 負(fù)載不均衡：當(dāng)某些partion數(shù)據(jù)特別多的時(shí)候，會(huì)導(dǎo)致broker負(fù)載不均衡,如下面圖，如果某個(gè)partition數(shù)據(jù)特別多，那么就會(huì)導(dǎo)致某個(gè)broker(輪船)承載過(guò)多的數(shù)據(jù)，但是另外的broker可能又比較空閑

pulsar計(jì)算分離架構(gòu)能夠非常好的解決這個(gè)問題:

• 對(duì)于計(jì)算：也就是我們的broker,提供消息隊(duì)列的讀寫,不存儲(chǔ)任何數(shù)據(jù)，無(wú)狀態(tài)對(duì)于我們擴(kuò)展非常友好，只要你機(jī)器足夠，就能隨便上。擴(kuò)容Broker往往適用于增加Consumer的吞吐，當(dāng)我們有一些大流量的業(yè)務(wù)或者活動(dòng)，比如電商大促，可以提前進(jìn)行broker的擴(kuò)容。
• 對(duì)于存儲(chǔ)：也就是我們的bookie,只提供消息隊(duì)列的存儲(chǔ)，如果對(duì)消息量有要求的，我們可以擴(kuò)容bookie,并且我們不需要遷移數(shù)據(jù)，擴(kuò)容十分方便。

消息存儲(chǔ)

名詞解析：

上圖是bookie的讀寫架構(gòu)圖，里面有一些名詞需要先介紹一下：

• Entry,Entry是存儲(chǔ)到bookkeeper中的一條記錄，其中包含Entry ID，記錄實(shí)體等。
• Ledger，可以認(rèn)為ledger是用來(lái)存儲(chǔ)Entry的，多個(gè)Entry序列組成一個(gè)ledger。
• Journal，其實(shí)就是bookkeeper的WAL(write ahead log)，用于存bookkeeper的事務(wù)日志，journal文件有一個(gè)最大大小，達(dá)到這個(gè)大小后會(huì)新起一個(gè)journal文件。
• Entry log，存儲(chǔ)Entry的文件，ledger是一個(gè)邏輯上的概念，entry會(huì)先按ledger聚合，然后寫入entry log文件中。同樣，entry log會(huì)有一個(gè)最大值，達(dá)到最大值后會(huì)新起一個(gè)新的entry log文件
• Index file，ledger的索引文件，ledger中的entry被寫入到了entry log文件中，索引文件用于entry log文件中每一個(gè)ledger做索引，記錄每個(gè)ledger在entry log中的存儲(chǔ)位置以及數(shù)據(jù)在entry log文件中的長(zhǎng)度。
• MetaData Storage，元數(shù)據(jù)存儲(chǔ)，是用于存儲(chǔ)bookie相關(guān)的元數(shù)據(jù)，比如bookie上有哪些ledger，bookkeeper目前使用的是zk存儲(chǔ)，所以在部署bookkeeper前，要先有zk集群。

整體架構(gòu)上的寫流程：

• Step1: broker發(fā)起寫請(qǐng)求，首先對(duì)Journal磁盤寫入WAL，熟悉mysql的朋友知道redolog，journal和redolog作用一樣都是用于恢復(fù)沒有持久化的數(shù)據(jù)。
• Step2: 然后再將數(shù)據(jù)寫入index和ledger，這里為了保持性能不會(huì)直接寫盤，而是寫pagecache,然后異步刷盤。
• Step3: 對(duì)寫入進(jìn)行ack。

讀流程為：

• Step1: 先讀取index,當(dāng)然也是先讀取cache，再走disk。
• Step2: 獲取到index之后，根據(jù)index去entry logger中去對(duì)應(yīng)的數(shù)據(jù)

如何高效讀寫?

在kafka中當(dāng)我們的topic變多了之后，由于kafka一個(gè)topic一個(gè)文件，就會(huì)導(dǎo)致我們的磁盤IO從順序?qū)懽兂呻S機(jī)寫。在rocketMq中雖然將多個(gè)topic對(duì)應(yīng)一個(gè)寫入文件，讓寫入變成了順序?qū)懀俏覀兊淖x取很容易導(dǎo)致我們的pagecache被各種覆蓋刷新，這對(duì)于我們的IO的影響是非常大的。所以pulsar在讀寫兩個(gè)方面針對(duì)這些問題都做了很多優(yōu)化：

• 寫流程：順序?qū)?+ pagecache。在寫流程中我們的所有的文件都是獨(dú)立磁盤，并且同步刷盤的只有Journal，Journal是順序?qū)懸粋€(gè)journal-wal文件,順序?qū)懶史浅８摺edger和index雖然都會(huì)存在多個(gè)文件，但是我們只會(huì)寫入pagecache,異步刷盤，所以隨機(jī)寫不會(huì)影響我們的性能。
• 讀流程：broker cache + bookie cache，在pulsar中對(duì)于追尾讀(tailing read)非常友好基本不會(huì)走io,一般情況下我們的consumer是會(huì)立即去拿producer發(fā)送的消息的，所以這部分在持久化之后依然在broker中作為cache存在，當(dāng)然就算broker沒有cache(比如broker是新建的)，我們的bookie也會(huì)在memtable中有自己的cache,通過(guò)多重cache減少讀流程走io。

我們可以發(fā)現(xiàn)在最理想的情況下讀寫的io是完全隔離開來(lái)的，所以在Pulsar中能很容易就支持百萬(wàn)級(jí)topic，而在我們的kafka和rocketmq中這個(gè)是非常困難的。

無(wú)限流式存儲(chǔ)

一個(gè)Topic實(shí)際上是一個(gè)ledgers流(Segment)，通過(guò)這個(gè)設(shè)計(jì)所以Pulsar他并不是一個(gè)單純的消息隊(duì)列系統(tǒng)，他也可以代替流式系統(tǒng)，所以他也叫流原生平臺(tái),可以替代flink等系統(tǒng)。

可以看見我們的Event Stream(topic/partition)，由多個(gè)Segment存儲(chǔ)組成，而每個(gè)segment由entry組成，這個(gè)可以看作是我們每批發(fā)送的消息通常會(huì)看作是一個(gè)entry。

Segment可以看作是我們寫入文件的一個(gè)基本維度，同一個(gè)Segment的數(shù)據(jù)會(huì)寫在同一個(gè)文件上面，不同Segment將會(huì)是不同文件，而Segment之間的在metadata中進(jìn)行保存。

分層存儲(chǔ)

在kafka和rocketmq中消息是會(huì)有一定的保存時(shí)間的，因?yàn)榇疟P會(huì)有空間限制，在pulsar中也提供這個(gè)功能，但是如果你想讓自己的消息永久存儲(chǔ)，那么可以使用分級(jí)存儲(chǔ)，我們可以將一些比較老的數(shù)據(jù)，定時(shí)的刷新到廉價(jià)的存儲(chǔ)中，比如s3,那么我們就可以無(wú)限存儲(chǔ)我們的消息隊(duì)列了。

數(shù)據(jù)復(fù)制

在pulsar中的數(shù)據(jù)復(fù)制和kafka,rocketmq都有很大的不同，在其他消息隊(duì)列中通常是其他副本主動(dòng)同步，通常這個(gè)時(shí)間就會(huì)變得不可預(yù)測(cè)，而在pulsar采用了類似qurom協(xié)議，給一組可用的bookie池，然后并發(fā)的寫入其中的一部分bookie,只要返回部分成功(通常大于1/2)就好。

• Ensemble Size(E)決定給定 ledger 可用的 bookie 池大小。
• Write Quorum Size(Qw)指定 Pulsar 向其中寫入 entry 的 bookie 數(shù)量。
• Ack Quorum Size(Qa)指定必須 ack 寫入的 bookie 數(shù)量。

采用這種并發(fā)寫的方式，會(huì)更加高效的進(jìn)行數(shù)據(jù)復(fù)制，尤其是當(dāng)數(shù)據(jù)副本比較多的時(shí)候。

Consumer

接下來(lái)我們來(lái)聊聊pulsar中最后一個(gè)比較重要的組成consumer。

訂閱模式

訂閱模式是用來(lái)定義我們的消息如何分配給不同的消費(fèi)者，不同消息隊(duì)列中間件都有自己的訂閱模式，一般我們常見的訂閱模式有：

• 集群模式：一條消息只能被一個(gè)集群內(nèi)的消費(fèi)者所消費(fèi)。
• 廣播模式：一條消息能被集群內(nèi)所有的消費(fèi)者消費(fèi)。

在pulsar中提供了4種訂閱模式，分別是獨(dú)占，災(zāi)備，共享，鍵共享：

• 獨(dú)占：顧名思義只能由一個(gè)消費(fèi)者獨(dú)占，如果同一個(gè)集群內(nèi)有第二個(gè)消費(fèi)者去注冊(cè)，第二個(gè)就會(huì)失敗，這個(gè)適用于全局有序的消息。
• 災(zāi)備：加強(qiáng)版獨(dú)占，如果獨(dú)占的那個(gè)掛了，會(huì)自動(dòng)的切換到另外一個(gè)好的消費(fèi)者，但是還是只能由一個(gè)獨(dú)占。
• 共享模式：這個(gè)模式看起來(lái)有點(diǎn)像集群模式，一條消息也是只能被一個(gè)集群內(nèi)消費(fèi)者消費(fèi)，但是和rocketmq不同的是，rocketmq是以partition維度，同一個(gè)Partition的數(shù)據(jù)都會(huì)被發(fā)到一個(gè)機(jī)器上。在Pulsar中消費(fèi)不會(huì)以partition維度，而是輪訓(xùn)所有消費(fèi)者進(jìn)行消息發(fā)送。這有個(gè)什么好處呢?如果你有100臺(tái)機(jī)器，但是你只有10個(gè)partition其實(shí)你只有10臺(tái)消費(fèi)者能運(yùn)轉(zhuǎn)，但是在pulsar中100臺(tái)機(jī)器都可以進(jìn)行消費(fèi)處理。
• 鍵共享：類似上面說(shuō)的partition維度去發(fā)送，在rocketmq中同一個(gè)key的順序消息都會(huì)被發(fā)送到一個(gè)partition，但是這里不會(huì)有partition維度，而只是按照key的hash去分配到固定的consumer,也解決了消費(fèi)者能力限制于partition個(gè)數(shù)問題。

消息獲取模式

不論是在kafka還是在rocketmq中我們都是client定時(shí)輪訓(xùn)我們的broker獲取消息，這種模式叫做長(zhǎng)輪訓(xùn)(Long-Polling)模式。這種模式有一個(gè)缺點(diǎn)網(wǎng)絡(luò)開銷比較大，我們來(lái)計(jì)算一下consumer被消費(fèi)的時(shí)延，我們假設(shè)broker和consumer之間的一次網(wǎng)絡(luò)延時(shí)為R,那么我們總共的時(shí)間為：

• 當(dāng)某一條消息A剛到broker的，這個(gè)時(shí)候long-polling剛好打包完數(shù)據(jù)返回，broker返回到consumer這個(gè)時(shí)間為R。
• consumer又再次發(fā)送request請(qǐng)求，這個(gè)又為R。
• 將我們的消息A返回給consumer這里又為R。

如果只考慮網(wǎng)絡(luò)時(shí)延，我們可以看見我們這條消息的消費(fèi)時(shí)延大概是3R，所以我們必須想點(diǎn)什么對(duì)其進(jìn)行一些優(yōu)化，有同學(xué)可能馬上就能想到，我們消息來(lái)了直接推送給我們的consumer不就對(duì)了，這下我們的時(shí)延只會(huì)有一次R,這個(gè)就是我們常見的推模式，但是簡(jiǎn)單的推模式是有問題的，如果我們有生產(chǎn)速度遠(yuǎn)遠(yuǎn)大于消費(fèi)速度，那么推送的消息肯定會(huì)干爆我們的內(nèi)存，這個(gè)就是背壓。那么我們?cè)趺唇鉀Q背壓呢?我們就可以優(yōu)化推送方式，將其變?yōu)閯?dòng)態(tài)推送，我們結(jié)合Long-polling,在long-polling請(qǐng)求時(shí)將Buffer剩余空間告知給Broker，由Broker負(fù)責(zé)推送數(shù)據(jù)。此時(shí)Broker知道最多可以推送多少條數(shù)據(jù)，那么就可以控制推送行為，不至于沖垮Consumer。

舉個(gè)例子：

Consumer發(fā)起請(qǐng)求時(shí)Buffer剩余容量為100，Broker每次最多返回32條消息，那么Consumer的這次long-polling請(qǐng)求Broker將在執(zhí)行3次push(共push96條消息)之后返回response給Consumer(response包含4條消息)。

如果采用long-polling模型，Consumer每發(fā)送一次請(qǐng)求Broker執(zhí)行一次響應(yīng)，這個(gè)例子需要進(jìn)行4次long-polling交互(共4個(gè)request和4個(gè)response，8次網(wǎng)絡(luò)操作;Dynamic Push/Pull中是1個(gè)request，三次push和一個(gè)response，共5次網(wǎng)絡(luò)操作)。

所以pulsar就采用了這種消息獲取模式，從consumer層進(jìn)一步優(yōu)化消息達(dá)到時(shí)間。我覺得這個(gè)設(shè)計(jì)非常巧妙，很多中間件的這種long-polling模式都可以參考這種思想去做一個(gè)改善。

總結(jié)

Apache Pulsar很多設(shè)計(jì)思想都和其他中間件不一樣，但無(wú)疑于其更加貼近于未來(lái)，大膽預(yù)測(cè)一下其他的一些消息中間件未來(lái)的發(fā)展也都會(huì)向其靠攏，目前國(guó)內(nèi)的Pulsar使用者也是越來(lái)越多，騰訊云提供了pulsar的云版本TDMQ，當(dāng)然還有一些其他的知名公司華為，知乎，虎牙等等有都在對(duì)其做一個(gè)逐步的嘗試，我相信pulsar真的是一個(gè)趨勢(shì)。最后也讓我想起了最近大江大河大結(jié)局的一句話：

所有的變化,都可能伴隨著痛苦和彎路,開放的道路,也不會(huì)是闊野坦途,但大江大河,奔涌向前的趨勢(shì),不是任何險(xiǎn)灘暗礁,能夠阻擋的。道之所在，雖千萬(wàn)人吾往矣。