作者 | 冉小龍，劉昱

RocketMQ 用戶可以無縫遷移到 Apache Pulsar 了。自此，Apache Pulsar 補齊了兼容主流消息隊列協議的能力。

我們很高興地宣布騰訊云中間件開源 RoP！RoP 將 RocketMQ 協議處理插件引入 Pulsar broker，這樣 Pulsar 就能支持原生 RocketMQ 協議了。

?

什么是 RoP？

與 KoP、MoP 和 AoP 相似，RoP 是一種可插拔的協議處理插件。

將 RoP 協議處理插件添加到現有 Pulsar 集群后，用戶無需修改代碼，便能將現有的 RocketMQ 應用程序和服務遷移到 Pulsar，同時還能使用 Pulsar 的強大功能，例如：

• 計算與存儲分離

• 多租戶

• 跨地域復制

• 分層分片

• 輕量化計算框架 -- Pulsar Functions

• ...

?

為什么開發 RoP？

Apache Pulsar 是下一代云原生分布式消息流平臺，集消息、存儲、輕量化函數式計算為一體。自 2016 年開源以來，Pulsar 已被廣泛采用，并于 2018 年被指定為 Apache 頂級項目。

RocketMQ 是一款強大的開源分布式消息系統，基于高可用分布式集群技術，提供低延時的、高可靠的消息發布與訂閱服務。

Pulsar 和 RocketMQ 擁有廣泛的用戶群體和強勁的開發支持，全球許多頭部公司都在使用這兩種消息服務。同時，我們也收到了用戶的需求，希望能在 Pulsar 與 RocketMQ 之間傳輸數據，并充分利用這兩種消息系統的優勢。

Apache Pulsar 通過對 Consumer 層的抽象，提供了隊列和流兩種消費模型的統一抽象。在 Client 與 Broker 的交互中，Pulsar 基于 Protobuf 的二進制協議，提供更高的性能和更低的延遲。除此之外，通過 Protobuf 協議，Pulsar 可以更容易地支持并實現多語言的客戶端，比如：Java、CPP、Python 和 Go 語言等客戶端。

但是，對于使用其他消息傳輸協議編寫的應用程序（例如，RocketMQ），由于使用的消息處理協議和 Pulsar 不同，如果 Pulsar 想要兼容 RocketMQ 協議，為了將 RocketMQ 的協議適配到 Pulsar 的消息協議層中，用戶需要重寫整個協議層，這給用戶的遷移和切換帶來了很大的成本。

為了解決這個問題，最直觀的處理方式是使用類似 Pulsar Connector 的形式，將用戶在 RocketMQ 中的現存數據通過 RocketMQ Wrapper 的方式導入到 Pulsar 集群，但是這需要業務端更改自己的業務代碼邏輯，同時需要確保兩邊的數據能夠保證一致，這給使用 RocketMQ 的用戶帶來了很大的技術挑戰。所以，能否給用戶提供一個開箱即用的遷移策略和方案并且用戶無需做任何代碼修改呢？這便是 RoP 誕生的最初目的。

?

怎樣開發 RoP？

