Pulsar负载均衡原理及优化

Posted on 2023-02-07 Edited on 2023-12-28 In Pulsar Views: Views: Disqus:

前言

前段时间我们在升级 Pulsar 版本的时候发现升级后最后一个节点始终没有流量。

虽然对业务使用没有任何影响，但负载不均会导致资源的浪费。

和同事沟通后得知之前的升级也会出现这样的情况，最终还是人工调用 Pulsar 的 admin API 完成的负载均衡。

这个问题我尝试在 Google 和 Pulsar 社区都没有找到类似的，不知道是大家都没碰到还是很少升级集群。

我之前所在的公司就是一个版本走到黑😂

Pulsar 负载均衡原理

当一个集群可以水平扩展后负载均衡就显得非常重要，根本目的是为了让每个提供服务的节点都能均匀的处理请求，不然扩容就没有意义了。

在分析这个问题的原因之前我们先看看 Pulsar 负载均衡的实现方案。

1 2	# Enable load balancer loadBalancerEnabled=true

我们可以通过这个 broker 的这个配置来控制负载均衡器的开关，默认是打开。

但具体使用哪个实现类来作为负载均衡器也可以在配置文件中指定：

1 2	# Name of load manager to use loadManagerClassName=org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl

默认使用的是 ModularLoadManagerImpl。

static LoadManager create(final PulsarService pulsar) {
    try {
        final ServiceConfiguration conf = pulsar.getConfiguration();
        // Assume there is a constructor with one argument of PulsarService.
        final Object loadManagerInstance = Reflections.createInstance(conf.getLoadManagerClassName(),
                Thread.currentThread().getContextClassLoader());
        if (loadManagerInstance instanceof LoadManager) {
            final LoadManager casted = (LoadManager) loadManagerInstance;
            casted.initialize(pulsar);
            return casted;
        } else if (loadManagerInstance instanceof ModularLoadManager) {
            final LoadManager casted = new ModularLoadManagerWrapper((ModularLoadManager) loadManagerInstance);
            casted.initialize(pulsar);
            return casted;
        }
    } catch (Exception e) {
        LOG.warn("Error when trying to create load manager: ", e);
    }
    // If we failed to create a load manager, default to SimpleLoadManagerImpl.
    return new SimpleLoadManagerImpl(pulsar);
}

当 broker 启动时会从配置文件中读取配置进行加载，如果加载失败会使用 SimpleLoadManagerImpl 作为兜底策略。

当 broker 是一个集群时，只有 leader 节点的 broker 才会执行负载均衡器的逻辑。

Leader 选举是通过 Zookeeper 实现的。

默然情况下成为 Leader 节点的 broker 会每分钟读取各个 broker 的数据来判断是否有节点负载过高需要做重平衡。

而是否重平衡的判断依据是由 org.apache.pulsar.broker.loadbalance.LoadSheddingStrategy 接口提供的，它其实只有一个函数：

public interface LoadSheddingStrategy {

    /**
     * Recommend that all of the returned bundles be unloaded.
     * @return A map from all selected bundles to the brokers on which they reside.
     */
    Multimap<String, String> findBundlesForUnloading(LoadData loadData, ServiceConfiguration conf);
}

根据所有 broker 的负载信息计算出一个需要被 unload 的 broker 以及 bundle。

这里解释下 unload 和 bundle 的概念：

bundle 是一批 topic 的抽象，将 bundle 和 broker 进行关联后客户端才能知道应当连接哪个 broker；而不是直接将 topic 与 broker 绑定，这样才能实现海量 topic 的管理。
unload 则是将已经与 broker 绑定的 bundle 手动解绑，从而触发负载均衡器选择一台合适的 broker 重新进行绑定；通常是整个集群负载不均的时候触发。