作者简介

冉小龙 - 腾讯云中间件团队研发工程师，Apache Pulsar committer、RoP 作者及 Maintainer、Apache BookKeeper contributor，Apache Pulsar Go client、Apache Pulsar Go Functions、StreamNative/pulsarctl 作者。

摘要

日前，腾讯云中间件团队联结 StreamNative 团队正式公布了 RoP 0.2.0 版本，该版本在架构上全新降级，用户在应用中能够完全避免音讯失落、音讯反复生产、只能生产一部分 Partition 的数据等问题。

RoP 的定义

与 KoP、MoP 和 AoP 类似，RoP 是一种可插拔的协定解决插件。

将 RoP 协定解决插件增加到现有 Pulsar 集群后，用户无需批改代码，便能将现有的 RocketMQ 应用程序和服务迁徙到 Pulsar，同时还能应用 Pulsar 的弱小性能，例如：

• 计算与存储拆散
• 多租户
• 跨地区复制
• 分层分片
• 轻量化计算框架 — Pulsar Functions
• ……

公布 RoP 0.2.0

2021 年 5 月 17 日，腾讯云中间件团队向社区奉献了 RoP 0.1.0 的 beta 版本，RoP(RocketMQ on Pulsar) 是 将 RocketMQ 协定解决插件引入 Pulsar Broker，这样 Pulsar 即可反对原生的 RocketMQ 协定，RocketMQ 用户能够无缝迁徙到 Apache Pulsar。

明天，咱们重磅公布 RoP 0.2.0，该版本在架构上全新降级，在性能和稳定性上失去了更大的晋升。提供了 ACL 鉴权和验证的性能，能够更好的确保用户数据的安全性，同时容许用户对 Partitioned Topic 进行扩容，能够取得更好的并发写入能力，并且欠缺了 RocketMQ 原生的管控端接口，能够更好的对服务进行解决和监控。

最新性能优化

在 0.2.0 版本中，腾讯云中间件团队在 0.1.0 的架构上进行全新设计，对 MessageID、音讯路由模型进行重构，确保不同场景下 RoP 音讯的准确性。

次要有以下三点优化内容：

1. 反对 RoP ACL 性能

ACL 机制是 RocketMQ 社区自带的一个能力，能够很好的对用户的数据进行鉴权和认证。RoP 0.2.0 版本复用了 RocketMQ 本身的 Hook 实现，利用 Pulsar 本身的鉴权机制，实现了对用户数据进行鉴权和认证的性能。

RoP ACL 的应用形式仍旧连续了 RocketMQ 的应用形式，只需定义 ACL_ACCESS_KEY 和 ACL_SECRET_KEY 字段，而后利用 RocketMQ 的 ACLRPCHook 函数加载即可，这样能够确保用户尽可能少的改变客户端的业务代码逻辑。

具体代码示例如下：

private static final String ACL_ACCESS_KEY = "eyJrZXlJZCI6InJvY2tldG1xLW13bmI3bWFwMjhqZSIsImFsZyI6IkhTMjU2In0."
+ "eyJzdWIiOiJyb2NrZXRtcS1td25iN21hcDI4amVfdGVzdCJ9.BDOjqqY25a6apnZTMZCqg0I0pxVFcqz7fvZbaTqkf5U"; // token
private static final String ACL_SECRET_KEY = "rop";
public static void producer() throws MQClientException {
DefaultMQProducer producer = new DefaultMQProducer("rocketmq-mwnb7map28je|nit", "ProducerGroupName",
getAclRPCHook());
...
}
static RPCHook getAclRPCHook() {return new AclClientRPCHook(new SessionCredentials(ACL_ACCESS_KEY, ACL_SECRET_KEY));
}

• ACL_ACCESS_KEY：即用户在 Namespace 级别下创立的 Token。
• ACL_SECRET_KEY：固定值，在 RoP 外部解析时，不会应用到这个字段。

2. 重构 MessageID

RocketMQ 与 Kafka 相似，都是应用 64 位的 Offset 来惟一标识一条音讯，然而在 Pulsar 中，应用 64 位的 LedgerID、64 位的 EntryID 来惟一标识一条音讯。针对这个问题，在 RoP 0.1.0 中，咱们应用如下的模式来结构 MessageID 对象：

