本文作者为中国移动云能力中心大数据团队软件开发工程师李思渊，本文针对 Apache Pulsar 在 Tenant 和 Namespace 级别限速的特性，对实现原理进行分析并实践验证，供大家参考。

一、背景

Apache Pulsar 是一个分布式的消息中间件，作为大数据场景下的一个重要的组件 Pulsar 被广泛应用于不同组件或平台之间传输数据。Pulsar 设计之初就采用了计算存储分离的架构，特别适合进行云原生的改造，对于消息中间件来说流量是最重要的一个指标，其相应的 Pulsar 云原生产品也是根据流量制定不同规格的产品，以满足用户的不同需求。

在底层的构建过程中，要想实现以流量划分规格，就需要用某种方式实现对 Pulsar 集群的限速。现有的方法是针对 Pulsar 集群下的每个 Broker 进行限速，最后实现集群的整体限速。

Pulsar 内部也是使用 Topic 及订阅的方式进行生产消费的，如果一个 Topic 的 Partition 分配不均匀，那么就可能出现流量集中在某些 Broker 上，这样就会导致流量分配不均匀，再结合限速策略，最后导致无法达到实际的流量限制。

Pulsar 现在引入了一个新的特性：Tenant 和 Namespace 等级的限速（参照 PIP-82），这里提供了一种新的限速方法，在 Tenant 和 Namespace 级别进行限速，这样不仅可以避免上述问题，也可以更加灵活地管理 Pulsar 集群下的 Tenant 和 Namespace 的资源分配。

二、设计

为了实现 Tenant 和 Namespace 级别的限速，这里引入了一个新的资源：Resourcegroup。其主要的设计目标有两个：

1.使用 Resourcegroup 描述消息的速率，Resourcegroup 可以注册到 Tenant 或 Namespace，之后 Pulsar 集群便会根据 Resourcegroup 内的相关的流量限制实现对这个层级的限速。2.当动态调整 Resourcegroup 内的指标时，Pulsar 集群也会动态地修改该 Resourcegroup 所属层级的限速。

Resourcegroup 实现的限速有以下特点：

1.Resourcegroup 不会实现一种严格的限速，在短时间内会有速率波动（高于或低于限速的 10%）。2.不允许用户在多层级下（Tenant/Namespace/Topic）设置限速以实现一种分层限速机制。3.在一个层级实现的限速，并不会把流量配额平均分配给低层级：例如给 Namespace 实现限速，但此 Namespace 下的 Topic 不会出现平均的速率分配。4.通过 Resourcegroup 实现的限速不会跨集群。

三、使用

Resourcegroup 内包含了 4 个值:

•byte-dispatch-rate 消费速率限制（byte）

•byte-publish-rate 生产速率限制（byte）

•msg-dispatch-rate 消费速率限制（消息数）

•msg-publish-rate 生产速率限制（消息数）

默认情况下不进行修改的值都是不作限制的，只有创建或更新 Resourcegroup 内的相应指标为具体数值时才会执行相应的限制策略。

创建一个 Resourcegroup，并指定初始限速：

./bin/pulsar-admin resourcegroups create rg-2 --msg-publish-rate 15

创建完毕后，为指定 Namespace 设置 Resourcegroup：

./bin/pulsar-admin namespaces set-resource-group public/rgns2 --resource-group-name rg-2

使用 pulsar-perf 脚本进行测试，分别开启 Producer 和 Consumer：

./bin/pulsar-perf produce persistent://public/rgns2/rg201 -r 1000 -s 1024
sume persistent://public/rgns2/rg201 -ss sub201

可以看到生产速率受到了限制。

由于生产速率的限制，消费侧也维持在相应水平。

Resourcegroup 的速率限制是动态的，这个时候我们可以尝试进行更新：

./bin/pulsar-admin resourcegroups update rg-2 --msg-publish-rate 150

可以看到生产速率提升到了新的速率限制。

消费速率也有相应的提升。

四、实现

4.1 消息传输限流流程

ResourceGroup 可以被设置到某个 Tenant 或 Namespace 下，之后就可以开始执行其限速功能，当外部的 Producer 或 Consumer 要生产或消费 Topic 的时候，其逻辑就会传到 ResourceGroup 创建的 RateLimiter，其内部会进行一个判断，如果流量还未达到限速，就可以通过 API 读写 Bookkeeper 内的数据，如果达到了限速就返回，直到下一个限速周期的到来。

4.2 内部限流实现

为了实现流量限制，首先就要知道各个 Broker 在此范围内的流量配额使用情况，内部实现机理就是建立一个 Non-persistent 模式的 Topic，Broker 将流量配额使用相关数据传到这个 Topic 内以便相互通信。

