Kafka-on-Pulsar 的開發(fā)歷程是怎樣的

這期內(nèi)容當(dāng)中小編將會給大家?guī)碛嘘P(guān)Kafka-on-Pulsar 的開發(fā)歷程是怎樣的，文章內(nèi)容豐富且以專業(yè)的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

Protocol Handler 是在 Pulsar 2.5.0 版本后加入的新機(jī)制，希望開發(fā)者們能利用 Pulsar 已有的基礎(chǔ)架構(gòu)，把 Pulsar 當(dāng)作一個(gè)可靠、高效、穩(wěn)定的流數(shù)據(jù)存儲，可以利用它去開發(fā)一些可插拔消息協(xié)議。

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

什么是 Apache Pulsar

Apache Pulsar 是一個(gè)事件流平臺。最初，Apache Pulsar 就采用云原生、分層分片的架構(gòu)。該架構(gòu)將服務(wù)和存儲分離開來，使系統(tǒng)實(shí)現(xiàn)更友好的容器化。

而現(xiàn)在，Pulsar 不僅僅是是一個(gè)消息中間件，更是一個(gè)消息+流數(shù)據(jù)結(jié)合的系統(tǒng)，即 Cloud-Native Event Streaming。

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

 Why KoP？

Plusar 為隊(duì)列和流工作負(fù)載提供統(tǒng)一的消息模型。Pulsar 支持自己基于 protobuf 的二進(jìn)制協(xié)議，以確保高性能和低延遲。protobuf 有利于實(shí)現(xiàn) Pulsar 客戶端。

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

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

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

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

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

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

那么關(guān)于第一種 proxy 做法，如果實(shí)現(xiàn)起來大概是什么樣呢？OVHcloud 就有過一次嘗試。

之前 OVHcloud 一直采用 Apache Kafka。盡管他們有在 Kafka 上運(yùn)行多個(gè)集群且每秒處理數(shù)百萬條消息的經(jīng)驗(yàn)，但仍面臨艱巨的運(yùn)營挑戰(zhàn)。所以，OVHcloud 放棄 Kafka，決定將其服務(wù)的產(chǎn)品轉(zhuǎn)移到 Pulsar，并在 Pulsar 上構(gòu)建其產(chǎn)品。

但是為了照顧到依舊使用 Kafka 系統(tǒng)的用戶，所以他們想在 Pulsar 里添加一個(gè) proxy 去支持 Kafka 協(xié)議。他們最初的做法就是將 Kafka 協(xié)議的一幀轉(zhuǎn)換成 Pulsar 協(xié)議。

Proxy 收到來自 Kafka 客戶端的任何一幀，通過自由狀態(tài)機(jī)將其轉(zhuǎn)換為 Pulsar 相應(yīng)的接口。

這個(gè)狀態(tài)機(jī)一種是用于接收 Kafka 請求，第二種是用于處理 Pulsar response。然后在其中間再添加一個(gè)狀態(tài)機(jī)進(jìn)行同步。

因?yàn)樵?TCP 層進(jìn)行這些操作，所以它的表現(xiàn)還是不錯(cuò)的。借由 Rust 的特性，整體運(yùn)行流暢。但是這個(gè)情況下，代碼仍需要一行行去寫，同時(shí) Kafka 協(xié)議里有一些是沒有辦法通過 proxy 方式實(shí)現(xiàn)。比如：group coordinator 和 offsets management。

還有一個(gè)比較關(guān)鍵的點(diǎn)是，因?yàn)橛?Rust 去構(gòu)寫，所以比較難開源。即便是開源出來也很難作為一個(gè)組件去插入到 Pulsar 系統(tǒng)中。

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

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

所以當(dāng)我們倒回去重看 Pulsar 架構(gòu)，下方模塊圖中最核心的：Broker、BookKeeper、ZooKeeper。Pulsar 就是基于 Managed ledger 實(shí)現(xiàn)的一套分布式流式存儲，包括如何存數(shù)據(jù)、如何防止數(shù)據(jù)丟失、流如何從本地機(jī)房復(fù)制到另一機(jī)房等。

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

將 Pulsar protocol handler 抽象出來，變成一個(gè)框架/接口。利用這個(gè)框架，可以直接訪問 Pulsar 已經(jīng)構(gòu)建好的存儲系統(tǒng)，剩下要做的只是協(xié)議的解析和轉(zhuǎn)換。

所以依據(jù)這個(gè)構(gòu)想，將 Kafka 協(xié)議帶入去實(shí)踐。在  Pulsar 2.5 版本時(shí)新加了一個(gè)「Pluggable protocol handler」的概念（PIP-41），將接口單獨(dú)抽離了出來。

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

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

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

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

通過對比 Pulsar 和 Kafka，我們發(fā)現(xiàn)這兩種系統(tǒng)有很多相似之處。這兩種系統(tǒng)都包括以下操作：

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

• 消費(fèi)狀態(tài)
  這兩個(gè)系統(tǒng)都維護(hù)訂閱中的 consumer（ Kafka 稱之為消費(fèi)組）的消費(fèi)狀態(tài)。Kafka 將消費(fèi)狀態(tài)存儲在 `__offsets` Topic，而 Pulsar 將消費(fèi)狀態(tài)存儲在 `cursors`。

1. Topic

2. 消息 ID 和偏移量

3. 消息

4. Topic 查找

5. 發(fā)布消息

6. 消費(fèi)消息

7. Group coordinator & 偏移量管理

• 多租戶
• 安全性
• 跨機(jī)房復(fù)制
• 分層存儲
• Schema
• 與已有的數(shù)據(jù)環(huán)境（如 Flink、Spark、Presto）集成

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

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

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

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