Apache Pulsar 在 PIP-41(https://github.com/apache/pulsar/wiki/PIP-41%3A-Pluggable-Protocol-Handler) 中介紹了一種全新的接入方式。通過在 Broker 端暴露 Protocol Handler 插件，將 Netty 的 channel 和 Pulsar 的 Broker Service(https://github.com/apache/pulsar/blob/907fcb5ba8/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java) 對象暴露給用戶。這允許用戶直接操作和調用 Pulsar 中比較低階的 API（例如：PersistentTopic 和 ManagerLedger）。基于這個協議，用戶無需更改代碼，只需將服務請求轉發到 RoP 中，RoP 利用 Protocol Handler 的插件將用戶的請求轉發到 Pulsar 中即可。

?

RoP 架構

通過對比 Pulsar 和 RocketMQ 之間的協議可以發現，二者在消息處理的思路上有不少相似之處，比如這兩種協議都包含如下操作：

• Topic Lookup: 所有 Clients 與任意 Broker 建立連接之前，會先去查找當前 Topic 的 Owner Broker。獲取到對應的 metadata 之后，Clients 會與 Owner Broker 之間建立 TCP 連接進行數據的交互。

• Produce: Clients 與 Topic 所在的所有 Owner Broker 之間進行通信并將消息 append 到對應的分布式日志中。

• Consume: Clients 與 Topic 所在的所有 Owner Broker 之間進行通信并從分布式日志中讀取指定的消息。

• Offset: Producer 生產到 topic 中的消息會分配一個唯一的 offset，Pulsar 中使用 MessageID 來標識 offset。消費者可以通過 offset 去日志中獲取指定位置的消息。

Apache Pulsar 的存儲層使用了 Apache BookKeeper，Pulsar 相當于 BookKeeper 的 Client，通過調用 ManagerLedger 對象能夠很容易的達到為分布式日志操作的目的。基于此，RoP 可以很好的將 RocketMQ 中對 commitLog 和 queueLog 的操作映射到 BookKeeper 中來。

?

RoP 概念

Offset 和 MessageID

在 RocketMQ 中，使用 offset 來標識消息的位置，當消息被生產到指定的 Topic 之后，會為每一個消息分配一個唯一的 offset；在 Pulsar 中，使用 MessageID 來唯一標識每條消息，每一個 MessageID 由三部分組成，ledgerID、entryID 和 partitionID。我們通過合理的劃分將 messageID 和 offset 進行映射，來唯一標識 Topic 中的每一條消息。

Message

對于一條消息，RocketMQ 和 Pulsar 都包含消息的 headers 和 payload 等字段，通過對消息協議的解析，我們可以輕松的將 RocketMQ message 轉換為 Pulsar 的 message 格式。為了更好的兼容 Tag 消息的功能，在消息協議的處理方面增加了 8 字節的特殊字段，用來區分該消息是否屬于 Tag 消息。

Topic Lookup

在 Pulsar 中，client 與 broker 建立連接之前，會根據當前傳入的 Topic 執行 Lookup 操作，在 Broker 集群中尋找當前 Topic 所在的 Owner Broker，然后將該 Owner Broker 的地址返回并與 client 建立 TCP 連接，再進行數據交互。在 RocketMQ 中，client 與 broker 建立連接之前，會先處理 GET_ROUTEINTO_BY_TOPIC 命令，獲取 topic 所在的路由信息后，建立對應的 TCP 連接，再進行數據交互。

?

如何使用 RoP?

目前，RoP 發布了 0.1.0 版本，你可以用過以下任一方式參與該項目：

• 想上手試試？

可在以下網址下載 RoP 和查閱用戶指南：https://github.com/streamnative/rop/blob/master/README.md。無論是快速啟動 standalone RoP 或在現有 Pulsar 集群中部署 RoP，都可輕松實現。

另外，為了方便快速使用并驗證 RoP，我們提供了 RocketMQ 的常見使用場景和用例，你可以直接使用這些代碼示例驗證服務：https://github.com/streamnative/rop/tree/master/examples/src/main/java/org/streamnative/rocketmq/example

• 想解決問題？

如有任何問題，可以在 RoP GitHub repo 中 創建 issue 或加入 RoP 微信群進行討論。無論哪種方式，RoP 資深專家都隨時在線：https://github.com/streamnative/rop/issues/new

• 想參與貢獻？

RoP 源碼開放并托管在 GitHub 上：https://github.com/streamnative/rop%EF%BC%89%E3%80%82。如需改進功能或修復 bug，歡迎提交 PR。

有道無術，術可成；有術無道，止于術

歡迎大家關注Java之道公眾號

好文章，我在看??

總結

以上是生活随笔為你收集整理的腾讯宣布开源 RoP：Apache Pulsar 支持原生 RocketMQ 协议的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。