Ribbon LoadBalanced底层机制源码探秘

﹏ヽ暗。殇╰゛Y 2024-05-24 21:38 63阅读 0赞

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

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

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

*  负载均衡器LoadBalancer原理
 *  Ribbon LoadBalanced底层机制源码探秘

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

1.  LoadBalanced作用原理
2.  拦截器到Rule的调角链路
3.  IPing机制

## 负载均衡器LoadBalancer原理 ##

**一句话概括：** 由`LoadBalanced`在`RestTemplate`上打标，Ribbon将带有负载均衡能力的拦截器注入标记好的`RestTemplate`中，以此实现负载均衡。

**从`@LoadBalanced`开始看起：**

它会将`RestTemplate`传送到`Ribbon`的自动装配类里进行改造。

![在这里插入图片描述][watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5bCP546L5pu-5piv5bCR5bm0_size_13_color_FFFFFF_t_70_g_se_x_16_pic_center]

*  `@LoadBalanced`：这个注解即修饰`RestTemplate`，还修饰`LoadBalancerAutoConfiguration`。它会将所有带有`LoadBalanced`注解的`RestTemplate`类，都传入到`LoadBalancerAutoConfiguration`中。这个注解的定义上还有一个`@Qualifier`注解，`@Qualifier`注解搭配`@Autowired`注解做自动装配，可以通过`name`属性，将指定的`Bean`装载到指定位置。

> 这里`LoadBalanced`也是借助`Qualifier`实现了一个给`RestTemplate`打标的功能，凡是被打标的`RestTemplate`都会被传送到`AutoConfig`中做进一步改造。

*  `LBAutoConfig`：从上一步中传送过来的`RestTemplate`，会经过`LBAutoConfig`的装配，将一系列的拦截器添加到`RestTemplate`中。`Ribbon`拦截器会拦截每个网络请求做一番处理，在这个过程中拦截器会找到对应的`LoadBalancer`对HTTP请求进行接管，接着`LoadBalancer`就会找到默认或指定的负载均衡策略来对HTTP请求进行转发。

> 拦截器是类似职责链设计模型的结构，常见的`ServletFilter`，权限控制器等都是类似的模式。

## Ribbon LoadBalanced底层机制源码探秘 ##

点进`LoadBalanced`注解，查到`LoadBalancerAutoConfiguration`使用了该注解

![请添加图片描述][watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5bCP546L5pu-5piv5bCR5bm0_size_19_color_FFFFFF_t_70_g_se_x_16_pic_center]  
该注解可以把修饰的`restTemplate`传送到`LoadBalancerAutoConfiguration`里

这个`restTemplate`只在`loadBalancedRestTemplateInitializerDeprecated`方法里被用到

@Bean
    public SmartInitializingSingleton loadBalancedRestTemplateInitializerDeprecated(final ObjectProvider<List<RestTemplateCustomizer>> restTemplateCustomizers) {
        
        return () -> {
        
            restTemplateCustomizers.ifAvailable((customizers) -> {
        
                Iterator var2 = this.restTemplates.iterator();
    
                while(var2.hasNext()) {
        
                    RestTemplate restTemplate = (RestTemplate)var2.next();
                    Iterator var4 = customizers.iterator();
    
                    while(var4.hasNext()) {
        
                        RestTemplateCustomizer customizer = (RestTemplateCustomizer)var4.next();
                        customizer.customize(restTemplate);
                    }
                }
    
            });
        };
    }

循环访问所有的`restTemplate`，`restTemplateCustomizers`是由外面初始化的bean注入进来的，使用`customizer`对`restTemplate`做手脚

@Bean
    @ConditionalOnMissingBean
    public RestTemplateCustomizer restTemplateCustomizer(final LoadBalancerInterceptor loadBalancerInterceptor) {
        
        return (restTemplate) -> {
        
            List<ClientHttpRequestInterceptor> list = new ArrayList(restTemplate.getInterceptors());
            list.add(loadBalancerInterceptor);
            restTemplate.setInterceptors(list);
        };
    }

先从`restTemplate`获取`getInterceptors()`，接下来list里添加一个`loadBalancerInterceptor`，它的注入：

@Bean
    public LoadBalancerInterceptor ribbonInterceptor(LoadBalancerClient loadBalancerClient, LoadBalancerRequestFactory requestFactory) {
        
        return new LoadBalancerInterceptor(loadBalancerClient, requestFactory);
    }

public void setInterceptors(List<ClientHttpRequestInterceptor> interceptors) {
        
        if (this.interceptors != interceptors) {
        
            this.interceptors.clear();
            this.interceptors.addAll(interceptors);
            AnnotationAwareOrderComparator.sort(this.interceptors);
        }
    }

这里把 `interceptors` 和本地保存的做一下比较，如果不一样则本地的`interceptors`全部清空，然后添加上新的，再`sort`一下

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

真正起作用的位置是在`LoadBalancerInterceptor`的`intercept`方法上

