Robin六种常用负载均衡算法源码解析

文章目录

1 经典轮询算法
2 随机算法
3 以响应时间为权重的轮询策略（重中之重）
4 重试策略
5 断言策略
6 最佳可用性策略

1 经典轮询算法

//Robin的负载均衡原理为 请求服务=请求次数%服务个数。//请求计数器
private AtomicInteger nextServerCyclicCounter;
public Server choose(ILoadBalancer lb, Object key) {if (lb == null) {log.warn("no load balancer");return null;}Server server = null;//这里的count指的是重试次数，while循环中count<10，//说明默认重试次数为10次int count = 0;while (server == null && count++ < 10) {List<Server> reachableServers = lb.getReachableServers();List<Server> allServers = lb.getAllServers();int upCount = reachableServers.size();int serverCount = allServers.size();if ((upCount == 0) || (serverCount == 0)) {log.warn("No up servers available from load balancer: " + lb);return null;}//重点来了，我们来看一下incrementAndGetModulo这个方法。int nextServerIndex = incrementAndGetModulo(serverCount);server = allServers.get(nextServerIndex);if (server == null) {/* Transient. */Thread.yield();continue;}if (server.isAlive() && (server.isReadyToServe())) {return (server);}// Next.server = null;}if (count >= 10) {log.warn("No available alive servers after 10 tries from load balancer: "+ lb);}return server;}   //获取服务索引的方法private int incrementAndGetModulo(int modulo) {//不难看出，这里采用乐观锁也就是自旋+cas的方法来获取服务索引。for (; ; ) {int current = nextServerCyclicCounter.get();int next = (current + 1) % modulo;if (nextServerCyclicCounter.compareAndSet(current, next))return next;}}

2 随机算法

public Server choose(ILoadBalancer lb, Object key) {if (lb == null) {return null;}Server server = null;while (server == null) {if (Thread.interrupted()) {return null;}List<Server> upList = lb.getReachableServers();List<Server> allList = lb.getAllServers();int serverCount = allList.size();if (serverCount == 0) {return null;}int index = chooseRandomInt(serverCount);server = upList.get(index);if (server == null) {//下面的代码只会在该服务由于某种原因被下线时才会执行,//这种情况只是暂时的，ribon会重新进行负载均衡。Thread.yield();continue;}if (server.isAlive()) {return (server);}// 下面的代码正常状态下不会执行。server = null;Thread.yield();}return server;}//具体的负载均衡方法调用了线程本地随机数api为当前线程生成随机数。protected int chooseRandomInt(int serverCount) {return ThreadLocalRandom.current().nextInt(serverCount);}

3 以响应时间为权重的轮询策略（重中之重）

 //该算法的核心思想：使用平均百分位响应时间的规则为每个服务器分配动态权重值，如果权重值没有初始化，退而使用轮询的方式进行负载均衡。//举个例子，假如有四个服务a,b,c,d,权重值分别为 10，20，30，40 ，这样权重列表值为，10，30，60，100，获取一个随机值（>=0&&<100）为25，权重列表中最小的大于随机权重值的索引既是选择的服务节点。// 保存索引0到当前索引的累积权重，例如该list索引2位置的变量保存的就是    服务节点0到服务节点2 的权重值。private volatile List<Double> accumulatedWeights = new ArrayList<Double>();
public Server choose(ILoadBalancer lb, Object key) {if (lb == null) {return null;}Server server = null;while (server == null) {//获取当前引用，以防它被其他线程更改。List<Double> currentWeights = accumulatedWeights;if (Thread.interrupted()) {return null;}List<Server> allList = lb.getAllServers();int serverCount = allList.size();if (serverCount == 0) {return null;}int serverIndex = 0;// 获取总的权重值，也就是currentWeights最后一个服务节点的权重值double maxTotalWeight = currentWeights.size() == 0 ? 0 : currentWeights.get(currentWeights.size() - 1); // 如果没有命中任何服务，此时总权重值还没有初始化，// 退而求其次，使用轮询算法进行负载均衡.if (maxTotalWeight < 0.001d || serverCount != currentWeights.size()) {server =  super.choose(getLoadBalancer(), key);if(server == null) {return server;}} else {//生成范围在[0,maxTotalWeight)的随机权重值。double randomWeight = random.nextDouble() * maxTotalWeight;// 根据随机权重值选择服务。遍历权重集合，//选择大于随即权重值的最小的服务索引.int n = 0;for (Double d : currentWeights) {if (d >= randomWeight) {serverIndex = n;break;} else {n++;}}server = allList.get(serverIndex);}if (server == null) {/* Transient. */Thread.yield();continue;}if (server.isAlive()) {return (server);}// Next.server = null;}return server;}//动态服务权重任务，继承了jdk原生定时任务，作用是定时的获取响应时间。class DynamicServerWeightTask extends TimerTask {public void run() {ServerWeight serverWeight = new ServerWeight();try {serverWeight.maintainWeights();} catch (Exception e) {logger.error("Error running DynamicServerWeightTask for {}", name, e);}}}class ServerWeight {public void maintainWeights() {ILoadBalancer lb = getLoadBalancer();if (lb == null) {return;}//如果已经serverWeightAssignmentInProgress已经被某个线程修改，直接退出任务体。if (!serverWeightAssignmentInProgress.compareAndSet(false,  true))  {return; }try {logger.info("Weight adjusting job started");AbstractLoadBalancer nlb = (AbstractLoadBalancer) lb;LoadBalancerStats stats = nlb.getLoadBalancerStats();if (stats == null) {// 如果获取不到运行时信息，直接退出任务体return;}//所有服务的总响应时间。double totalResponseTime = 0;// 找到最大为95%的响应时间for (Server server : nlb.getAllServers()) {// 如果缓存中没有的话就会自动获取平均响应时间ServerStats ss = stats.getSingleServerStat(server);totalResponseTime += ss.getResponseTimeAvg();}// 每个服务器的权重为(所有服务器的响应时间之和-响应时间)// 因此，响应时间越长，权重就越小，被选中的可能性就越小Double weightSoFar = 0.0;// 创建新的列表并热交换引用List<Double> finalWeights = new ArrayList<Double>();for (Server server : nlb.getAllServers()) {ServerStats ss = stats.getSingleServerStat(server);//每个节点的权重值double weight = totalResponseTime - ss.getResponseTimeAvg();//累加从0 到 索引值的权重值weightSoFar += weight;finalWeights.add(weightSoFar);   }setWeights(finalWeights);} catch (Exception e) {logger.error("Error calculating server weights", e);} finally {serverWeightAssignmentInProgress.set(false);}}}

4 重试策略

这个策略主要用于给其他策略添加重试逻辑，另外，负载均衡策略是级联的。

 //该方法必要时会循环。注意，可能会超时，这取决于子策略，因为我们不会产生额外的线程并提前返回。 public Server choose(ILoadBalancer lb, Object key) {long requestTime = System.currentTimeMillis();// 这个最大重试时间默认为500ms,可以通过set 赋值。long deadline = requestTime + maxRetryMillis;Server answer = null;// 在这里可以看到调用了子策略的选择方法answer = subRule.choose(key);// 如果没有选择出可用的server，并且还没有到达最大可重试时间，// 会开启一个可中断定时任务去重新尝试获取server,如果在再次失败，// 会调用 yield（）让出时间片，让其他线程先执行。if (((answer == null) || (!answer.isAlive()))&& (System.currentTimeMillis() < deadline)) {InterruptTask task = new InterruptTask(deadline- System.currentTimeMillis());while (!Thread.interrupted()) {answer = subRule.choose(key);if (((answer == null) || (!answer.isAlive()))&& (System.currentTimeMillis() < deadline)) {// 暂停并重试，希望它是短暂的Thread.yield();} else {break;}}//最终将该定时任务销毁。task.cancel();}if ((answer == null) || (!answer.isAlive())) {return null;} else {return answer;}}

5 断言策略

可能大家会对断言有一点陌生，这里简单介绍一下.
predicate 在java8 被引入，Predicate接口主要用来判断一个参数是否符合要求。这个predicate 通过分配给他的lamda 表达式来判断参数是否符合要求。
下面具体介绍断言策略。
这个策略将服务器过滤逻辑委托给{@link AbstractServerPredicate}实例。
过滤之后，服务器以轮询的方式从过滤列表中返回。

    public Server choose(Object key) {ILoadBalancer lb = getLoadBalancer();// 这里就很好理解了，断言规则过滤完后使用轮询规则从获取到的服务列表中选择一个server 作为结果返回。Optional<Server> server = getPredicate().chooseRoundRobinAfterFiltering(lb.getAllServers(), key);if (server.isPresent()) {return server.get();} else {return null;}       }

很明显，这里的AbstractServerPredicate 是重中之重，我们继续追踪源码。

     /在断言筛选给定的服务器列表和负载平衡器key之后，以轮询方式选择服务器/public Optional<Server> chooseRoundRobinAfterFiltering(List<Server> servers, Object loadBalancerKey) {// 接着来看 getEligibleServersList<Server> eligible = getEligibleServers(servers, loadBalancerKey);if (eligible.size() == 0) {return Optional.absent();}return Optional.of(eligible.get(incrementAndGetModulo(eligible.size())));}/*** 从服务器列表中获取通过此断言过滤的服务器*/public List<Server> getEligibleServers(List<Server> servers, Object loadBalancerKey) {// 如果loadBalancerKey 为空，只需要使用 ServerOnlyPredicate  来过滤 servers就行。我们继续来看 ServerOnlyPredicate  if (loadBalancerKey == null) {return ImmutableList.copyOf(Iterables.filter(servers, this.getServerOnlyPredicate()));            } else {// 否则，使用正常的PredicateKey 来进行过滤。List<Server> results = Lists.newArrayList();for (Server server: servers) {if (this.apply(new PredicateKey(loadBalancerKey, server))) {results.add(server);}}return results;            }}//这里采用使用了子类（也就是具体的断言）来过滤。private final Predicate<Server> serverOnlyPredicate =  new Predicate<Server>() {@Overridepublic boolean apply(@Nullable Server input) {                    return AbstractServerPredicate.this.apply(new PredicateKey(input));}};

6 最佳可用性策略

这种策略会跳过带有断路器的服务器，选择并发请求最低的服务器。
该规则通常应该与{@link ServerListSubsetFilter}一起工作，它对规则可见的服务器进行了限制。
这确保它只需要在少量服务器中找到最小的并发请求。同时，每个客户都会得到一个随机列表
避免了并发请求最少的服务器被大量的客户选择，并被立即压垮的问题。

     public Server choose(Object key) {// 如果负载均衡器状态为空，退而使用轮询策略。if (loadBalancerStats == null) {return super.choose(key);}List<Server> serverList = getLoadBalancer().getAllServers();int minimalConcurrentConnections = Integer.MAX_VALUE;long currentTime = System.currentTimeMillis();Server chosen = null;for (Server server: serverList) {// 获取当前服务器状态ServerStats serverStats = loadBalancerStats.getSingleServerStat(server);//跳过带有断路器的服务器if (!serverStats.isCircuitBreakerTripped(currentTime)) {//获取并发访问量。int concurrentConnections = serverStats.getActiveRequestsCount(currentTime);//这一块逻辑很简单，就是找出所有服务器中 并发请求最低的服务器。if (concurrentConnections < minimalConcurrentConnections) {minimalConcurrentConnections = concurrentConnections;chosen = server;}}}// 如果选中服务器为空，退而使用轮询策略。if (chosen == null) {return super.choose(key);} else {return chosen;}}

Robin六种常用负载均衡算法源码解析相关推荐

【详解】Ribbon 负载均衡服务调用原理及默认轮询负载均衡算法源码解析、手写
Ribbon 负载均衡服务调用一.什么是 Ribbon 二.LB负载均衡(Load Balancer)是什么 1.Ribbon 本地负载均衡客户端 VS Nginx 服务端负载均衡的区别 2.LB负 ...
RocketMQ Consumer 负载均衡算法源码学习 -- AllocateMessageQueueConsistentHash
RocketMQ 提供了一致性hash 算法来做Consumer 和 MessageQueue的负载均衡. 源码中一致性hash 环的实现是很优秀的,我们一步一步分析. 一个Hash环包含多个节点, ...
Dubbo负载均衡的源码流程(2022.5.30)
Dubbo负载均衡的源码流程 1.默认负载均衡策略:RandomLoadBalance(随机策略) 2.负载均衡策略存在以下五种: 2.1 RandomLoadBalance(随机) 2.2 Roun ...
java 一致性hash算法均衡分发_Dubbo一致性哈希负载均衡的源码和Bug，了解一下？...
本文是对于Dubbo负载均衡策略之一的一致性哈希负载均衡的详细分析.对源码逐行解读.根据实际运行结果,配以丰富的图片,可能是东半球讲一致性哈希算法在Dubbo中的实现最详细的文章了. 文中所示源码,没 ...
Java 常用负载均衡算法解析
前言负载均衡在Java领域中有着广泛深入的应用,不管是大名鼎鼎的nginx,还是微服务治理组件如dubbo,feign等,负载均衡的算法在其中都有着实际的使用负载均衡的核心思想在于其底层的算法思想 ...
@cacheable 服务器不一致_Dubbo一致性哈希负载均衡的源码和Bug，了解一下？
持续输出原创文章,点击蓝字关注我吧本文是对于Dubbo负载均衡策略之一的一致性哈希负载均衡的详细分析.对源码逐行解读.根据实际运行结果,配以丰富的图片,可能是东半球讲一致性哈希算法在Dubbo中的实 ...
以太坊Geth 共识算法源码解析
共识算法目前以太坊中有两个公式算法的实现,分别为clique和ethash.其中clique是PoA共识的实现,ethash是PoW共识的实现,其相应的代码位于go-ethereum/consens ...
ZooKeeper的FastLeaderElection算法源码解析
Zookeeper服务器在启动的时候会通过一定的选举算法从多个server中选出leader server,剩下的server则作为follower.目前实现的选举算法有FastLeaderElect ...
Spring源码深度解析(郝佳)-Spring 常用注解使用及源码解析
我们在看Spring Boot源码时,经常会看到一些配置类中使用了注解,本身配置类的逻辑就比较复杂了,再加上一些注解在里面,让我们阅读源码更加难解释了,因此,这篇博客主要对配置类上的一些注解的使用 ...