Ribbon 负载均衡策略 —— 图解、源码级解析

喜欢ヅ旅行 2024-05-24 21:38 101阅读 0赞

> ? Java学习：[社区快速通道][Link 1]  
>   
> 
> 
> ? 深入浅出RocketMQ设计思想：[深入浅出RocketMQ设计思想][RocketMQ]  
>   
> 
> 
> ? 绝对不一样的职场干货：[大厂最佳实践经验指南][Link 2]  
>   
>   
> ? 最近更新：2023年6月4日  
>   
> 
> 
>   
> 
> 
> ? 点赞 ? 收藏 ⭐留言 ? 都是我最大的动力！

--------------------

#### 文章目录 ####

*  负载均衡策略
 *   *  RandomRule
     *  RoundRobinRule
     *  RetryRule
     *  WeightedResponseTimeRule
     *  BestAvailableRule
     *  AvailabilityFilteringRule
     *  ZoneAvoidanceRule
 *  Ribbon 负载均衡策略源码
 *   *  RandomRule源码
     *  RoundRobinRule源码
     *  BestAvailableRule源码
     *  RetryRule源码

**通过本文你可以学习到：**

1.  常见的7种负载均衡策略思想
2.  自旋锁的使用方式
3.  防御性编程

## 负载均衡策略 ##

### RandomRule ###

该策略会从当前可用的服务节点中，随机挑选一个节点访问，使用了yield+自旋的方式做重试，还采用了严格的防御性编程。

### RoundRobinRule ###

该策略会从一个节点一步一步地向后选取节点，如下图所示：  
![在这里插入图片描述][watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5bCP546L5pu-5piv5bCR5bm0_size_8_color_FFFFFF_t_70_g_se_x_16_pic_center]  
在多线程环境下，两个请求同时访问这个Rule也不会读取到相同节点：这靠的是RandomRobinRule底层的自旋锁+CAS的同步操作。

CAS+自旋锁这套组合技是高并发下最廉价的线程安全手段，因为这套操作不需要锁定系统资源。但缺点是，自旋锁如果迟迟不能释放，将会带来CPU资源的浪费，因为自旋本身并不会执行任何业务逻辑，而是单纯的使CPU空转。所以通常情况下会对自旋锁的旋转次数做一个限制，比如JDK中`synchronize`底层的锁升级策略，就对自旋次数做了动态调整。

while (true) {
        
        // cas操作
        if (cas(expected, update)) {
        
            // 业务逻辑代码
            // break或退出return
        }
    }

> Eureka为了防止服务下线被重复调用，就使用AtomicBoolean的CAS方法做同步控制；

> 奈飞提供的SpringCloud组件有特别多用到CAS的地方，感兴趣的小伙伴们可以发现一下

### RetryRule ###

RetryRule是一个类似装饰器模式的规则，装饰器相当于一层套一层的套娃，每一层都会加上一层独特的功能。

**经典的装饰器模式示意图：**  
![在这里插入图片描述][watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5bCP546L5pu-5piv5bCR5bm0_size_14_color_FFFFFF_t_70_g_se_x_16_pic_center]  
借助上面的思路，`RetryRule`就是给其他负载均衡策略加上重试功能。在`RetryRule`里还藏着一个`subRule`，这才是真正被执行的负载均衡策略，`RetryRule`正是要为它添加重试功能（如果初始化时没指定`subRule`，将默认使用`RoundRibinRule`）。

### WeightedResponseTimeRule ###

这个规则继承自`RoundRibbonRule`，他会根据服务节点的响应时间计算权重，响应时间越长权重就越低，响应越快则权重越高，权重的高低决定了机器被选中概率的高低。也就是说，响应时间越小的机器，被选中的概率越大。

> 服务器刚启动的时候，对各个服务节点采样不足，因此会采用轮询策略，当积累到一定的样本时候，才会切换到`WeightedResponseTimeRule`模式。

### BestAvailableRule ###

