Dubbo的负载均衡算法

Posted 2021-02-15 wolf_w

目录

• 1 简介
  • 1.1 自适应默认算法
  • 1.2 抽象基类
    • 1.2.1 选择Invoker
    • 1.2.2 计算权重
• 2 负载均衡算法实现
  • 2.1 加权随机算法
  • 2.2 最小活跃数算法
  • 2.3 一致性哈希
  • 2.4 加权轮询算法

1 简介

Dubbo提供了4种负载均衡机制：

• 权重随机算法：RandomLoadBalance
• 最少活跃调用数算法：LeastActiveLoadBalance
• 一致性哈希算法：ConsistentHashLoadBalance
• 加权轮询算法：RoundRobinLoadBalance

Dubbo的负载均衡算法均实现自LoadBalance接口，其类图结构如下：

1.1 自适应默认算法

Dubbo的负载均衡算法利用Dubbo的自适应机制，默认的实现为RandomLoadBalance（即：权重随机算法），接口如下：

// 默认为RandomLoadBalance算法
@SPI(RandomLoadBalance.NAME)
public interface LoadBalance {
    // 该方法为自适应方法
    @Adaptive("loadbalance")
    <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;
}

1.2 抽象基类

1.2.1 选择Invoker

抽象基类针对select()方法判断invokers集合的数量，如果集合中只有一个Invoker，则无需使用算法，直接返回即可：

@Override
public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    if (invokers == null || invokers.isEmpty())
        return null;
    // 如果只有一个元素，无需负载均衡算法
    if (invokers.size() == 1)
        return invokers.get(0);
    // 负载均衡算法：该方法为抽象方法，由子类实现
    return doSelect(invokers, url, invocation);
}

1.2.2 计算权重

抽象基类除了对select()方法进行了重写，还封装了服务提供者的权重计算逻辑。Dubbo的权重计算，考虑到服务预热（服务器启动后，以低功率方式运行一段时间，当服务器运行效率逐渐达到最佳状态）：

若当前的服务器运行时长小于服务预热时间，对服务器降权，让服务器的权重降低。

protected int getWeight(Invoker<?> invoker, Invocation invocation) {
    // 从URL中获取服务器Invoker的权重信息
    int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);
    if (weight > 0) {
        // 获取服务器Invoker的启动时间戳
        long timestamp = invoker.getUrl().getParameter(Constants.REMOTE_TIMESTAMP_KEY, 0L);
        if (timestamp > 0L) {
            int uptime = (int) (System.currentTimeMillis() - timestamp);
            // 获取服务器Invoker的预热时长（默认10分钟）
            int warmup = invoker.getUrl().getParameter(Constants.WARMUP_KEY, Constants.DEFAULT_WARMUP);
            if (uptime > 0 && uptime < warmup) {
                weight = calculateWarmupWeight(uptime, warmup, weight);
            }
        }
    }
    return weight;
}
// (uptime / warmup) * weight，即：(启动时长 / 预热时长) * 原权重值
static int calculateWarmupWeight(int uptime, int warmup, int weight) {
    int ww = (int) ((float) uptime / ((float) warmup / (float) weight));
    return ww < 1 ? 1 : (ww > weight ? weight : ww);
}

2 负载均衡算法实现

2.1 加权随机算法

该算法思想简单：假如有服务器：

• A，权重weight=2
• B，权重weight=3
• C，权重weight=5

这些服务器totalWeight = AWeight + BWeight + CWeight = 10。这样生成10以内的随机数

• [0, 2)：属于服务器A
• [2, 5)：属于服务器B
• [5, 10)：属于服务器C

加权随机算法由RandomLoadBalance，其核心源码实现如下:

private final Random random = new Random();

@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    int length = invokers.size();
    // 总权重值
    int totalWeight = 0;
    // 是否所有的Invoker服务器的权重值都相等，若都相等，则在Invoker列表中随机选中一个即可
    boolean sameWeight = true;
    for (int i = 0; i < length; i++) {
        int weight = getWeight(invokers.get(i), invocation);
        totalWeight += weight; // Sum
        if (sameWeight && i > 0 && weight != getWeight(invokers.get(i - 1), invocation)) {
            sameWeight = false;
        }
    }
    // 生成totalWeight以内的随机数offset，然后该offset挨个减Invoker的权重，一旦小于0，就选中当前的Invoker
    if (totalWeight > 0 && !sameWeight) {
        int offset = random.nextInt(totalWeight);
        for (int i = 0; i < length; i++) {
            offset -= getWeight(invokers.get(i), invocation);
            if (offset < 0) {
                return invokers.get(i);
            }
        }
    }
    return invokers.get(random.nextInt(length));
}

