最近，我一直在研究 Pulsar 及其與 Kafka 的比較。通過快速搜索，你會看到這兩個最著名的開源消息傳遞系統(tǒng)之間正在進行的"戰(zhàn)爭"。

[[379649]] 

圖片來自 Pexels

作為 Kafka 的用戶，我著實對 Kafka 的某些問題感到困惑，但 Pulsar 卻讓人眼前一亮、令我非常興奮。所以最后，我設(shè)法花了一些時間了解背景資料，并且做了很多研究。

在本文中，我將重點介紹 Pulsar 的優(yōu)勢，并說明 Pulsar 勝于 Kafka 的理由。讓我們開始!

Kafka 基礎(chǔ)知識

Kafka 是消息傳遞系統(tǒng)之王。它由 LinkedIn 于 2011 年創(chuàng)建，并在 Confluent 的支持下得到了廣泛的傳播。

Confluent 已向開源社區(qū)發(fā)布了許多新功能和附加組件，例如用于模式演化的 Schema Registry，用于從其他數(shù)據(jù)源輕松流式傳輸?shù)?Kafka Connect 等。

數(shù)據(jù)庫到 Kafka，Kafka Streams 進行分布式流處理，最近使用 KSQL 對 Kafka topic 執(zhí)行類似 SQL 的查詢等等。

Kafka 快速，易于安裝，非常受歡迎，可用于廣泛的范圍或用例。從開發(fā)人員的角度來看，盡管 Apache Kafka 一直很友好，但在操作運維方面卻是一團糟。

因此，讓我們回顧一下 Kafka 的一些痛點：

 

Kafka 演示[2]

Kakfa的諸多痛點如下：

• 擴展 Kafka 十分棘手，這是由于 broker 與存儲數(shù)據(jù)的耦合架構(gòu)結(jié)構(gòu)所致。剝離一個 broker 意味著它必須復(fù)制 topic 分區(qū)和副本，這非常耗時。
• 沒有與租戶完全隔離的本地多租戶。
• 存儲會變得非常昂貴，盡管可以長時間存儲數(shù)據(jù)，但是由于成本問題卻很少用到它。
• 萬一副本不同步，有可能丟失消息。
• 必須提前計劃和計算 broker、topic、分區(qū)和副本的數(shù)量(確保計劃的未來使用量增長)，以避免擴展問題，這非常困難。
• 如果僅需要消息傳遞系統(tǒng)，則使用偏移量可能會很復(fù)雜。
• 集群重新平衡會影響相連的生產(chǎn)者和消費者的性能。
• MirrorMaker[3] Geo 復(fù)制機制存在問題。像 Uber 這樣的公司已經(jīng)創(chuàng)建了自己的解決方案來克服這些問題。

如您所見，大多數(shù)問題與操作運維方面有關(guān)。盡管安裝起來相對容易，但 Kafka 難以管理和調(diào)優(yōu)。而且，它也缺乏應(yīng)有的靈活和彈性。

Pulsar 基礎(chǔ)知識

Pulsar 由 Yahoo!在 2013 年創(chuàng)建，并于 2016 年捐贈給 Apache 基金會。Pulsar 現(xiàn)在是 Apache 軟件基金會的頂級項目。

Yahoo!、Verizon、Twitter 等公司已在生產(chǎn)中使用它來處理成千上萬消息。它具有運行成本低、靈活等特性。Pulsar 旨在解決 Kafka 的大部分難題，使其更易于擴展。

Pulsar 非常靈活：它既可以應(yīng)用于像 Kafka 這樣的分布式日志應(yīng)用場景，也可以應(yīng)用于像 RabbitMQ 這樣的純消息傳遞系統(tǒng)場景。

它支持多種類型的訂閱、多種交付保證、保留策略以及處理模式演變的方法，以及其他諸多特性。

 

Pulsar 架構(gòu)圖[4]

Pulsar 的特性如下：

• 內(nèi)置多租戶，不同的團隊可以使用相同的集群并將其隔離，解決了許多管理難題。它支持隔離、身份驗證、授權(quán)和配額。
• 多層體系結(jié)構(gòu)：Pulsar 將所有 topic 數(shù)據(jù)存儲在由 Apache BookKeeper 支持的專業(yè)數(shù)據(jù)層中。

存儲和消息傳遞的分離解決了擴展、重新平衡和維護集群的許多問題。它還提高了可靠性，幾乎不可能丟失數(shù)據(jù)。