在过滤掉故障服务以后，它会基于过去30分钟的统计结果选取当前并发量最小的服务节点作为目标地址。如果统计结果尚未生成，则采用轮询的方式选定节点。

### AvailabilityFilteringRule ###

这个规则底层依赖`RandomRobinRule`来选取节点，但必须要满足它的最低要求的节点才会被选中。如果节点满足了要求，无论其响应时间或者当前并发量是什么，都会被选中。

每次`AvailabilityFilteringRule`都会请求`RobinRule`挑选一个节点，然后对这个节点做以下两步检查：

1.  是否处于熔断状态
2.  节点当前的请求连接数超过阈值，超过了则表示节点目前太忙

如果被选中的`server`挂了，那么AFR会自动重试（最多10次），让`RobinRule`重新选择一个服务节点

### ZoneAvoidanceRule ###

这个过滤器包含了组合过滤条件，分别是Zone级别和可用性级别。  
![在这里插入图片描述][watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5bCP546L5pu-5piv5bCR5bm0_size_13_color_FFFFFF_t_70_g_se_x_16_pic_center]

*  **Zone Filter：** Zone可以理解为机房所属的大区域，这里会对这个Zone下面所有的服务节点进行健康情况过滤。
 *  **可用性过滤：** 这里和`AvailabilityFilteringRule`的验证过程很像，会过滤掉当前并发量较大，或者处于熔断状态的服务节点。

## Ribbon 负载均衡策略源码 ##

### RandomRule源码 ###

先从`RandomRule`看起，核心的方法是：  
![请添加图片描述][f797f22e59e544c68be41fa694aa64bd.png]

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) {
        
                /*
                 * No servers. End regardless of pass, because subsequent passes
                 * only get more restrictive.
                 */
                return null;
            }
    
            int index = chooseRandomInt(serverCount);
            server = upList.get(index);
    
            if (server == null) {
        
                Thread.yield();
                continue;
            }
    
            if (server.isAlive()) {
        
                return (server);
            }
    
            server = null;
            Thread.yield();
        }
    
        return server;
    
    }

在`RandomRule`里方法的入参`key`没有用到，所以可以先暂时忽略

`while`循环逻辑是如果`server`为空，则找到一个可用的`server`

if (Thread.interrupted()) {
        
        return null;
    }

如果线程暂停了，则直接返回空（防御性编程）

List<Server> upList = lb.getReachableServers();
    List<Server> allList = lb.getAllServers();

`allList`存储的是所有的服务，`upList`存储的是可运行状态的服务

int serverCount = allList.size();
    if (serverCount == 0) {
        
        return null;
    }

服务中心上没有`server`注册，则返回空

int index = chooseRandomInt(serverCount);
    server = upList.get(index);

随机选择一个server

**其中，chooseRandomInt的逻辑如下：**

protected int chooseRandomInt(int serverCount) {
        
        return ThreadLocalRandom.current().nextInt(serverCount);
    }

返回0到`serverCount`中间的任意一个值

> java中的随机是可以预测到结果的，真随机数一般会掺杂一些不可预测的数据，比如当前cpu的温度

--------------------

**回到`RandomRule`的`choose`方法：**

如果发现随机选择的`server`为空表示此时`serverList`正在被修正，此时让出线程资源，进行下一次循环，对应最开始的防御性编程

if (server == null) {
        
        Thread.yield();
        continue;
    }

if (server.isAlive()) {
        
        return (server);
    }

如果`server`可用直接`return`

server = null;
    Thread.yield();

如果不可用则`server`置为空，下一次循环会选一个新的，最后让出资源。

所以该方法每次进入下一次循环时都会让出线程。

--------------------

### RoundRobinRule源码 ###

接下来看`RoundRobinRule`

