3. UniformLoadShedder

根据前面分析可见，ThresholdShedder跟OverloadShedder都无法处理低载broker资源浪费的问题，因此UniformLoadShedder诞生了。

注记：虽然ThresholdShedder后面加入了lowerBoundaryShedding特性来解决低载broker的问题，但这是在加入UniformLoadShedder之后的事情了。

3.1 最高最低载broker

下面详细介绍一下UniformLoadShedder的算法。它首先会统计出最大、最小消息速率和最大、最小流量吞吐以及对应的broker，如下：

MutableObject<String> msgRateOverloadedBroker = new MutableObject<>();
MutableObject<String> msgThroughputOverloadedBroker = new MutableObject<>();
MutableObject<String> msgRateUnderloadedBroker = new MutableObject<>();
MutableObject<String> msgThroughputUnderloadedBroker = new MutableObject<>();
MutableDouble maxMsgRate = new MutableDouble(-1);
MutableDouble maxThroughput = new MutableDouble(-1);
MutableDouble minMsgRate = new MutableDouble(Integer.MAX_VALUE);
MutableDouble minThroughput = new MutableDouble(Integer.MAX_VALUE);

brokersData.forEach((broker, data) -> {
    double msgRate = data.getLocalData().getMsgRateIn() + data.getLocalData().getMsgRateOut();
    double throughputRate = data.getLocalData().getMsgThroughputIn()
            + data.getLocalData().getMsgThroughputOut();
    if (msgRate > maxMsgRate.getValue()) {
        msgRateOverloadedBroker.setValue(broker);
        maxMsgRate.setValue(msgRate);
    }

    if (throughputRate > maxThroughput.getValue()) {
        msgThroughputOverloadedBroker.setValue(broker);
        maxThroughput.setValue(throughputRate);
    }

    if (msgRate < minMsgRate.getValue()) {
        msgRateUnderloadedBroker.setValue(broker);
        minMsgRate.setValue(msgRate);
    }

    if (throughputRate < minThroughput.getValue()) {
        msgThroughputUnderloadedBroker.setValue(broker);
        minThroughput.setValue(throughputRate);
    }
});

然后计算最大最小的差距，有两个阈值，分别对应消息速率和吞吐大小。

# Message-rate percentage threshold between highest and least loaded brokers for
# uniform load shedding. (eg: broker1 with 50K msgRate and broker2 with 30K msgRate
# will have 66% msgRate difference and load balancer can unload bundles from broker-1
# to broker-2)
loadBalancerMsgRateDifferenceShedderThreshold=50

# Message-throughput threshold between highest and least loaded brokers for
# uniform load shedding. (eg: broker1 with 450MB msgRate and broker2 with 100MB msgRate
# will have 4.5 times msgThroughout difference and load balancer can unload bundles
# from broker-1 to broker-2)
loadBalancerMsgThroughputMultiplierDifferenceShedderThreshold=4

注记：这两个配置值的类型都是double。

double msgRateDifferencePercentage = ((maxMsgRate.getValue() - minMsgRate.getValue()) * 100)
                / (minMsgRate.getValue());
        double msgThroughputDifferenceRate = maxThroughput.getValue() / minThroughput.getValue();

• loadBalancerMsgRateDifferenceShedderThreshold

负载最高和负载最低的broker之间的消息速率百分比阈值，默认值为50，因此当最大消息速率是最小消息速率的1.5倍时，即可触发bundle unload。 例如：具有 50K msgRate 的broker1 和具有 30K msgRate 的broker2 将具有 (50-30) / 30 = 66% msgRate 差异，那么负载均衡器可以将bundle从broker1 卸载到broker2。

虽然配置描述这是一个百分比值，但是其实可以允许超过100的值，从而允许最大最小消息速率之间更大的差距。设最大、最小消息速率为X、Y，阈值为P，则触发条件为：

$$\frac{X-Y}{Y} * 100 > P \Leftrightarrow X > (1+\frac{P}{100})Y$$

当P为100时，则允许最大一倍的差距。

• loadBalancerMsgThroughputMultiplierDifferenceShedderThreshold

负载最高和负载最低的broker之间的消息吞吐量倍数阈值，默认值为4，因此当最大流量吞吐是最小流量吞吐的4倍时，即可触发bundle unload。 例如：broker1 的吞吐为 450MB，broker2 的吞吐为 100MB，msgThroughout 的差异为 450 / 100 = 4.5 倍，那么负载均衡器可以将bundle从 Broker1 卸载到 Broker2。

boolean isMsgRateThresholdExceeded = conf.getLoadBalancerMsgRateDifferenceShedderThreshold() > 0
                && msgRateDifferencePercentage > conf.getLoadBalancerMsgRateDifferenceShedderThreshold();
boolean isMsgThroughputThresholdExceeded = conf.getLoadBalancerMsgThroughputMultiplierDifferenceShedderThreshold() > 0
                && msgThroughputDifferenceRate > conf.getLoadBalancerMsgThroughputMultiplierDifferenceShedderThreshold();

如果阈值配置为不大于0的值，则会禁用根据消息速率或者流量吞吐来shedding；如果两个阈值都达到了，则会优先根据消息速率来执行shedding。

3.2 挑选bundle

接下来计算需要卸载多少消息速率/吞吐对应的bundle，算法为：最大最小的差值 乘以 maxUnloadPercentage，默认为0.2。

