Kafka-on-Pulsar 的開發歷程是怎樣的

這期內容當中小編將會給大家帶來有關Kafka-on-Pulsar 的開發歷程是怎樣的，文章內容豐富且以專業的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

Protocol Handler 是在 Pulsar 2.5.0 版本后加入的新機制，希望開發者們能利用 Pulsar 已有的基礎架構，把 Pulsar 當作一個可靠、高效、穩定的流數據存儲，可以利用它去開發一些可插拔消息協議。

所以 Kafka-on-Pulsar 是基于 Protocol Handler 進行開發的，支持 Kafka 2.0 協議的插件。只需下載 KoP 插件并安裝到已有的 Pulsar broker 里，就能在 Pulsar 里支持 Kafka 協議。優點是簡化了從 Kafka 遷移到 Pulsar 的流程，不需要更改兩遍代碼，直接無縫遷移。接下來就一起詳細看看KoP 的開發歷程吧。

什么是 Apache Pulsar

Apache Pulsar 是一個事件流平臺。最初，Apache Pulsar 就采用云原生、分層分片的架構。該架構將服務和存儲分離開來，使系統實現更友好的容器化。

而現在，Pulsar 不僅僅是是一個消息中間件，更是一個消息+流數據結合的系統，即 Cloud-Native Event Streaming。

我們之前寫過很多關于 Pulsar 的具體詳情，感興趣的可以查看 ：Apache Pulsar 介紹。

 Why KoP？

Plusar 為隊列和流工作負載提供統一的消息模型。Pulsar 支持自己基于 protobuf 的二進制協議，以確保高性能和低延遲。protobuf 有利于實現 Pulsar 客戶端。

而且，該項目也支持 Java，Go，Python 和 C ++ 語言以及社區提供的第三方客戶端。但是，對于使用其他消息傳輸協議編寫的應用程序，用戶必須重寫這些應用程序，否則這些應用程序無法采用 Pulsar 新的統一消息傳輸協議。

為了解決這一問題，Pulsar 社區之前也開發了一些應用程序，以便將 Kafka 應用程序從其他消息系統遷移到 Pulsar。例如，Pulsar 在 Kafka Java API 上提供了 Kafka wrapper。

Kafka wrapper 允許用戶在不改變代碼的情況下將其使用的 Kafka Java 客戶端應用程序從 Kafka 切換到 Pulsar。Pulsar 還提供豐富的 connector 生態系統，用于連接 Pulsar 和其他數據系統。

但是，那些想要從其他 Kafka 應用程序切換到 Pulsar 的用戶仍然有強烈的需求。

因此，就產生了“在 Pulsar 上支持 Kafka 協議”的想法。最初的猜想是添加一個 proxy，比如好多公司會在 Kafka 之前加一個類似 HTTP proxy，后續再轉換成 Pulsar 協議。

第二種猜想是能否直接將 Kafka 協議直接接入到 Pulsar broker 里，也就是目前 KoP 的成型。

那么關于第一種 proxy 做法，如果實現起來大概是什么樣呢？OVHcloud 就有過一次嘗試。

之前 OVHcloud 一直采用 Apache Kafka。盡管他們有在 Kafka 上運行多個集群且每秒處理數百萬條消息的經驗，但仍面臨艱巨的運營挑戰。所以，OVHcloud 放棄 Kafka，決定將其服務的產品轉移到 Pulsar，并在 Pulsar 上構建其產品。

但是為了照顧到依舊使用 Kafka 系統的用戶，所以他們想在 Pulsar 里添加一個 proxy 去支持 Kafka 協議。他們最初的做法就是將 Kafka 協議的一幀轉換成 Pulsar 協議。

Proxy 收到來自 Kafka 客戶端的任何一幀，通過自由狀態機將其轉換為 Pulsar 相應的接口。

這個狀態機一種是用于接收 Kafka 請求，第二種是用于處理 Pulsar response。然后在其中間再添加一個狀態機進行同步。

因為在 TCP 層進行這些操作，所以它的表現還是不錯的。借由 Rust 的特性，整體運行流暢。但是這個情況下，代碼仍需要一行行去寫，同時 Kafka 協議里有一些是沒有辦法通過 proxy 方式實現。比如：group coordinator 和 offsets management。

還有一個比較關鍵的點是，因為用 Rust 去構寫，所以比較難開源。即便是開源出來也很難作為一個組件去插入到 Pulsar 系統中。