2.2 最小活跃数算法

活跃调用数越小，表明该Provider的效率越高，单位时间内可以处理更多的请求。此时如果有新的请求，应该将请求优先分配给该Provider。Dubbo中，每个Provider都会保持一个active，表示该Provider的活跃数：

没收到一个请求，active的值加1，请求处理完成后，active的值减1。

Dubbo使用LeastActiveLoadBalance实现最小活跃数算法，LeastActiveLoadBalance引入了权重，若多个Provider的活跃数相同

则权重较高的Provider被选中的几率更大
若权重相同，则随机选取一个Provider

该算法的核心思想：

• 循环遍历所有Invoker，找出活跃数最小的所有Invoker
• 如果有多个Invoker具有多个相同的最小活跃数，此时需要记录这些Invoker。Dubbo使用额外的int数组，来记录这些Invoker在原invokers列表中的索引位置，这种情况下，则需要依据随机权重算法，最终决定该选择的Invoker

private final Random random = new Random();
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    int length = invokers.size(); // Number of invokers
    // “最小活跃数”的值
    int leastActive = -1;
    // 具有相同“最小活跃数”的Invoker（即Provider）的数量
    int leastCount = 0;
    // “最小活跃数”的Invoker可能不止一个，因此需要记录所有具有“最小活跃数”的Invoker在invokers列表中的索引位置
    int[] leastIndexs = new int[length];
    // 记录所有“最小活跃数”的Invoker总的权重值
    int totalWeight = 0;
    // 记录第一个“最小活跃数”的Invoker的权重值
    int firstWeight = 0;
    // 所有的“最小活跃数”的Invoker的权重值是否都相等
    boolean sameWeight = true;
    for (int i = 0; i < length; i++) {
        Invoker<T> invoker = invokers.get(i);
        // 活跃数
        int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();
        // 权重
        int afterWarmup = getWeight(invoker, invocation); 
        if (leastActive == -1 || active < leastActive) {
            leastActive = active; 
            leastCount = 1; 
            leastIndexs[0] = i; 
            totalWeight = afterWarmup; 
            firstWeight = afterWarmup; 
            sameWeight = true;
        } else if (active == leastActive) { 
            leastIndexs[leastCount++] = i; 
            totalWeight += afterWarmup;
            if (sameWeight && i > 0 && afterWarmup != firstWeight) {
                sameWeight = false;
            }
        }
    }
    if (leastCount == 1) {
        return invokers.get(leastIndexs[0]);
    }
    if (!sameWeight && totalWeight > 0) {
        int offsetWeight = random.nextInt(totalWeight) + 1;
        for (int i = 0; i < leastCount; i++) {
            int leastIndex = leastIndexs[i];
            offsetWeight -= getWeight(invokers.get(leastIndex), invocation);
            if (offsetWeight <= 0)
                return invokers.get(leastIndex);
        }
    }
    return invokers.get(leastIndexs[random.nextInt(leastCount)]);
}

2.3 一致性哈希

该算法起初用于缓存系统的负载均衡，算法过程如下：

① 根据IP或其他信息为缓存节点生成一个hash值，将该hash值映射到[0, 2^32-1]的圆环中。
② 当查询或写入缓存时，为缓存的key生成一个hash值，查找第一个大于等于该hash值的缓存节点，以应用该缓存
③ 若当前节点挂了，则依然可以计算得到一个新的缓存节点

Dubbo通过引入虚拟节点，让所有的Invoker在圆环上分散开，避免数据倾斜（由于节点不够分散，导致大量请求落在同一个节点，而其他节点只会接受到少量请求）。Dubbo对一个Invoker默认使用160个虚拟节点：

如：Invoker1-1, invoker1-2, invoker1-3... invoker1-160

Dubbo使用ConsistentHashLoadBalance来实现一致性哈希算法，其内部定义ConsistentHashSelector内部类完成节点选择。Dubbo中使用方法的入参数值进行哈希，来选择落点到哪一个Invoker上。