• 另外，在讀取數(shù)據(jù)時可以直連 BookKeeper，且不影響實時攝取。例如，可以使用 Presto 對 topic 執(zhí)行 SQL 查詢，類似于 KSQL，但不會影響實時數(shù)據(jù)處理。
• 虛擬 topic：由于采用 n 層體系結(jié)構(gòu)，因此對 topic 的數(shù)量沒有限制，topic 及其存儲是分離的。用戶還可以創(chuàng)建非持久性 topic。
• N 層存儲：Kafka 的一個問題是，存儲費用可能變高。因此，它很少用于存儲"冷"數(shù)據(jù)，并且消息經(jīng)常被刪除，Apache Pulsar 可以借助分層存儲自動將舊數(shù)據(jù)卸載到 Amazon S3 或其他數(shù)據(jù)存儲系統(tǒng)，并且仍然向客戶端展示透明視圖;Pulsar 客戶端可以從時間開始節(jié)點讀取，就像所有消息都存在于日志中一樣。
• Pulsar Function：易于部署、輕量級計算過程、對開發(fā)人員友好的 API，無需運行自己的流處理引擎(如 Kafka)。
• 安全性：它具有內(nèi)置的代理、多租戶安全性、可插拔的身份驗證等特性。
• 快速重新平衡：分區(qū)被分為易于重新平衡的分片。
• 服務(wù)器端重復(fù)數(shù)據(jù)刪除和無效字段：無需在客戶端中執(zhí)行此操作，也可以在壓縮期間刪除重復(fù)數(shù)據(jù)。
• 內(nèi)置 Schema registry(架構(gòu)注冊表)：支持多種策略，易于操作。
• 地理復(fù)制和內(nèi)置 Discovery：易于將集群復(fù)制到多個區(qū)域。
• 集成的負載均衡器和 Prometheus 指標(biāo)。
• 多重集成：Kafka、RabbitMQ 等。
• 支持多種編程語言，例如 GoLang、Java、Scala、Node、Python…...
• 分片和數(shù)據(jù)分區(qū)在服務(wù)器端透明進行，客戶端不需要了解分片與分區(qū)數(shù)據(jù)。

 

Pulsar 特性列表[5]

Pulsar 入門

Pulsar 入門非常容易，使用前提是安裝 JDK。

①下載 Pulsar 并解壓縮(備注：目前 Apache Pulsar 最新版本為 2.7.0)：

$ wget https://archive.apache.org/dist/pulsar/pulsar-2.6.1/apache-pulsar-2.6.1-bin.tar.gz 

②下載連接器(可選)：

$ wget https://archive.apache.org/dist/pulsar/pulsar-2.6.1/connectors/{connector}-2.6.1.nar 

③下載 nar 文件后，將文件復(fù)制到 Pulsar 目錄中的 Connectors 目錄。

④啟動 Pulsar!

$ bin/pulsar standalone 

Pulsar 提供了一個稱為 Pulsar-Client 的 CLI 工具，我們可以使用它與集群進行交互。

生產(chǎn)消息：

$ bin/pulsar-client produce my-topic --messages "hello-pulsar" 

消費消息：

$ bin/pulsar-client consume my-topic -s "first-subscription" 

Akka 流示例

舉一個客戶端示例，我們在 Akka 上使用 Pulsar4s。

首先，我們需要創(chuàng)建一個 Source 來消費數(shù)據(jù)流，所需要的只是一個函數(shù)，該函數(shù)將按需創(chuàng)建消費者并查找消息 ID：

val topic = Topic("persistent://standalone/mytopic") 
val consumerFn = () => client.consumer(ConsumerConfig(topic, subscription)) 

然后，我們傳遞 ConsumerFn 函數(shù)來創(chuàng)建源：

import com.sksamuel.pulsar4s.akka.streams._ 
val pulsarSource = source(consumerFn, Some(MessageId.earliest)) 

Akka 源的物化值是 Control 的一個實例，該對象提供了一種"關(guān)閉"方法，可用于停止消費消息。現(xiàn)在，我們可以像往常一樣使用 Akka Streams 處理數(shù)據(jù)。

要創(chuàng)建一個接收器：

val topic = Topic("persistent://standalone/mytopic") 
val producerFn = () => client.producer(ProducerConfig(topic)) 
import com.sksamuel.pulsar4s.akka.streams._ 
val pulsarSink = sink(producerFn) 

完整示例摘自 Pulsar4s[6]：