剛好去年 StreamNative 的一條推特引起了 OVHcloud 的注意。這是 StreamNative 第一次舉行線下 Pulsar meetup 時翟佳老師分享的 KoP demo。

經過幾次雙方經驗互談交流后，雙方合力推出了更完善的「KoP」。利用 Pulsar 和 BookKeeper 的事件流存儲架構和 Pulsar 的可插拔協議處理插件框架來提供一種精簡而全面的解決方案。

所以當我們倒回去重看 Pulsar 架構，下方模塊圖中最核心的：Broker、BookKeeper、ZooKeeper。Pulsar 就是基于 Managed ledger 實現的一套分布式流式存儲，包括如何存數據、如何防止數據丟失、流如何從本地機房復制到另一機房等。

Pulsar 協議本身是一個很輕量級的東西，即上圖中的 Pulsar protocol handler。它主要是處理 TCP 過來的請求格式，然后將請求轉化和讀取的操作。所以 Pulsar 協議最核心部分在存儲層面、分布式均衡層面等。

將 Pulsar protocol handler 抽象出來，變成一個框架/接口。利用這個框架，可以直接訪問 Pulsar 已經構建好的存儲系統，剩下要做的只是協議的解析和轉換。

所以依據這個構想，將 Kafka 協議帶入去實踐。在  Pulsar 2.5 版本時新加了一個「Pluggable protocol handler」的概念（PIP-41），將接口單獨抽離了出來。

Pulsar protocol handler 的使用是類似 Pulsar function/connector，只需將其插入到 Pulsar broker 中，就可以讓 Pulsar 具有讀取和解析其他協議的能力。這個機制只需要調整兩個配置：

配置完成后，重啟集群即可支持「其他類型協議」的處理能力。當然這個特性只在 Pulsar 2.5 版本后才支持，所以如需嘗試，可以先將 Pulsar 系統升級到 2.5 版本。

所以在此機制下過程就會變得更加明了簡單。只需在 Pulsar 里實現 Kafka protocol handler 即可，剩下的上圖實線綠色部分是 Kafka 原生客戶端。只需將數據接入到 Pulsar 集群，就可以處理 Kafka 請求。

應為 Pulsar 和 Kafka 在一些層面有很多相似之處。比如日志層，Pulsar 和 Kafka 都采用非常相似的數據模型，用于發布/訂閱消息和事件流，Pulsar 和 Kafka 都采用分布式日志。

通過對比 Pulsar 和 Kafka，我們發現這兩種系統有很多相似之處。這兩種系統都包括以下操作：

• Topic 查找
  所有客戶端都連接到任一 broker 以查找 Topic 的元數據（即 owner broker）。獲取元數據之后，客戶端與 owner broker 建立持久的 TCP 連接。
• 發布
  客戶端與 Topic 區的 owner broker 進行對話，以將消息追加到分布式日志中。
• 消費
  客戶端與 Topic 分區的 owner broker 進行對話，以便從分布式日志中讀取消息。
• 偏移量
  為發布給 Topic 分區的消息分配偏移量。在 Pulsar 中，偏移量被稱為 MessageId。consumer 可以使用偏移量來查找日志中的給定位置，以便讀取消息。

• 消費狀態
  這兩個系統都維護訂閱中的 consumer（ Kafka 稱之為消費組）的消費狀態。Kafka 將消費狀態存儲在 `__offsets` Topic，而 Pulsar 將消費狀態存儲在 `cursors`。

1. Topic

2. 消息 ID 和偏移量

3. 消息

4. Topic 查找

5. 發布消息

6. 消費消息

7. Group coordinator & 偏移量管理

• 多租戶
• 安全性
• 跨機房復制
• 分層存儲
• Schema
• 與已有的數據環境（如 Flink、Spark、Presto）集成

目前在做一個項目：Pulsar Registry，類似于 DocHub。也可以看作一個應用商店，會集中一些組件/插件合集，可以期待一下。

不能，0.10/0.11 版本以上才可以平滑升級。

KoP 最終的目的，是方便用戶將 Kafka 上已有的應用遷移到 Pulsar 上，同時通過 KoP 的方式讓用戶可以更方便地構建產品。未來 KoP 也會加大對 schema 和 Kafka 版本的支持與多兼容性。

上述就是小編為大家分享的Kafka-on-Pulsar 的開發歷程是怎樣的了，如果剛好有類似的疑惑，不妨參照上述分析進行理解。如果想知道更多相關知識，歡迎關注億速云行業資訊頻道。