public ClientHttpResponse intercept(final HttpRequest request, final byte[] body, final ClientHttpRequestExecution execution) throws IOException {
        
        URI originalUri = request.getURI();
        String serviceName = originalUri.getHost();
        Assert.state(serviceName != null, "Request URI does not contain a valid hostname: " + originalUri);
        return (ClientHttpResponse)this.loadBalancer.execute(serviceName, this.requestFactory.createRequest(request, body, execution));
    }

先从`url`中得到`uri`，再从`uri`里得到`serviceName`（要去访问的`serviceName`），然后执行`execute`方法

public <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint) throws IOException {
        
        ILoadBalancer loadBalancer = this.getLoadBalancer(serviceId);
        Server server = this.getServer(loadBalancer, hint);
        if (server == null) {
        
            throw new IllegalStateException("No instances available for " + serviceId);
        } else {
        
            RibbonLoadBalancerClient.RibbonServer ribbonServer = new RibbonLoadBalancerClient.RibbonServer(serviceId, server, this.isSecure(server, serviceId), this.serverIntrospector(serviceId).getMetadata(server));
            return this.execute(serviceId, (ServiceInstance)ribbonServer, (LoadBalancerRequest)request);
        }
    }

这里到了真正执行任务的时候了，先根据`serviceId`获取一个`LoadBalancer`，拿到负载均衡策略之后用`getServer`获取到真正的`server`

protected Server getServer(ILoadBalancer loadBalancer, Object hint) {
        
        return loadBalancer == null ? null : loadBalancer.chooseServer(hint != null ? hint : "default");
    }

@Override
    public Server chooseServer(Object key) {
        
        if (!ENABLED.get() || getLoadBalancerStats().getAvailableZones().size() <= 1) {
        
            logger.debug("Zone aware logic disabled or there is only one zone");
            return super.chooseServer(key);
        }
        Server server = null;
        try {
        
            LoadBalancerStats lbStats = getLoadBalancerStats();
            Map<String, ZoneSnapshot> zoneSnapshot = ZoneAvoidanceRule.createSnapshot(lbStats);
            logger.debug("Zone snapshots: {}", zoneSnapshot);
            if (triggeringLoad == null) {
        
                triggeringLoad = DynamicPropertyFactory.getInstance().getDoubleProperty(
                        "ZoneAwareNIWSDiscoveryLoadBalancer." + this.getName() + ".triggeringLoadPerServerThreshold", 0.2d);
            }
    
            if (triggeringBlackoutPercentage == null) {
        
                triggeringBlackoutPercentage = DynamicPropertyFactory.getInstance().getDoubleProperty(
                        "ZoneAwareNIWSDiscoveryLoadBalancer." + this.getName() + ".avoidZoneWithBlackoutPercetage", 0.99999d);
            }
            Set<String> availableZones = ZoneAvoidanceRule.getAvailableZones(zoneSnapshot, triggeringLoad.get(), triggeringBlackoutPercentage.get());
            logger.debug("Available zones: {}", availableZones);
            if (availableZones != null &&  availableZones.size() < zoneSnapshot.keySet().size()) {
        
                String zone = ZoneAvoidanceRule.randomChooseZone(zoneSnapshot, availableZones);
                logger.debug("Zone chosen: {}", zone);
                if (zone != null) {
        
                    BaseLoadBalancer zoneLoadBalancer = getLoadBalancer(zone);
                    server = zoneLoadBalancer.chooseServer(key);
                }
            }
        } catch (Exception e) {
        
            logger.error("Error choosing server using zone aware logic for load balancer={}", name, e);
        }
        if (server != null) {
        
            return server;
        } else {
        
            logger.debug("Zone avoidance logic is not invoked.");
            return super.chooseServer(key);
        }
    }

如果只定义了一个`defaultZone`，则会调用父类的`chooseServer`方法

public Server chooseServer(Object key) {
        
        if (counter == null) {
        
            counter = createCounter();
        }
        counter.increment();
        if (rule == null) {
        
            return null;
        } else {
        
            try {
        
                return rule.choose(key);
            } catch (Exception e) {
        
                logger.warn("LoadBalancer [{}]:  Error choosing server for key {}", name, key, e);
                return null;
            }
        }
    }

这里使用默认的负载均衡策略`RandomRule`

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

回到`RibbonLoadBalancerClient`的`execute`方法，获取到的服务器不为空则：

RibbonLoadBalancerClient.RibbonServer ribbonServer = new RibbonLoadBalancerClient.RibbonServer(serviceId, server, this.isSecure(server, serviceId), this.serverIntrospector(serviceId).getMetadata(server));
    return this.execute(serviceId, (ServiceInstance)ribbonServer, (LoadBalancerRequest)request);

构建一个`RibbonServer`，最后`execute`就是真正发起请求了

[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_13_color_FFFFFF_t_70_g_se_x_16_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/24/8e4770d9bbf0493f9f076bc11b1c71b4.png
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5bCP546L5pu-5piv5bCR5bm0_size_19_color_FFFFFF_t_70_g_se_x_16_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/24/373f85c3eaaa44a49b5756f37fadb6e0.png