object Example { 
  import com.sksamuel.pulsar4s.{ConsumerConfig, MessageId, ProducerConfig, PulsarClient, Subscription, Topic} 
  import org.apache.pulsar.client.api.Schema 
  implicit val system: ActorSystem = ActorSystem() 
  implicit val materializer: ActorMaterializer = ActorMaterializer() 
  implicit val schema: Schema[Array[Byte]] = Schema.BYTES 
  val client = PulsarClient("pulsar://localhost:6650") 
  val intopic = Topic("persistent://sample/standalone/ns1/in") 
  val outtopic = Topic("persistent://sample/standalone/ns1/out") 
  val consumerFn = () => client.consumer(ConsumerConfig(topics = Seq(intopic), subscriptionName = Subscription("mysub"))) 
  val producerFn = () => client.producer(ProducerConfig(outtopic)) 
  val control = source(consumerFn, Some(MessageId.earliest)) 
    .map { consumerMessage => ProducerMessage(consumerMessage.data) } 
    .to(sink(producerFn)).run() 
  Thread.sleep(10000) 
  control.stop() 
} 

Pulsar Function 示例

Pulsar Function 處理來自一個或多個 topic 的消息，對其進行轉(zhuǎn)換并將結(jié)果輸出到另一個 topic：

 

Pulsar Function[7]

可以在兩個接口之間進行選擇以編寫函數(shù)：

• 語言原生接口：不需要特定的 Pulsar 庫或特殊的依賴項;無法訪問上下文，僅支持 Java 和 Python。
• Pulsar Function SDK：可用于 Java/Python/ Go，并提供更多功能，比如訪問上下文對象。

只需編寫一個簡單的函數(shù)即可使用語言原生接口轉(zhuǎn)換消息：

def process(input): 
return "{}!".format(input) 

用 Python 編寫的這個簡單函數(shù)只是向所有傳入的字符串添加一個感嘆號，并將結(jié)果字符串發(fā)布到 topic。

使用 SDK 需要導(dǎo)入依賴項，例如在 Go 中，我們可以編寫：

package main 
import ( 
"context" 
"fmt" 
"github.com/apache/pulsar/pulsar-function-go/pf" 
) 
func HandleRequest(ctx context.Context, in []byte) error { 
fmt.Println(string(in) + "!") 
return nil 
} 
func main() { 
pf.Start(HandleRequest) 
} 

如果要發(fā)布無服務(wù)器功能并將其部署到集群，可以使用 Pulsar-Admin CL;如果使用 Python，我們可以編寫：

$ bin/pulsar-admin functions create \ 
--py ~/router.py \ 
--classname router.RoutingFunction \ 
--tenant public \ 
--namespace default \ 
--name route-fruit-veg \ 
--inputs persistent://public/default/basket-items 
Pulsar Function 的一個重要功能是用戶可以在發(fā)布該函數(shù)時設(shè)置交付保證： 
$ bin/pulsar-admin functions create \ 
--name my-effectively-once-function \ 
--processing-guarantees EFFECTIVELY_ONCE 

有以下選擇：

 

Pulsar 的優(yōu)勢

與 Kafka 相比，讓我們回顧下 Pulsar 的主要優(yōu)勢：

• 更多功能：Pulsar Function、多租戶、Schema registry、n 層存儲、多種消費模式和持久性模式等。
• 更大的靈活性：3 種訂閱類型(獨占，共享和故障轉(zhuǎn)移)，用戶可以在一個訂閱上管理多個 topic。
• 持久性選項：非持久(快速)、持久、壓縮(每個消息僅最后一個鍵)，用戶可以選擇交付保證。Pulsar 具有服務(wù)器端重復(fù)數(shù)據(jù)刪除和無效字樣多保留政策和 TTL 的特性。
• 無需提前定義擴展需求。
• 支持隊列與流兩種消息消費模型，所以 Pulsar 既可以代替 RabbitMQ 也可以代替 Kafka。
• 存儲與 broker 分離，因此擴展性更好，重新平衡更快、更可靠。
• 易于操作運維：架構(gòu)解耦和 n 層存儲。
• 與 Presto 的 SQL 集成，可直接查詢存儲而不會影響 broker。
• 借助 n 層自動存儲選項，可以更低成本地存儲。
• 更快：基準(zhǔn)測試[8]在各種情況下都表現(xiàn)出更好的性能。Pulsar 具有較低的延遲和更好的擴展功能。
• Pulsar Function 支持無服務(wù)器計算，無需部署管理。
• 集成 Schema registry。
• 集成的負載平衡器和 Prometheus 指標(biāo)。
• 地理復(fù)制效果更好，更易于設(shè)置。Pulsar 內(nèi)置 Discover-ability。
• 創(chuàng)建 topic 數(shù)量沒有限制。
• 與 Kafka 兼容，易于集成。