在 N 个周期内，每个 Broker 将会强制报告一次 Resourcegroup 的流量配额使用情况，当某个 Broker 的流量相较上一次报告发生了显著变化（如变化超过 10%），则每一个 Broker 都将会根据此信息对此 Broker 的 Resourcegroup 流量配额使用情况进行跟踪。

传输的信息格式如下：

{   broker : “broker-1.example.com”,   usage : {      “resource-group-1” : {         topics: 1,         publishedMsg : 100,         publishedBytes : 100000,      },      “resource-group-2” : {         topics: 1,         publishedMsg : 1000,         publishedBytes : 500000,      },      “resource-group-3” : {         topics: 1,         publishedMsg : 80000,         publishedBytes : 9999999,      },   }}

对于单个 Broker 来说，它会读取此 topic 内的数据，并将其存储在一个 Hash Map 中,之后 Broker 内部就会进行相应的处理，根据 Resourcegroup 的流量配额使用情况，调整自己的现在的流量。

启动流程如下：

在启动 Broker 的时，就会启动 ResourceGroupService，ResourceGroupService 初始化的时候，会启动定期任务 aggreagteLocalUsagePeriodicTask 和 calculateQuotaPeriodicTask。

•aggreagteLocalUsagePeriodicTask 负责定期聚合所有已知 BrokerService 的流量使用情况，会周期性地执行 aggregateResourceGroupLocalUsages 方法，针对生产消费两种行为，aggregateResourceGroupLocalUsages 会调用 updateStatsWithDiff 方法，找出最新更新的统计数据，若有差异则更新统计信息。

private void updateStatsWithDiff(String topicName, String tenantString, String nsString,                                 long accByteCount, long accMesgCount, ResourceGroupMonitoringClass monClass) {    ConcurrentHashMap<String, BytesAndMessagesCount> hm;    switch (monClass) {        ……        case Publish:            hm = this.topicProduceStats;            break;        case Dispatch:            hm = this.topicConsumeStats;            break;    }
    BytesAndMessagesCount bmDiff = new BytesAndMessagesCount();    BytesAndMessagesCount bmOldCount;    BytesAndMessagesCount bmNewCount = new BytesAndMessagesCount();
    bmNewCount.bytes = accByteCount;    bmNewCount.messages = accMesgCount;
    bmOldCount = hm.get(topicName);    if (bmOldCount == null) {        bmDiff.bytes = bmNewCount.bytes;        bmDiff.messages = bmNewCount.messages;    } else {        bmDiff.bytes = bmNewCount.bytes - bmOldCount.bytes;        bmDiff.messages = bmNewCount.messages - bmOldCount.messages;    }    if (bmDiff.bytes <= 0 || bmDiff.messages <= 0) {        return;    }
    try {        boolean statsUpdated = this.incrementUsage(tenantString, nsString, monClass, bmDiff);        hm.put(topicName, bmNewCount);    } catch (Throwable t) {        log.error(……);    }}

•calculateQuotaPeriodicTask 负责定期计算所有 ResourceGroup 的更新的流量配额，周期性地执行 calculateQuotaForAllResourceGroups 方法，其内部的 computeLocalQuota 方法会根据配置值和实际值计算本地的流量配额。residual 为流量剩余可用值，myUsageFraction 计算当前流量和总流量的比值，根据比例调整接下来的流量变化。

public long computeLocalQuota(long confUsage, long myUsage, long[] allUsages) throws PulsarAdminException {    ……    float residual = confUsage - totalUsage;    float myUsageFraction = (float) myUsage / totalUsage;    float calculatedQuota = max(myUsage + residual * myUsageFraction, 1);    val longCalculatedQuota = (long) calculatedQuota;    return longCalculatedQuota;}

4.3 流量调整方式

当 Broker 获取了现在所有 Broker 流量的使用情况后，它就会根据实际情况对流量进行调整。例如：如果某个 ResourceGroup 限制流量为 100，某时刻有三个 Broker 分别使用了 10、50、30 的流量。第一阶段，后台会先计算出剩余流量配额为 10，这时每个 Broker 就会增加未使用的流量配额，此时三个 Broker 的流量使用分别为：20、60、40。第二阶段后台计算后，总使用流量超出了配额，这时后台会按照比例对 Broker 的流量进行限制，最终三个 Broker 的流量变为：11.11、55.56、33.33。如图所示：

五、总结

在实际生产情况下，以往的基于 Broker 的限流对于重点要求集群限流的场景来说显得过于粗暴，无法达到理想的效果，现在 Tenant 和 Namespace 级别的限速可以优雅地解决这一问题，我们也在积极和社区沟通并持续改进和完善这一功能，期待未来能成为一个主流且可靠的限流方案。