public Server choose(ILoadBalancer lb, Object key) {
        
        if (lb == null) {
        
            log.warn("no load balancer");
            return null;
        }
    
        Server server = null;
        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;
            }
    
            int nextServerIndex = incrementAndGetModulo(serverCount);
            server = allServers.get(nextServerIndex);
    
            if (server == null) {
        
                Thread.yield();
                continue;
            }
    
            if (server.isAlive() && (server.isReadyToServe())) {
        
                return (server);
            }
    
            server = null;
        }
    
        if (count >= 10) {
        
            log.warn("No available alive servers after 10 tries from load balancer: "
                    + lb);
        }
        return server;
    }

`while`循环里面有一个计数器，如果重试10次依然没有结果返回就不重试了。

List<Server> reachableServers = lb.getReachableServers();
    List<Server> allServers = lb.getAllServers();
    int upCount = reachableServers.size();
    int serverCount = allServers.size();

`reachableServers`就是`up`状态的`server`

if ((upCount == 0) || (serverCount == 0)) {
        
        log.warn("No up servers available from load balancer: " + lb);
        return null;
    }

没有可用服务器则返回空

int nextServerIndex = incrementAndGetModulo(serverCount);
    server = allServers.get(nextServerIndex);

选择哪个下标的`server`，进入`incrementAndGetModulo`方法

private int incrementAndGetModulo(int modulo) {
        
        for (;;) {
        
            int current = nextServerCyclicCounter.get();
            int next = (current + 1) % modulo;
            if (nextServerCyclicCounter.compareAndSet(current, next))
                return next;
        }
    }

使用了自旋锁，`nextServerCyclicCounter`是一个线程安全的数字。

if (server == null) {
        
        Thread.yield();
        continue;
    }

如果获取到的`server`为空则让出资源，继续下一次循环

if (server.isAlive() && (server.isReadyToServe())) {
        
        return (server);
    }

`server`是正常的则返回

server = null;

最后没有让出线程资源，因为重试10次后就退出循环了

--------------------

### BestAvailableRule源码 ###

接下来看`BestAvailableRule`

@Override
    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;
        }
    }

if (loadBalancerStats == null) {
        
        return super.choose(key);
    }

如果`loadBalancerStats`为空则调用父类的`choose`方法，父类方法直接委托给`RoundRobinRule`来完成`choose`。

`for`循环里先从`loadBalancerStats`中获取到当前服务的状态

ServerStats serverStats = loadBalancerStats.getSingleServerStat(server);

public ServerStats getSingleServerStat(Server server) {
        
        return getServerStats(server);
    }

protected ServerStats getServerStats(Server server) {
        
        try {
        
            return serverStatsCache.get(server);
        } catch (ExecutionException e) {
        
            ServerStats stats = createServerStats(server);
            serverStatsCache.asMap().putIfAbsent(server, stats);
            return serverStatsCache.asMap().get(server);
        }
    }

这里是从缓存中获取`server的stats`，如果获取失败则默认创建一个`stats`并添加到缓存中，然后从`cache`中再获取一次。

随后判断是否处于熔断状态

if (!serverStats.isCircuitBreakerTripped(currentTime)) {
        ...}

public boolean isCircuitBreakerTripped(long currentTime) {
        
        long circuitBreakerTimeout = getCircuitBreakerTimeout();
        if (circuitBreakerTimeout <= 0) {
        
            return false;
        }
        return circuitBreakerTimeout > currentTime;
    }

首先获得熔断的`TimeOut`（表示截止到未来某个时间熔断终止），如果大于当前时间说明处于熔断状态。

熔断的`TimeOut`由下面方法计算得到：

private long getCircuitBreakerTimeout() {
        
        long blackOutPeriod = getCircuitBreakerBlackoutPeriod();
        if (blackOutPeriod <= 0) {
        
            return 0;
        }
        return lastConnectionFailedTimestamp + blackOutPeriod;
    }

返回上一次连接失败的时间戳 + blackOutPeriod

其中又调用了

