学习目标

Dubbo的负载均衡策略有哪些及各自特点

第1章负载均衡

1.1 RandomLoadBalance

1.1.1 使用场景

在大请求量的情况下，想要让请求落到所有节点的比率跟机器的权重比率接近那么就可以使用随机算法。

随机算法是dubbo默认的负载均衡算法，在dubbo随机的负载均衡算法有两种

1、完全随机

2、权重随机

完全随机就是不考虑权重从服务列表中根据服务列表的长度来随机选择一个，这样做是没考虑机器性能差异的。

权重随机，是给不同机器的机器加不通的权重，机器性能好点的权重高，性能差点的权重低，权重高的机器能随机分配更多的请求，反之亦然。

1.1.2 源码分析

@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation
                                  invocation) {
    // Number of invokers
    int length = invokers.size();
    //判断是否需要权重随机
    if (!needWeightLoadBalance(invokers,invocation)){
        return invokers.get(ThreadLocalRandom.current().nextInt(length));
    }
    // Every invoker has the same weight?
    boolean sameWeight = true;
    // the maxWeight of every invokers, the minWeight = 0 or the maxWeight of the last invoker
    int[] weights = new int[length];
    // The sum of weights
    int totalWeight = 0;
    for (int i = 0; i < length; i++) {
        //获取每一个invoker的权重
        int weight = getWeight(invokers.get(i), invocation);
        // Sum
        totalWeight += weight;
        // save for later use
        weights[i] = totalWeight;
        //如果各节点权重是不相同的
        if (sameWeight && totalWeight != weight * (i + 1)) {
            sameWeight = false;
        }
    }
    if (totalWeight > 0 && !sameWeight) {
        // If (not every invoker has the same weight & at least one invoker's weight>0),select randomly based on totalWeight.
        int offset = ThreadLocalRandom.current().nextInt(totalWeight);
        // Return a invoker based on the random value.
        for (int i = 0; i < length; i++) {
            //判断offset落在哪一个权重区间
            if (offset < weights[i]) {
                return invokers.get(i);
            }
        }
    }
    //如果权重相同，则是真随机
    // If all invokers have the same weight value or totalWeight=0, return evenly.
    return invokers.get(ThreadLocalRandom.current().nextInt(length));
}

1.2 RoundRobinLoadBalance

1.2.1 使用场景

无需记录当前所有服务器的链接状态，所以它一种无状态负载均衡算法，实现简单，适用于每台服务器性能相近的场景下。为了解决轮询算法应用场景的局限性。当遇到每台服务器的性能不一致的情况，我们需要对轮询过程进行加权，以调控每台服务器的负载。

经过加权后，每台服务器能够得到的请求数比例，接近或等于他们的权重比。比如服务器 A、B、C 权重比为 5:3:2。那么在10次请求中，服务器 A 将收到其中的5次请求，服务器 B 会收到其中的3次请求，服务器 C 则收到其中的2次请求。

1.2.2 源码分析

@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation
                                  invocation) {
    String key = invokers.get(0).getUrl().getServiceKey() + "." +
        invocation.getMethodName();
    ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.computeIfAbsent(key, k
                                                                                    -> new ConcurrentHashMap<>());
    int totalWeight = 0;
    long maxCurrent = Long.MIN_VALUE;
    long now = System.currentTimeMillis();
    Invoker<T> selectedInvoker = null;
    WeightedRoundRobin selectedWRR = null;
    for (Invoker<T> invoker : invokers) {
        String identifyString = invoker.getUrl().toIdentityString();
        int weight = getWeight(invoker, invocation);
        WeightedRoundRobin weightedRoundRobin = map.computeIfAbsent(identifyString, k -> {
            WeightedRoundRobin wrr = new WeightedRoundRobin();
            wrr.setWeight(weight);
            return wrr;
        });
        if (weight != weightedRoundRobin.getWeight()) {
            //weight changed
            weightedRoundRobin.setWeight(weight);
        }
        long cur = weightedRoundRobin.increaseCurrent();
        weightedRoundRobin.setLastUpdate(now);
        if (cur > maxCurrent) {
            maxCurrent = cur;
            selectedInvoker = invoker;
            selectedWRR = weightedRoundRobin;
        }
        totalWeight += weight;
    }
    if (invokers.size() != map.size()) {
        map.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() >
                                RECYCLE_PERIOD);
    }
    if (selectedInvoker != null) {
        selectedWRR.sel(totalWeight);
        return selectedInvoker;
    }
    // should not happen here
    return invokers.get(0);
}

从上述代码我们可以看到，MockClusterInvoker当中就是走了三套逻辑：

1、没有配置mock的情况

2、mock=”force:”的情况

3、配置了mock的其他情况

我们都知道如果配置了force:就代表要进行强制降级，就不会走后端的rpc调用了，所以这里是直接调用到了

result = doMockInvoke(invocation, null);

private Result doMockInvoke(Invocation invocation, RpcException e) {
    Result result = null;
    Invoker<T> minvoker;
    //选择一个MockInvoker的实例，这里是选不到的
    List<Invoker<T>> mockInvokers = selectMockInvoker(invocation);
    if (CollectionUtils.isEmpty(mockInvokers)) {
        //所以代码会走这里，创建一个MockInvoker对象
        minvoker = (Invoker<T>) new MockInvoker(getUrl(), directory.getInterface());
    } else {
        minvoker = mockInvokers.get(0);
    }
    try {
        //调用mock的实现类方法
        result = minvoker.invoke(invocation);
    } catch (RpcException me) {
        if (me.isBiz()) {
            result = AsyncRpcResult.newDefaultAsyncResult(me.getCause(), invocation);
        } else {
            throw new RpcException(me.getCode(), getMockExceptionMessage(e, me),
                                   me.getCause());
        }
    } catch (Throwable me) {
        throw new RpcException(getMockExceptionMessage(e, me), me.getCause());
    }
    return result;
}

