负载均衡机制

深入理解Dubbo负载均衡的实现原理，掌握随机、轮询、一致性Hash、最少活跃调用等负载均衡算法，理解路由规则、集群容错的实现机制。

📋 负载均衡概览

负载均衡是分布式系统中的核心功能，Dubbo提供了多种负载均衡算法和集群容错策略。

🎯 一、负载均衡接口

1.1 LoadBalance接口

LoadBalance是负载均衡的核心接口：

java

@SPI(RandomLoadBalance.NAME)
public interface LoadBalance {
    @Adaptive("loadbalance")
    <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) 
            throws RpcException;
}

核心方法：

select：从多个Invoker中选择一个进行调用

SPI机制：

@SPI：指定默认实现为RandomLoadBalance
@Adaptive：通过URL参数loadbalance动态选择实现

1.2 负载均衡算法

Dubbo提供了以下负载均衡算法：

RandomLoadBalance：随机算法（默认）
RoundRobinLoadBalance：加权轮询算法
LeastActiveLoadBalance：最少活跃调用算法
ConsistentHashLoadBalance：一致性Hash算法

🎯 二、随机算法

2.1 RandomLoadBalance

随机算法是Dubbo的默认负载均衡算法，简单高效：

java

public class RandomLoadBalance extends AbstractLoadBalance {
    public static final String NAME = "random";
    
    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, 
            Invocation invocation) {
        int length = invokers.size();
        boolean sameWeight = true;
        int[] weights = new int[length];
        int firstWeight = getWeight(invokers.get(0), invocation);
        weights[0] = firstWeight;
        int totalWeight = firstWeight;
        
        for (int i = 1; i < length; i++) {
            int weight = getWeight(invokers.get(i), invocation);
            weights[i] = weight;
            totalWeight += weight;
            if (sameWeight && weight != firstWeight) {
                sameWeight = false;
            }
        }
        
        if (totalWeight > 0 && !sameWeight) {
            int offset = ThreadLocalRandom.current().nextInt(totalWeight);
            for (int i = 0; i < length; i++) {
                offset -= weights[i];
                if (offset < 0) {
                    return invokers.get(i);
                }
            }
        }
        
        return invokers.get(ThreadLocalRandom.current().nextInt(length));
    }
}

算法特点：

支持权重
随机选择，避免热点问题
时间复杂度O(n)

🎯 三、轮询算法

3.1 RoundRobinLoadBalance

加权轮询算法根据权重进行轮询选择：

java

public class RoundRobinLoadBalance extends AbstractLoadBalance {
    public static final String NAME = "roundrobin";
    
    private static final int RECYCLE_PERIOD = 60000;
    
    protected static class WeightedRoundRobin {
        private int weight;
        private AtomicLong current = new AtomicLong(0);
        private long lastUpdate;
        
        public int getWeight() {
            return weight;
        }
        
        public void setWeight(int weight) {
            this.weight = weight;
            current.set(0);
        }
        
        public long increaseCurrent() {
            return current.addAndGet(weight);
        }
        
        public void sel(int total) {
            current.addAndGet(-1 * total);
        }
        
        public long getLastUpdate() {
            return lastUpdate;
        }
        
        public void setLastUpdate(long lastUpdate) {
            this.lastUpdate = lastUpdate;
        }
    }
    
    @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.get(key);
        if (map == null) {
            methodWeightMap.putIfAbsent(key, new ConcurrentHashMap<String, WeightedRoundRobin>());
            map = methodWeightMap.get(key);
        }
        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();
            WeightedRoundRobin weightedRoundRobin = map.get(identifyString);
            int weight = getWeight(invoker, invocation);
            if (weight < 0) {
                weight = 0;
            }
            if (weightedRoundRobin == null) {
                weightedRoundRobin = new WeightedRoundRobin();
                weightedRoundRobin.setWeight(weight);
                map.putIfAbsent(identifyString, weightedRoundRobin);
            }
            if (weight != weightedRoundRobin.getWeight()) {
                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;
        }
        return invokers.get(0);
    }
}

算法特点：

支持权重
平滑轮询
时间复杂度O(n)

🎯 四、一致性Hash算法

4.1 ConsistentHashLoadBalance

一致性Hash算法根据请求的参数进行Hash，保证相同参数的请求总是路由到同一提供者：

java

public class ConsistentHashLoadBalance extends AbstractLoadBalance {
    public static final String NAME = "consistenthash";
    
    private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = 
            new ConcurrentHashMap<String, ConsistentHashSelector<?>>();
    
    @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;
        int identityHashCode = System.identityHashCode(invokers);
        ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) 
                selectors.get(key);
        if (selector == null || selector.identityHashCode != identityHashCode) {
            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> {
        private final TreeMap<Long, Invoker<T>> virtualInvokers;
        private final int replicaNumber;
        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);
            String[] index = 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]);
            }
            for (Invoker<T> invoker : invokers) {
                String address = invoker.getUrl().getAddress();
                for (int i = 0; i < replicaNumber / 4; i++) {
                    byte[] digest = md5(address + i);
                    for (int h = 0; h < 4; h++) {
                        long m = hash(digest, h);
                        virtualInvokers.put(m, invoker);
                    }
                }
            }
        }
        
        public Invoker<T> select(Invocation invocation) {
            String key = toKey(invocation.getArguments());
            byte[] digest = md5(key);
            return selectForKey(hash(digest, 0));
        }
        
        private Invoker<T> selectForKey(long hash) {
            Map.Entry<Long, Invoker<T>> entry = virtualInvokers.ceilingEntry(hash);
            if (entry == null) {
                entry = virtualInvokers.firstEntry();
            }
            return entry.getValue();
        }
    }
}