• PartitionID: 8 位，能够容许一个 Topic 最多创立 256 个 Partitions
• LedgerID: 32 位
• EntryID: 24 位

应用如上的形式可能存在 MessageID 的音讯精度失落，在零碎运行一段时间之后，无奈持续创立出新的 LedgerID，导致整个集群的服务对外不可用的状况。这个问题与晚期的 KoP 版本所面临的是同样的窘境，所以在 RoP 0.2.0 中，咱们采纳了和 KoP 雷同的解决形式，应用 PIP 70: Introduce lightweight broker entry metadata 的解决思路，在 Broker 的协定头中，附加了一个 64 位的 index/publish-time 字段，这样无需在客户端侧进行协定的解析即可在每一条音讯中附加一个 64 位的字段来应用。

PIP-70 是应用插件的形式进行加载的，所以在服务启动时，咱们须要做如下配置：

brokerEntryMetadataInterceptors=org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor

Note: Broker Entry Metadata 是在 Pulsar 2.8.0 的版本中才反对的，所以须要确保 Pulsar Broker 的版本在 2.8.0 及以上。

须要阐明的是，RocketMQ 和 Kafka 在 Offset 的应用形式上又有所不同。RocketMQ 中有两个 Offset，一个是 Queue Offset，用来示意音讯在 MessageQueue 中的地位，MessageQueue 实质上是一个数组，一条音讯进来数组的下标就会 +1。一个是 CommitLog Offset 用来示意音讯存储在 CommitLog 中的地位，音讯存储是由 ConsumeQueue 和 CommitLog 配合实现，ConsumeQueue 是逻辑队列，CommitLog 是真正存储音讯文件的，ConsumeQueue 存储的是指向物理存储的地址。Topic 下的每个 MessageQueue 都有对应的 ConsumeQueue 文件，内容也会被长久化到磁盘。

所以，在 MessageID 重构的实现中，区别于 Kafka 中只有一个全局的 Offset 来标识音讯的唯一性，在 RoP 中须要针对这两种 Offset 的状况别离进行解决，具体如下：

• RESERVED_BITS: 1 字节的保留位，防止第一个字节呈现正数等状况导致 Offset 计算有误。
• RETRY_TOPIC_TAG_BITS：1 字节的标记为，用来标识这个 Topic 是否为 Retry 类型的 Topic
• PULSAR_PARTITION_ID_BITS：10 字节的 Partition Num，用来记录一个 PartitionedTopic 下有多少个 Partitions，最大反对 1024 个 Partitions。
• OFFSET_BITS：52 字节用来标识音讯的 Offset。

3. 重构音讯的路由模型

在 RoP 0.1.0 的版本中，在音讯路由的实现上，RocketMQ 和 Pulsar 都是首先通过 Topic 查找 的操作找到对应的 Owner Broker 节点，而后将该 Broker 的地址返回。然而在这个动作中，疏忽了一个重要的问题，即 RocketMQ 与 Kafka 和 Pulsar 都是不同的，它的 Queue 不是全局惟一的。

RocketMQ 路由协定次要包含两局部：

1. Broker 服务的 IP 地址信息；
2. 某个 Broker 上对应的 Topic 分区总数以及分区可读写信息。

在 RocketMQ 路由协定中，没有全局标识 Topic 的分区的惟一 ID（例如在 Pulsar/Kafka 中，分区 ID 在集群中是惟一的）；而在 RocketMQ 中，分区路由信息是由 Broker 标识加上该 Broker 上的程序从 0→N 的 Index 来标识 Topic 的分区。

因而 RocketMQ 协定中，客户端只须要获取到 Topic 对应 Broker 上分区总数，就能通过计算取得该 Broker 上分区的 ID；所有的申请都是基于【Broker-Tag】+【Broker-Topic-Seq】构建惟一路由查问原语来申请服务。简略来说：RocketMQ 的分区是有状态的，他绑定在特定的 Broker 之上；分区一旦调配在某个 Broker 上，一生与之相干且不能迁徙。客户端解析分区路由信息是通过计算失去；比方：某个 TopicA 有 5 个分区，别离落在 BrokerA 和 BrokerB 上，BrokerA 有 3 个，BrokerB 有 2 个；那么协定记录为（BrokerA，3）（BrokerB，2），客户端通过计算就失去全副的分区数据：

