背景

我們最近在做新業(yè)務(wù)的技術(shù)選型，其中涉及到了對消息中間件的選擇；結(jié)合我們的實際情況希望它能滿足以下幾個要求：

友好的云原生支持：因為現(xiàn)在的主力語言是 Go，同時在運維上能夠足夠簡單。
官方支持多種語言的 SDK：還有一些 Python、Java 相關(guān)的代碼需要維護。
最好是有一些方便好用的特性，比如：延時消息、死信隊列、多租戶等。

當然還有一些水平擴容、吞吐量、低延遲這些特性就不用多說了，幾乎所有成熟的消息中間件都能滿足這些要求。

基于以上的篩選條件，Pulsar 進入了我們的視野。

作為 Apache 下的頂級項目，以上特性都能很好的支持。

下面我們來它有什么過人之處。

架構(gòu)

從官方的架構(gòu)圖中可以看出 Pulsar 主要有以下組件組成：

Broker 無狀態(tài)組件，可以水平擴展，主要用于生產(chǎn)者、消費者連接；與 Kafka 的 broker 類似，但沒有數(shù)據(jù)存儲功能，因此擴展更加輕松。
BookKeeper 集群：主要用于數(shù)據(jù)的持久化存儲。
Zookeeper 用于存儲 broker 與 BookKeeper 的元數(shù)據(jù)。

整體一看似乎比 Kafka 所依賴的組件還多，這樣確實會提供系統(tǒng)的復(fù)雜性；但同樣的好處也很明顯。

Pulsar 的存儲于計算是分離的，當需要擴容時會非常簡單，直接新增 broker 即可，沒有其他的心智負擔。

當存儲成為瓶頸時也只需要擴容 BookKeeper，不需要人為的做重平衡，BookKeeper 會自動負載。

同樣的操作，Kafka 就要復(fù)雜的多了。

特性

多租戶

多租戶也是一個剛需功能，可以在同一個集群中對不同業(yè)務(wù)、團隊的數(shù)據(jù)進行隔離。

persistent://core/order/create-order

以這個 topic 名稱為例，在 core 這個租戶下有一個 order 的 namespace，最終才是 create-order 的 topic 名稱。

在實際使用中租戶一般是按照業(yè)務(wù)團隊進行劃分，namespace 則是當前團隊下的不同業(yè)務(wù)；這樣便可以很清晰的對 topic 進行管理。

通常有對比才會有傷害，在沒有多租戶的消息中間件中是如何處理這類問題的呢：

干脆不分這么細，所有業(yè)務(wù)線混著用，當團隊較小時可能問題不大；一旦業(yè)務(wù)增加，管理起來會非常麻煩。
自己在 topic 之前做一層抽象，但其實本質(zhì)上也是在實現(xiàn)多租戶。
各個業(yè)務(wù)團隊各自維護自己的集群，這樣當然也能解決問題，但運維復(fù)雜度自然也就提高了。

以上就很直觀的看出多租戶的重要性了。

Function 函數(shù)計算

Pulsar 還支持輕量級的函數(shù)計算，例如需要對某些消息進行數(shù)據(jù)清洗、轉(zhuǎn)換，然后再發(fā)布到另一個 topic 中。

這類需求就可以編寫一個簡單的函數(shù)，Pulsar 提供了 SDK 可以方便的對數(shù)據(jù)進行處理，最后使用官方工具發(fā)布到 broker 中。

在這之前這類簡單的需求可能也需要自己處理流處理引擎。

應(yīng)用

除此之外的上層應(yīng)用，比如生產(chǎn)者、消費者這類概念與使用大家都差不多。

比如 Pulsar 支持四種消費模式：

Exclusive：獨占模式，同時只有一個消費者可以啟動并消費數(shù)據(jù)；通過 SubscriptionName 標明是同一個消費者），適用范圍較小。
Failover 故障轉(zhuǎn)移模式：在獨占模式基礎(chǔ)之上可以同時啟動多個 consumer，一旦一個 consumer 掛掉之后其余的可以快速頂上，但也只有一個 consumer 可以消費；部分場景可用。
Shared 共享模式：可以有 N 個消費者同時運行，消息按照 round-robin 輪詢投遞到每個 consumer 中；當某個 consumer 宕機沒有 ack 時，該消息將會被投遞給其他消費者。這種消費模式可以提高消費能力，但消息無法做到有序。
KeyShared 共享模式：基于共享模式；相當于對同一個topic中的消息進行分組，同一分組內(nèi)的消息只能被同一個消費者有序消費。

第三種共享消費模式應(yīng)該是使用最多的，當對消息有順序要求時可以使用 KeyShared 模式。

SDK

官方支持的 SDK 非常豐富；我也在官方的 SDK 的基礎(chǔ)之上封裝了一個內(nèi)部使用的 SDK。

因為我們使用了 dig 這樣的輕量級依賴注入庫，所以使用起來大概是這個樣子：

 SetUpPulsar(lookupURL)
 container := dig.New()
 container.Provide(func() ConsumerConfigInstance {
  return NewConsumer(&pulsar.ConsumerOptions{
   Topic:            "persistent://core/order/create-order",
   SubscriptionName: "order-sub",
   Type:             pulsar.Shared,
   Name:             "consumer01",
  }, ConsumerOrder)

 })

 container.Provide(func() ConsumerConfigInstance {
  return NewConsumer(&pulsar.ConsumerOptions{
   Topic:            "persistent://core/order/update-order",
   SubscriptionName: "order-sub",
   Type:             pulsar.Shared,
   Name:             "consumer02",
  }, ConsumerInvoice)

 })

 container.Invoke(StartConsumer)