Pulsar 的劣勢

Pulsar 并不完美，Pulsar 也存在一些問題：

• 相對缺乏支持、文檔和案例。
• n 層體系結(jié)構(gòu)導(dǎo)致需要更多組件：BookKeeper。
• 插件和客戶端相對 Kafka 較少。
• 云中的支持較少，Confluent 具有托管云產(chǎn)品。

不過，上面的情況都在得到快速改善，目前 Pulsar 也逐漸被越來越多的公司和組織使用。

Apache Pulsar 商業(yè)支持公司 StreamNative 也推出了 StreamNative Cloud，Apache Pulsar 正在快速成長，我們都可以看到令人欣喜的變化。

Confluent 曾發(fā)布博客對比 Pulsar 和 Kafka ，但請注意，這些問題可能有偏見。

Pulsar 使用場景

Pulsar 可用于廣泛的場景：

• 發(fā)布/訂閱隊列消息傳遞。
• 分布式日志。
• 事件溯源，用于永久性事件存儲。
• 微服務(wù)。
• SQL 分析。
• Serverless 功能。

什么時候應(yīng)該考慮 Pulsar?

• 同時需要像 RabbitMQ 這樣的隊列和 Kafka 這樣的流處理程序。
• 需要易用的地理復(fù)制。
• 實現(xiàn)多租戶，并確保每個團隊的訪問權(quán)限。
• 需要長時間保留消息，并且不想將其卸載到另一個存儲中。
• 需要高性能，基準(zhǔn)測試表明 Pulsar 提供了更低的延遲和更高的吞吐量。

如果在云端，請注意考慮基于云的解決方案。云提供商擁有涵蓋某些場景的不同服務(wù)。

例如，對于隊列消息，云提供商提供了許多服務(wù)，比如 Google pub / sub;對于分布式日志，有 Confluent 云或 AWS Kinesis;StreamNative 也提供了基于 Pulsar 的云端服務(wù)。

云提供商還提供了非常好的安全性。Pulsar 的優(yōu)勢在于可以在一個平臺上提供許多功能。

一些團隊可能將其用作微服務(wù)的消息傳遞系統(tǒng)，而另一些團隊則將其用作數(shù)據(jù)處理的分布式日志。

結(jié)論

我是 Kafka 的忠實粉絲，我對 Pulsar 如此感興趣的原因是：競爭驅(qū)動創(chuàng)新。

Kafka 是一種成熟，富有彈性且經(jīng)過考驗的產(chǎn)品，在世界范圍內(nèi)獲得了巨大成功，無法想象大多數(shù)公司沒有它會怎樣。

但是我確實看到 Kafka 成為其自身成功的受害者，由于需要支持許多大型公司導(dǎo)致巨大的增長減慢了功能開發(fā)的速度、移除 ZooKeeper 依賴項等重要功能花費的時間太長，這為諸如 Pulsar 等工具蓬勃發(fā)展創(chuàng)造了空間。

Pulsar 雖然年輕卻勢頭很猛，在將 Pulsar 納入組織之前，需進行分析、基準(zhǔn)測試、研究并進行 POC。

從小處著手，在將 Kafka 遷移到 Pulsar 之前進行概念驗證，并在決定進行完全遷移之前評估影響。

引用鏈接：

• [1] 《Pulsar Advantages Over Kafka》：

https://itnext.io/pulsar-advantages-over-kafka-7e0c2affe2d6

• [2] Kafka 演示：

https://talks.rmoff.net/pZC6Za/slides

• [3] MirrorMaker：

https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=27846330

• [4] Pulsar 架構(gòu)圖：

https://pulsar.apache.org/docs/en/concepts-architecture-overview/

• [5] Pulsar 特性列表：

https://pulsar.apache.org/

• [6] Pulsar4s：

https://github.com/sksamuel/pulsar4s/blob/master/pulsar4s-akka-streams/src/test/scala/com/sksamuel/pulsar4s/akka/streams/Example.scala

• [7] Pulsar Function：

https://pulsar.apache.org/docs/en/functions-overview/

• [8] 基準(zhǔn)測試：

https://medium.com/swlh/performance-comparison-between-apache-pulsar-and-kafka-latency-79fb0367f407

作者：聞數(shù)起舞

編輯：陶家龍

出處：轉(zhuǎn)載自 Java 高級架構(gòu)，原中文版本由聞數(shù)起舞翻譯自 Lewis Fairweather 的文章《Pulsar Advantages Over Kafka》[1]，文章轉(zhuǎn)載時有改動。