private long getCircuitBreakerBlackoutPeriod() {
        
        int failureCount = successiveConnectionFailureCount.get();
        int threshold = connectionFailureThreshold.get();
        if (failureCount < threshold) {
        
            return 0;
        }
        int diff = (failureCount - threshold) > 16 ? 16 : (failureCount - threshold);
        int blackOutSeconds = (1 << diff) * circuitTrippedTimeoutFactor.get();
        if (blackOutSeconds > maxCircuitTrippedTimeout.get()) {
        
            blackOutSeconds = maxCircuitTrippedTimeout.get();
        }
        return blackOutSeconds * 1000L;
    }

`failureCount`是失败的个数，从一个计数器里获得，阈值从一个缓存的属性中获得，之后计算两个的差值，再根据缓存中的一些属性计算最终的秒数，最后乘以1000返回。

--------------------

回到`BestAvailableRule`的`choose`方法，只有不处于熔断状态才能继续走后面的流程

if (concurrentConnections < minimalConcurrentConnections) {
        
        minimalConcurrentConnections = concurrentConnections;
        chosen = server;
    }

选出连接数最小的服务器

if (chosen == null) {
        
        return super.choose(key);
    } else {
        
        return chosen;
    }

最后返回

核心是找到一个最轻松的服务器。

--------------------

### RetryRule源码 ###

查看`RetryRule`源码：

public Server choose(ILoadBalancer lb, Object key) {
        
       long requestTime = System.currentTimeMillis();
       long deadline = requestTime + maxRetryMillis;
    
       Server answer = null;
    
       answer = subRule.choose(key);
    
       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)) {
        
                /* pause and retry hoping it's transient */
                Thread.yield();
             } else {
        
                break;
             }
          }
    
          task.cancel();
       }
    
       if ((answer == null) || (!answer.isAlive())) {
        
          return null;
       } else {
        
          return answer;
       }
    }

long requestTime = System.currentTimeMillis();
    long deadline = requestTime + maxRetryMillis;

先记录当前时间和`deadline`，在截止时间之前可以一直重试。

answer = subRule.choose(key);

方法里面是由`subRule`来实现具体的负载均衡逻辑，这里默认类型是`RoundRobinRule`

如果选到的是空或者选到的不是up的，且时间在ddl之前则进入重试逻辑：

while (!Thread.interrupted()) {
        
       answer = subRule.choose(key);
    
       if (((answer == null) || (!answer.isAlive()))
             && (System.currentTimeMillis() < deadline)) {
        
          /* pause and retry hoping it's transient */
          Thread.yield();
       } else {
        
          break;
       }
    }

如果线程中断了就中断重试。之后重新选择服务器，如果又没选到则把资源让出去，下一次`while`循环再选，在`while`循环之前会起一个任务

InterruptTask task = new InterruptTask(deadline - System.currentTimeMillis());

到了截止时间之后，程序会中断重试的流程

task.cancel();

最后返回

if ((answer == null) || (!answer.isAlive())) {
        
       return null;
    } else {
        
       return answer;
    }

[Link 1]: https://bbs.csdn.net/forums/TOBEJAVAGOD
[RocketMQ]: https://blog.csdn.net/hnu_csee_wjw/category_11789299.html
[Link 2]: https://blog.csdn.net/hnu_csee_wjw/category_11693862.html
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5bCP546L5pu-5piv5bCR5bm0_size_8_color_FFFFFF_t_70_g_se_x_16_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/24/891eb1cc72f3491bbe9927502374f699.png
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5bCP546L5pu-5piv5bCR5bm0_size_14_color_FFFFFF_t_70_g_se_x_16_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/24/c7140f7eaac9456b82eeea6779e9d302.png
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5bCP546L5pu-5piv5bCR5bm0_size_13_color_FFFFFF_t_70_g_se_x_16_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/24/c10b8c1ca516499c88151e09edbdb129.png
[f797f22e59e544c68be41fa694aa64bd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/24/6a89505566a64f69a0196f4fefa8327d.png