• BrokerA-TopicA-0，BrokerA-TopicA-1，BrokerA-TopicA-2
• BrokerB-TopicA-0，BrokerB-TopicA-1

因为下面的路由关系的起因，所以没有方法通过 GET_ROUTEINTO_BY_TOPIC 这个协定申请去和 Pulsar 的 查找 协定去做映射。实质起因是像 Kafka/Pulsar 这种，它的 Partition 信息是全局惟一的，在执行 Topic 路由策略之后，能精确的返回某一个 Topic 的 Partition 所对应的 Owner Broker 是谁。然而 RocketMQ 的 Topic 路由返回的是两个字段，一个是 Broker Name，一个是 Queue 的数量。具体的 QueueID，是 Client 依据 Broker 返回的数量固定的从 0 开始递增计算。所以在 Topic 的路由映射中，RocketMQ 和 Pulsar 本身的路由协定没方法一一映射。为了解决这个问题，在 RoP 0.2.0 中，形象了一层 Proxy 用来保护 Topic 与 Broker 之间的映射关系。为了达到这个目标，这里次要有以下几方面的事件须要思考：

1. 这些映射关系存储在哪里？
2. 如何调配路由关系？
3. 当路由关系发生变化之后，如何解决？

针对第一个问题，综合考量，咱们抉择将路由的映射关系存储到 ZooKeeper 集群中来，因为以后 RoP 的服务自身也须要依赖 ZooKeeper 集群，不会引入新的组件；其次 ZooKeeper 本身的一致性能力能很好的满足这个场景需要。

针对第二个问题，咱们是在 RoP 接口创立分区主题的同时，顺次 查找 各个分区所在的 Broker 节点，按照初始主题所在节点信息为基准，将映射关系写入到 ZooKeeper 集群中。这样做的益处在于：

• 复用 Pulsar 本身分区分配机制计算的后果，实现简略。
• 初始调配后，虚构节点和物理节点处于一个节点上，性能好。
• 如果配合路由关系重均衡能力，能够实现最优性能。

针对第三个问题，咱们通过减少 Master-Slave 模式，能够缩小单节点故障对系统的影响。ZooKeeper 元数据如下，只须要减少 Broker 相干信息，即可实现各个节点的互为主从关系，达到主节点不可用时从节点能够持续提供服务。因为以后 Offset 信息都存储在 Compact Topic 中，全副节点同时订阅，所以各个节点的元数据能够保障统一，能够实现主从切换。上面是测试环境中部署 RoP 集群中的路由映射关系：

所以，为了保障 RoP 集群能有较好的容错能力，在部署 RoP 集群中倡议应用偶数台节点。能够通过如下参数配置决定以后 Master 节点有几个 Slave 节点作为其备份节点：

RoPBrokerReplicationNum=2

假如有 6 台 Broker 节点，RoPBrokerReplicationNum=2, 那么就阐明此时只有三台 Master Broker 节点对外提供服务。然而对于 Pulsar 来说，Broker 节点之间是 对等 的，当创立 Topic 的时候，可能会调配到任意节点上，所以对于不在 Owner Broker 节点上的申请，在 RoP Proxy 层做了一层代理，会先对该 Topic 进行 查找 的操作，而后将申请转发到 Owner Broker 的节点上来返回。

将来布局

为了更好的践行开源协同和开源共建的理念，目前，上述性能均已奉献回社区。除此之外针对 RocketMQ 商业版本的任意提早音讯性能，腾讯云中间件团队也基于 Pulsar 原生的个性开发了相干的插件来进行反对。RoP 的提早音讯性能除了反对多级别的提早音讯之外还具备反对任意提早音讯的能力。

之后，腾讯云中间件团队将会在确保 RoP 我的项目稳固的同时，继续开发 RoP 相干的性能，诸如音讯轨迹，音讯查问和回溯以及监控等能力，进一步欠缺 RoP 的性能以及周边生态。

RoP 我的项目地址：https://github.com/streamnati…

特地鸣谢

感激腾讯云中间件团队韩明泽和张勇华对本文提供的技术细节校验和反对。

关注公众号「Apache Pulsar」，获取干货与动静

退出 Apache Pulsar 中文交换群 👇🏻

点击链接，下载最新版本 RoP！