public class ConsistentHashLoadBalance extends AbstractLoadBalance {
    // key：group+version+methodName，即调用方法的完整名称
    // value：ConsistentHashSelector，利用该类完成节点选择
    private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors =  new ConcurrentHashMap<>();

    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        // 获取调用方法名
        String methodName = RpcUtils.getMethodName(invocation);
        String key = invokers.get(0).getUrl().getServiceKey() + "." + methodName;

        // 获取invokers的hashcode
        int identityHashCode = System.identityHashCode(invokers);
        ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
        // 如果 invokers 是一个新的 List 对象，意味着服务提供者数量发生了变化
        // 此时 selector.identityHashCode != identityHashCode 条件成立
        if (selector == null || selector.identityHashCode != identityHashCode) {
            // 创建新的 ConsistentHashSelector
            selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, identityHashCode));
            selector = (ConsistentHashSelector<T>) selectors.get(key);
        }

        // 利用选择器，选择节点
        return selector.select(invocation);
    }
    private static final class ConsistentHashSelector<T> {...}
}

ConsistentHashLoadBalance#doSelect()方法主要是做前置工作，每个方法的一致性哈希选择具体由ConsistentHashSelector来实现：

private static final class ConsistentHashSelector<T> {
    // 使用TreeMap存储Invoker虚拟节点，保证查询效率
    private final TreeMap<Long, Invoker<T>> virtualInvokers;
    // 虚拟节点数量
    private final int replicaNumber;
    // invokers的hashCode值
    private final int identityHashCode;
    // 参数索引数组
    private final int[] argumentIndex;

    ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
        this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
        this.identityHashCode = identityHashCode;
        URL url = invokers.get(0).getUrl();
        this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
        // 获取参与 hash 计算的参数下标值，默认对方法的第一个参数进行 hash 运算
        String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0"));
        argumentIndex = new int[index.length];
        for (int i = 0; i < index.length; i++) {
            argumentIndex[i] = Integer.parseInt(index[i]);
        }
        // 如下方法创建虚拟节点，每个invoker生成replicaNumber个虚拟结点，
        for (Invoker<T> invoker : invokers) {
            String address = invoker.getUrl().getAddress();
            for (int i = 0; i < replicaNumber / 4; i++) {
                byte[] digest = md5(address + i);  // digest是一个长度为16的数组
                // 对 digest 部分字节进行4次 hash 运算，得到四个不同的 long 型正整数
                for (int h = 0; h < 4; h++) {
                    // 根据h的值，对digest进行位运算。h=0，则0~3字节位运算；h=1，则4~7字节位运算。
                    long m = hash(digest, h);  
                    virtualInvokers.put(m, invoker);
                }
            }
        }
    }
}    

选择节点，使用TreeMap#tailMap方法，可以获取到TreeMap中大于Key的子Map，子Map中第一个Entry即为选择的Invoker。因为TreeMap不是环形结构，因此如果没有取到任何值，则认为是第一个Key的值。如下：

public Invoker<T> select(Invocation invocation) {
    // 根据argumentIndex，决定方法的哪几个参数值作为Key
    String key = toKey(invocation.getArguments());
    byte[] digest = md5(key);
    return selectForKey(hash(digest, 0));
}
private Invoker<T> selectForKey(long hash) {
    /*
        TreeMap<Integer, String> tree_map = new TreeMap<Integer, String>();
        tree_map.put(10, "Geeks");
        tree_map.put(15, "4");
        tree_map.put(20, "Geeks");
        tree_map.put(25, "Welcomes");
        tree_map.put(30, "You");
        System.out.println("The tailMap is " + tree_map.tailMap(15));
        // The tailMap is {15=4, 20=Geeks, 25=Welcomes, 30=You}
    */
    Map.Entry<Long, Invoker<T>> entry = virtualInvokers.tailMap(hash, true).firstEntry();
    if (entry == null) {
        entry = virtualInvokers.firstEntry();
    }
    return entry.getValue();
}

private String toKey(Object[] args) {
    StringBuilder buf = new StringBuilder();
    for (int i : argumentIndex) {
        if (i >= 0 && i < args.length) {
            buf.append(args[i]);
        }
    }
    return buf.toString();
}

2.4 加权轮询算法

待补充