算法特点：

保证相同参数的请求路由到同一提供者
支持虚拟节点
适用于有状态服务

🎯 五、最少活跃调用算法

5.1 LeastActiveLoadBalance

最少活跃调用算法优先选择活跃数最少的Invoker：

java

public class LeastActiveLoadBalance extends AbstractLoadBalance {
    public static final String NAME = "leastactive";
    
    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, 
            Invocation invocation) {
        int length = invokers.size();
        int leastActive = -1;
        int leastCount = 0;
        int[] leastIndexes = new int[length];
        int[] weights = new int[length];
        int totalWeight = 0;
        int firstWeight = 0;
        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 weight = invoker.getUrl().getMethodParameter(
                    invocation.getMethodName(), WEIGHT_KEY, DEFAULT_WEIGHT);
            weights[i] = weight;
            
            if (leastActive == -1 || active < leastActive) {
                leastActive = active;
                leastCount = 1;
                leastIndexes[0] = i;
                totalWeight = weight;
                firstWeight = weight;
                sameWeight = true;
            } else if (active == leastActive) {
                leastIndexes[leastCount++] = i;
                totalWeight += weight;
                if (sameWeight && i > 0 && weight != firstWeight) {
                    sameWeight = false;
                }
            }
        }
        
        if (leastCount == 1) {
            return invokers.get(leastIndexes[0]);
        }
        if (!sameWeight && totalWeight > 0) {
            int offsetWeight = ThreadLocalRandom.current().nextInt(totalWeight);
            for (int i = 0; i < leastCount; i++) {
                int leastIndex = leastIndexes[i];
                offsetWeight -= weights[leastIndex];
                if (offsetWeight < 0) {
                    return invokers.get(leastIndex);
                }
            }
        }
        
        return invokers.get(leastIndexes[ThreadLocalRandom.current().nextInt(leastCount)]);
    }
}

算法特点：

优先选择活跃数最少的提供者
支持权重
适用于慢速提供者的场景

🎯 六、集群容错

6.1 Cluster接口

Cluster定义了集群容错的接口：

java

@SPI(FailoverCluster.NAME)
public interface Cluster {
    @Adaptive
    <T> Invoker<T> join(Directory<T> directory) throws RpcException;
}

6.2 容错策略

Dubbo提供了以下容错策略：

FailoverCluster：失败自动切换（默认）
FailfastCluster：快速失败
FailsafeCluster：失败安全
FailbackCluster：失败自动恢复
ForkingCluster：并行调用
BroadcastCluster：广播调用

6.3 FailoverCluster

失败自动切换是默认的容错策略，调用失败后会自动切换到其他提供者：

java

public class FailoverCluster implements Cluster {
    public final static String NAME = "failover";
    
    @Override
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        return new FailoverClusterInvoker<T>(directory);
    }
}

📖 学习资源

视频教程

书籍推荐

《深度剖析Apache Dubbo核心技术内幕》 - 华仔
《分布式系统常用技术及案例分析》 - 柳伟卫

网站资源

✅ 学习检查清单

[ ] 已理解负载均衡接口
[ ] 已理解随机算法
[ ] 已理解轮询算法
[ ] 已理解一致性Hash算法
[ ] 已理解最少活跃调用算法
[ ] 已理解集群容错策略
[ ] 已绘制负载均衡流程图
[ ] 已总结核心知识点

上一章节：服务注册与发现
下一章节：容错机制
返回目录：Dubbo源码深度解析

开发与调试

测试能力

运维与监控

NoSQL

中间件

存储

接入层

操作系统

框架

硬件知识

缓存

品质

经验底蕴

能力

任务拆分

文档能力

协议设计

基本理论

接入层架构

数据层架构

通讯机制

逻辑层架构

负载均衡机制 ​

📋 负载均衡概览 ​

🎯 一、负载均衡接口 ​

1.1 LoadBalance接口 ​

1.2 负载均衡算法 ​

🎯 二、随机算法 ​

2.1 RandomLoadBalance ​

🎯 三、轮询算法 ​

3.1 RoundRobinLoadBalance ​

🎯 四、一致性Hash算法 ​

4.1 ConsistentHashLoadBalance ​

🎯 五、最少活跃调用算法 ​

5.1 LeastActiveLoadBalance ​

🎯 六、集群容错 ​

6.1 Cluster接口 ​

6.2 容错策略 ​

6.3 FailoverCluster ​

📖 学习资源 ​

视频教程 ​

书籍推荐 ​

网站资源 ​

✅ 学习检查清单 ​

负载均衡机制

📋 负载均衡概览

🎯 一、负载均衡接口

1.1 LoadBalance接口

1.2 负载均衡算法

🎯 二、随机算法

2.1 RandomLoadBalance

🎯 三、轮询算法

3.1 RoundRobinLoadBalance

🎯 四、一致性Hash算法

4.1 ConsistentHashLoadBalance

🎯 五、最少活跃调用算法

5.1 LeastActiveLoadBalance

🎯 六、集群容错

6.1 Cluster接口

6.2 容错策略

6.3 FailoverCluster

📖 学习资源

视频教程

书籍推荐

网站资源

✅ 学习检查清单