我為什么放棄Kafka，選擇Pulsar？
最近，我一直在研究 Pulsar 及其與 Kafka 的比較。通過快速搜索，你會看到這兩個最著名的開源消息傳遞系統之間正在進行的"戰爭"。
圖片來自 Pexels
作為 Kafka 的用戶，我著實對 Kafka 的某些問題感到困惑，但 Pulsar 卻讓人眼前一亮、令我非常興奮。所以最后，我設法花了一些時間了解背景資料，并且做了很多研究。
在本文中，我將重點介紹 Pulsar 的優勢，并說明 Pulsar 勝于 Kafka 的理由。讓我們開始!
Kafka 基礎知識
Kafka 是消息傳遞系統之王。它由 LinkedIn 于 2011 年創建，并在 Confluent 的支持下得到了廣泛的傳播。
Confluent 已向開源社區發布了許多新功能和附加組件，例如用于模式演化的 Schema Registry，用于從其他數據源輕松流式傳輸的 Kafka Connect 等。
數據庫到 Kafka，Kafka Streams 進行分布式流處理，最近使用 KSQL 對 Kafka topic 執行類似 SQL 的查詢等等。
Kafka 快速，易于安裝，非常受歡迎，可用于廣泛的范圍或用例。從開發人員的角度來看，盡管 Apache Kafka 一直很友好，但在操作運維方面卻是一團糟。
因此，讓我們回顧一下 Kafka 的一些痛點：
Kafka 演示[2]
Kakfa的諸多痛點如下：
- 擴展 Kafka 十分棘手，這是由于 broker 與存儲數據的耦合架構結構所致。剝離一個 broker 意味著它必須復制 topic 分區和副本，這非常耗時。
- 沒有與租戶完全隔離的本地多租戶。
- 存儲會變得非常昂貴，盡管可以長時間存儲數據，但是由于成本問題卻很少用到它。
- 萬一副本不同步，有可能丟失消息。
- 必須提前計劃和計算 broker、topic、分區和副本的數量(確保計劃的未來使用量增長)，以避免擴展問題，這非常困難。
- 如果僅需要消息傳遞系統，則使用偏移量可能會很復雜。
- 集群重新平衡會影響相連的生產者和消費者的性能。
- MirrorMaker[3] Geo 復制機制存在問題。像 Uber 這樣的公司已經創建了自己的解決方案來克服這些問題。
如您所見，大多數問題與操作運維方面有關。盡管安裝起來相對容易，但 Kafka 難以管理和調優。而且，它也缺乏應有的靈活和彈性。
Pulsar 基礎知識
Pulsar 由 Yahoo!在 2013 年創建，并于 2016 年捐贈給 Apache 基金會。Pulsar 現在是 Apache 軟件基金會的頂級項目。
Yahoo!、Verizon、Twitter 等公司已在生產中使用它來處理成千上萬消息。它具有運行成本低、靈活等特性。Pulsar 旨在解決 Kafka 的大部分難題，使其更易于擴展。
Pulsar 非常靈活：它既可以應用于像 Kafka 這樣的分布式日志應用場景，也可以應用于像 RabbitMQ 這樣的純消息傳遞系統場景。
它支持多種類型的訂閱、多種交付保證、保留策略以及處理模式演變的方法，以及其他諸多特性。
Pulsar 架構圖[4]
Pulsar 的特性如下：
- 內置多租戶，不同的團隊可以使用相同的集群并將其隔離，解決了許多管理難題。它支持隔離、身份驗證、授權和配額。
- 多層體系結構：Pulsar 將所有 topic 數據存儲在由 Apache BookKeeper 支持的專業數據層中。
存儲和消息傳遞的分離解決了擴展、重新平衡和維護集群的許多問題。它還提高了可靠性，幾乎不可能丟失數據。
- 另外，在讀取數據時可以直連 BookKeeper，且不影響實時攝取。例如，可以使用 Presto 對 topic 執行 SQL 查詢，類似于 KSQL，但不會影響實時數據處理。
- 虛擬 topic：由于采用 n 層體系結構，因此對 topic 的數量沒有限制，topic 及其存儲是分離的。用戶還可以創建非持久性 topic。
- N 層存儲：Kafka 的一個問題是，存儲費用可能變高。因此，它很少用于存儲"冷"數據，并且消息經常被刪除，Apache Pulsar 可以借助分層存儲自動將舊數據卸載到 Amazon S3 或其他數據存儲系統，并且仍然向客戶端展示透明視圖;Pulsar 客戶端可以從時間開始節點讀取，就像所有消息都存在于日志中一樣。
- Pulsar Function：易于部署、輕量級計算過程、對開發人員友好的 API，無需運行自己的流處理引擎(如 Kafka)。
- 安全性：它具有內置的代理、多租戶安全性、可插拔的身份驗證等特性。
- 快速重新平衡：分區被分為易于重新平衡的分片。
- 服務器端重復數據刪除和無效字段：無需在客戶端中執行此操作，也可以在壓縮期間刪除重復數據。
- 內置 Schema registry(架構注冊表)：支持多種策略，易于操作。
- 地理復制和內置 Discovery：易于將集群復制到多個區域。
- 集成的負載均衡器和 Prometheus 指標。
- 多重集成：Kafka、RabbitMQ 等。
- 支持多種編程語言，例如 GoLang、Java、Scala、Node、Python…...
- 分片和數據分區在服務器端透明進行，客戶端不需要了解分片與分區數據。
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 文件后，將文件復制到 Pulsar 目錄中的 Connectors 目錄。
④啟動 Pulsar!
- $ bin/pulsar standalone
Pulsar 提供了一個稱為 Pulsar-Client 的 CLI 工具，我們可以使用它與集群進行交互。
生產消息：
- $ bin/pulsar-client produce my-topic --messages "hello-pulsar"
消費消息：
- $ bin/pulsar-client consume my-topic -s "first-subscription"
Akka 流示例
舉一個客戶端示例，我們在 Akka 上使用 Pulsar4s。
首先，我們需要創建一個 Source 來消費數據流，所需要的只是一個函數，該函數將按需創建消費者并查找消息 ID：
- val topic = Topic("persistent://standalone/mytopic")
- val consumerFn = () => client.consumer(ConsumerConfig(topic, subscription))
然后，我們傳遞 ConsumerFn 函數來創建源：
- import com.sksamuel.pulsar4s.akka.streams._
- val pulsarSource = source(consumerFn, Some(MessageId.earliest))
Akka 源的物化值是 Control 的一個實例，該對象提供了一種"關閉"方法，可用于停止消費消息。現在，我們可以像往常一樣使用 Akka Streams 處理數據。
要創建一個接收器：
- 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 的消息，對其進行轉換并將結果輸出到另一個 topic：
Pulsar Function[7]
可以在兩個接口之間進行選擇以編寫函數：
- 語言原生接口：不需要特定的 Pulsar 庫或特殊的依賴項;無法訪問上下文，僅支持 Java 和 Python。
- Pulsar Function SDK：可用于 Java/Python/ Go，并提供更多功能，比如訪問上下文對象。
只需編寫一個簡單的函數即可使用語言原生接口轉換消息：
- def process(input):
- return "{}!".format(input)
用 Python 編寫的這個簡單函數只是向所有傳入的字符串添加一個感嘆號，并將結果字符串發布到 topic。
使用 SDK 需要導入依賴項，例如在 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)
- }
如果要發布無服務器功能并將其部署到集群，可以使用 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 的一個重要功能是用戶可以在發布該函數時設置交付保證：
- $ bin/pulsar-admin functions create \
- --name my-effectively-once-function \
- --processing-guarantees EFFECTIVELY_ONCE
有以下選擇：
Pulsar 的優勢
與 Kafka 相比，讓我們回顧下 Pulsar 的主要優勢：
- 更多功能：Pulsar Function、多租戶、Schema registry、n 層存儲、多種消費模式和持久性模式等。
- 更大的靈活性：3 種訂閱類型(獨占，共享和故障轉移)，用戶可以在一個訂閱上管理多個 topic。
- 持久性選項：非持久(快速)、持久、壓縮(每個消息僅最后一個鍵)，用戶可以選擇交付保證。Pulsar 具有服務器端重復數據刪除和無效字樣多保留政策和 TTL 的特性。
- 無需提前定義擴展需求。
- 支持隊列與流兩種消息消費模型，所以 Pulsar 既可以代替 RabbitMQ 也可以代替 Kafka。
- 存儲與 broker 分離，因此擴展性更好，重新平衡更快、更可靠。
- 易于操作運維：架構解耦和 n 層存儲。
- 與 Presto 的 SQL 集成，可直接查詢存儲而不會影響 broker。
- 借助 n 層自動存儲選項，可以更低成本地存儲。
- 更快：基準測試[8]在各種情況下都表現出更好的性能。Pulsar 具有較低的延遲和更好的擴展功能。
- Pulsar Function 支持無服務器計算，無需部署管理。
- 集成 Schema registry。
- 集成的負載平衡器和 Prometheus 指標。
- 地理復制效果更好，更易于設置。Pulsar 內置 Discover-ability。
- 創建 topic 數量沒有限制。
- 與 Kafka 兼容，易于集成。
Pulsar 的劣勢
Pulsar 并不完美，Pulsar 也存在一些問題：
- 相對缺乏支持、文檔和案例。
- n 層體系結構導致需要更多組件：BookKeeper。
- 插件和客戶端相對 Kafka 較少。
- 云中的支持較少，Confluent 具有托管云產品。
不過，上面的情況都在得到快速改善，目前 Pulsar 也逐漸被越來越多的公司和組織使用。
Apache Pulsar 商業支持公司 StreamNative 也推出了 StreamNative Cloud，Apache Pulsar 正在快速成長，我們都可以看到令人欣喜的變化。
Confluent 曾發布博客對比 Pulsar 和 Kafka ，但請注意，這些問題可能有偏見。
Pulsar 使用場景
Pulsar 可用于廣泛的場景：
- 發布/訂閱隊列消息傳遞。
- 分布式日志。
- 事件溯源，用于永久性事件存儲。
- 微服務。
- SQL 分析。
- Serverless 功能。
什么時候應該考慮 Pulsar?
- 同時需要像 RabbitMQ 這樣的隊列和 Kafka 這樣的流處理程序。
- 需要易用的地理復制。
- 實現多租戶，并確保每個團隊的訪問權限。
- 需要長時間保留消息，并且不想將其卸載到另一個存儲中。
- 需要高性能，基準測試表明 Pulsar 提供了更低的延遲和更高的吞吐量。
如果在云端，請注意考慮基于云的解決方案。云提供商擁有涵蓋某些場景的不同服務。
例如，對于隊列消息，云提供商提供了許多服務，比如 Google pub / sub;對于分布式日志，有 Confluent 云或 AWS Kinesis;StreamNative 也提供了基于 Pulsar 的云端服務。
云提供商還提供了非常好的安全性。Pulsar 的優勢在于可以在一個平臺上提供許多功能。
一些團隊可能將其用作微服務的消息傳遞系統，而另一些團隊則將其用作數據處理的分布式日志。
結論
我是 Kafka 的忠實粉絲，我對 Pulsar 如此感興趣的原因是：競爭驅動創新。
Kafka 是一種成熟，富有彈性且經過考驗的產品，在世界范圍內獲得了巨大成功，無法想象大多數公司沒有它會怎樣。
但是我確實看到 Kafka 成為其自身成功的受害者，由于需要支持許多大型公司導致巨大的增長減慢了功能開發的速度、移除 ZooKeeper 依賴項等重要功能花費的時間太長，這為諸如 Pulsar 等工具蓬勃發展創造了空間。
Pulsar 雖然年輕卻勢頭很猛，在將 Pulsar 納入組織之前，需進行分析、基準測試、研究并進行 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 架構圖：
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] 基準測試：
https://medium.com/swlh/performance-comparison-between-apache-pulsar-and-kafka-latency-79fb0367f407
作者：聞數起舞
編輯：陶家龍
出處：轉載自 Java 高級架構，原中文版本由聞數起舞翻譯自 Lewis Fairweather 的文章《Pulsar Advantages Over Kafka》[1]，文章轉載時有改動。