if (isMsgRateThresholdExceeded || isMsgThroughputThresholdExceeded) {
    MutableInt msgRateRequiredFromUnloadedBundles = new MutableInt(
            (int) ((maxMsgRate.getValue() - minMsgRate.getValue()) * conf.getMaxUnloadPercentage()));
    MutableInt msgThroughputRequiredFromUnloadedBundles = new MutableInt(
            (int) ((maxThroughput.getValue() - minThroughput.getValue())
                    * conf.getMaxUnloadPercentage()));

# In the UniformLoadShedder and AvgShedder strategy, the maximum unload ratio.
# For AvgShedder, recommend to set to 0.5, so that it will distribute the load evenly
#  between the highest and lowest brokers.
maxUnloadPercentage = 0.2

3.2.1 根据消息速率

如果根据消息速率来执行shedding，则做一些检查，比如说超载broker的bundle个数要大于1，将要卸载的总消息速率大小要达到阈值minUnloadMessage，默认配置为1000。

if (isMsgRateThresholdExceeded) {
                LocalBrokerData overloadedBrokerData =
                        brokersData.get(msgRateOverloadedBroker.getValue()).getLocalData();
                if (overloadedBrokerData.getBundles().size() > 1
                        && (msgRateRequiredFromUnloadedBundles.getValue() >= conf.getMinUnloadMessage())) {

@FieldContext(
        dynamic = true,
        category = CATEGORY_LOAD_BALANCER,
        doc = "In the UniformLoadShedder and AvgShedder strategy, the minimum message that triggers unload."
)
private int minUnloadMessage = 1000;

通过检查后，则开始挑选bundle，同样是从大到小挑选，但是还多了一个限制，unload的bundle总个数不能超过maxUnloadBundleNumPerShedding，默认无限制。

@FieldContext(
        dynamic = true,
        category = CATEGORY_LOAD_BALANCER,
        doc = "For each uniform balanced unload, the maximum number of bundles that can be unloaded."
                + " The default value is -1, which means no limit"
)
private int maxUnloadBundleNumPerShedding = -1;

3.2.2 根据流量吞吐

如果根据吞吐来执行shedding，也有类似的检查，卸载的流量大小起码要达到minUnloadMessageThroughput，默认为1MB。

} else {
                LocalBrokerData overloadedBrokerData =
                        brokersData.get(msgThroughputOverloadedBroker.getValue()).getLocalData();
                if (overloadedBrokerData.getBundles().size() > 1
                        &&
                        msgThroughputRequiredFromUnloadedBundles.getValue() >= conf.getMinUnloadMessageThroughput()) {
                    // Sort bundles by throughput, then pick the bundle which can help to reduce load uniformly with
                    // under-loaded broker

@FieldContext(
        dynamic = true,
        category = CATEGORY_LOAD_BALANCER,
        doc = "In the UniformLoadShedder and AvgShedder strategy, the minimum throughput that triggers unload."
)
private int minUnloadMessageThroughput = 1 * 1024 * 1024;

挑选bundle的逻辑基本类似，这里就不赘述了。

3.3 评价

介绍完UniformLoadShedder的算法，我们显然可以得到如下信息：

• UniformLoadShedder没有低载broker资源浪费的问题，它考虑了低载broker的情况，这也是它提出来的动机。

• UniformLoadShedder没有处理负载抖动的逻辑，例如，当流量在短时间内急剧上升或大幅下降时，它会采纳这些负载数据点，进而触发 bundle unload。然而，没过多久该 topic 的流量便会恢复正常，如此一来很可能会再次触发 bundle unload，而这类 bundle unload 实际上是应当避免的。这种场景颇为常见，如下图所示：

• 而ThresholdShedder在打分时会对历史打分加上很高的权重，资源使用率的巨大变化不会立马反映到负载分数上，负载分数需要经过多轮迭代才能接近资源使用率，因此能处理这种负载抖动的场景。

• UniformLoadShedder在判定高载与低载broker的时候，不会依赖CPU使用率、网卡使用率等指标来判定的，而是根据消息速率、流量吞吐大小来判定的；而ThresholdShedder和OverloadShedder都依赖CPU使用率等机器资源指标来判定。

  • 若集群中的机器为异构类型，即不同机器的硬件配置存在差异，或者 broker 所在机器有其他进程共享资源，那么 UniformLoadShedder 极有可能误判高负载与低负载的 broker，进而把负载从高性能且低负载的 broker 迁移至低性能却高负载的 broker 上。所以，不建议用户在异构环境中使用 UniformLoadShedder。

  • 简单举个例子，如果broker1的性能是broker2的两倍，此时broker1承担150MB/s的负载，CPU使用率为40%，broker2承担100MB/s的负载，CPU使用率为60%，此时UniformLoadShedder仍然会判断broker1是高载broker，broker2是低载broker，将broker1的负载卸到broker2，这显然是一个极为错误的决策。

• UniformLoadShedder执行一次shedding只会从一台最高载broker上卸载bundle，对于大集群来说，这可能会使得集群消耗相当长的时间才能完成所有负载均衡任务。

  • 比如说当前集群有100台高载broker，扩容100台机器，粗略估算需要执行100次shedding才能完成均衡任务，但是由于LoadSheddingStrategy策略的执行时间间隔由配置loadBalancerSheddingIntervalMinutes决定，默认为1min一次，因此需要100min才能完成所有任务，对于使用大分区 topic 的用户而言，在这 100 分钟内，其任务极有可能多次遭遇连接中断的情况，这无疑会对用户使用体验产生极为不利的影响。

  • 从用户体验视角来看，当遭遇诸如 broker 扩缩容、用户上下线引发的负载增减这类偶发事件时，理想效果是能够快速完成一次负载均衡，且在此之后基本无需再执行负载均衡操作，而非长时间持续执行负载均衡。