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

成都創(chuàng)新互聯(lián)專(zhuān)注于鳳泉企業(yè)網(wǎng)站建設(shè),響應(yīng)式網(wǎng)站建設(shè),成都做商城網(wǎng)站。鳳泉網(wǎng)站建設(shè)公司,為鳳泉等地區(qū)提供建站服務(wù)。全流程按需設(shè)計(jì)網(wǎng)站，專(zhuān)業(yè)設(shè)計(jì)，全程項(xiàng)目跟蹤，成都創(chuàng)新互聯(lián)專(zhuān)業(yè)和態(tài)度為您提供的服務(wù)

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

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

什么是 Apache Pulsar

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

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

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

 Why KoP？

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

1. Topic

2. 消息 ID 和偏移量

3. 消息

4. Topic 查找

5. 發(fā)布消息

6. 消費(fèi)消息

7. Group coordinator & 偏移量管理

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

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

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

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

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