我们再看看mockInvoker的invoke方法

@Override
public Result invoke(Invocation invocation) throws RpcException {
    if (invocation instanceof RpcInvocation) {
        ((RpcInvocation) invocation).setInvoker(this);
    }
    String mock = null;
    if (getUrl().hasMethodParameter(invocation.getMethodName())) {
        mock = getUrl().getParameter(invocation.getMethodName() + "." + MOCK_KEY);
    }
    if (StringUtils.isBlank(mock)) {
        //获取配置的mock的属性值
        mock = getUrl().getParameter(MOCK_KEY);
    }
    if (StringUtils.isBlank(mock)) {
        throw new RpcException(new IllegalAccessException("mock can not be null. url :" +url));
    }
    //把force:前缀去掉，获取后面的值
    mock = normalizeMock(URL.decode(mock));
    //如果是return开头
    if (mock.startsWith(RETURN_PREFIX)) {
        //获取return后面的值
        mock = mock.substring(RETURN_PREFIX.length()).trim();
        try {
            //获取返回值类型
            Type[] returnTypes = RpcUtils.getReturnTypes(invocation);
            //把return后面的值包装成返回值类型
            Object value = parseMockValue(mock, returnTypes);
            //注解把结果返回没走后端rpc调用
            return AsyncRpcResult.newDefaultAsyncResult(value, invocation);
        } catch (Exception ew) {
            throw new RpcException("mock return invoke error. method :" +
                                   invocation.getMethodName()
                                   + ", mock:" + mock + ", url: " + url, ew);
        }
        //如果是throw
    } else if (mock.startsWith(THROW_PREFIX)) {
        mock = mock.substring(THROW_PREFIX.length()).trim();
        if (StringUtils.isBlank(mock)) {
            throw new RpcException("mocked exception for service degradation.");
        } else { // user customized class
            //获取异常实例
            Throwable t = getThrowable(mock);
            //直接往上抛异常
            throw new RpcException(RpcException.BIZ_EXCEPTION, t);
        }
    } else { //impl mock
        //mock实现类的方式
        try {
            Invoker<T> invoker = getInvoker(mock);
            //调用mock实例
            return invoker.invoke(invocation);
        } catch (Throwable t) {
            throw new RpcException("Failed to create mock implementation class " + mock,t);
        }
    }
}

我们看一下normalizeMock方法

public static String normalizeMock(String mock) {
    if (mock == null) {
        return mock;
    }
    mock = mock.trim();
    if (mock.length() == 0) {
        return mock;
    }
    //如果是只有一个return 则加上一个return null
    if (RETURN_KEY.equalsIgnoreCase(mock)) {
        return RETURN_PREFIX + "null";
    }
    if (ConfigUtils.isDefault(mock) || "fail".equalsIgnoreCase(mock) ||
        "force".equalsIgnoreCase(mock)) {
        return "default";
    }
    if (mock.startsWith(FAIL_PREFIX)) {
        mock = mock.substring(FAIL_PREFIX.length()).trim();
    }
    //把force:去掉
    if (mock.startsWith(FORCE_PREFIX)) {
        mock = mock.substring(FORCE_PREFIX.length()).trim();
    }
    if (mock.startsWith(RETURN_PREFIX) || mock.startsWith(THROW_PREFIX)) {
        mock = mock.replace('`', '"');
    }
    return mock;
}

我们获取到mock的返回值内容后需要把该返回值包装成方法返回值类型，所以这里必须要有一个返回值类型的包装，我们看一下parseMockValue方法：

public static Object parseMockValue(String mock, Type[] returnTypes) throws Exception {
    Object value = null;
    if ("empty".equals(mock)) {
        value = ReflectUtils.getEmptyObject(returnTypes != null && returnTypes.length > 0 ?
                                            (Class<?>) returnTypes[0] : null);
    } else if ("null".equals(mock)) {
        value = null;
    } else if ("true".equals(mock)) {
        value = true;
    } else if ("false".equals(mock)) {
        value = false;
    } else if (mock.length() >= 2 && (mock.startsWith("\"") && mock.endsWith("\"")
                                      || mock.startsWith("\'") && mock.endsWith("\'"))) {
        value = mock.subSequence(1, mock.length() - 1);
    } else if (returnTypes != null && returnTypes.length > 0 && returnTypes[0] ==
               String.class) {
        value = mock;
    } else if (StringUtils.isNumeric(mock, false)) {
        value = JSON.parse(mock);
    } else if (mock.startsWith("{")) {
        value = JSON.parseObject(mock, Map.class);
    } else if (mock.startsWith("[")) {
        value = JSON.parseObject(mock, List.class);
    } else {
        value = mock;
    }
    if (ArrayUtils.isNotEmpty(returnTypes)) {
        value = PojoUtils.realize(value, (Class<?>) returnTypes[0], returnTypes.length > 1
                                  ? returnTypes[1] : null);
    }
    return value;
}

从上面的逻辑来看，如果是force:return 则会不走rpc直接返回一个结果，然后把这个结果包装成方法的返回值类型。