Ribbon源码深度解析

墨蓝 2022-02-01 06:29 341阅读 0赞

### 什么是Ribbon ###

Ribbon是Netflix公司开源的一个负载均衡的项目，它属于上述的第二种，是一个客户端负载均衡器，运行在客户端上。它是一个经过了云端测试的IPC库，可以很好地控制HTTP和TCP客户端的一些行为。 Feign已经默认使用了Ribbon。能够实现负载均衡、容错、多协议（HTTP，TCP，UDP）支持异步和反应模型、缓存和批处理。在Spring Cloud 中，作为开发者，做的最多的可能是将RestTemplate和Ribbon相结合，本文主要介绍基于@LoadBalanced的RestTemplate来实现的负载均衡的源码解析。

### **Ribbon使用简介** ###

maven引入依赖

<dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-ribbon</artifactId>
    </dependency>

配置restTemplate

@Configuration
    public class RibbonConfig {
        @Bean
        @LoadBalanced
        RestTemplate restTemplate() {
            return new RestTemplate();
        }
    }

消费另外一个的服务的接口

@Service
    public class HelloService {
    
        @Autowired
        RestTemplate restTemplate;
    
        public String hiService(String name) {
            return restTemplate.getForObject("http://SERVICE-HI/hi?name="+name,String.class);
        }
    
    }

### Ribbon源码分析 ###

@LoadBalanced

@Target({ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD})
    @Retention(RetentionPolicy.RUNTIME)
    @Documented
    @Inherited
    @Qualifier
    public @interface LoadBalanced {
    }

我们看到只是实例化了一个空类，没有任何实现。按照常规惯例，我们到引入的jar包下看下spring-cloud-commons-\{version\}-sources.jar!META-INF/spring.factories，

# AutoConfiguration
    org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
    org.springframework.cloud.client.CommonsClientAutoConfiguration,\
    org.springframework.cloud.client.discovery.composite.CompositeDiscoveryClientAutoConfiguration,\
    org.springframework.cloud.client.discovery.noop.NoopDiscoveryClientAutoConfiguration,\
    org.springframework.cloud.client.discovery.simple.SimpleDiscoveryClientAutoConfiguration,\
    org.springframework.cloud.client.hypermedia.CloudHypermediaAutoConfiguration,\
    org.springframework.cloud.client.loadbalancer.AsyncLoadBalancerAutoConfiguration,\
    org.springframework.cloud.client.loadbalancer.LoadBalancerAutoConfiguration,\
    org.springframework.cloud.client.loadbalancer.reactive.ReactiveLoadBalancerAutoConfiguration,\
    org.springframework.cloud.client.serviceregistry.ServiceRegistryAutoConfiguration,\
    org.springframework.cloud.commons.httpclient.HttpClientConfiguration,\
    org.springframework.cloud.commons.util.UtilAutoConfiguration,\
    org.springframework.cloud.client.serviceregistry.AutoServiceRegistrationAutoConfiguration
    
    
    # Environment Post Processors
    org.springframework.boot.env.EnvironmentPostProcessor=\
    org.springframework.cloud.client.HostInfoEnvironmentPostProcessor

与LoadBalanced相关的LoadBalancerAutoConfiguration即LoadBalancer自动配置类。会在项目启动时被添加到Spring容器中那么我们来详细分析下这个类。

@Configuration
    @ConditionalOnClass({RestTemplate.class})
    @ConditionalOnBean({LoadBalancerClient.class})
    @EnableConfigurationProperties({LoadBalancerRetryProperties.class})

**LoadBalancerClient**

LoadBalancerAutoConfiguration中最重要的组件，它作为负载均衡的一个客户端。它在spring-cloud-commons包下：LoadBalancerClient是一个接口，它继承ServiceInstanceChooser，它的实现类是RibbonLoadBalancerClient。其中LoadBalancerClient接口，有如下三个方法，其中excute()为执行请求，reconstructURI()用来重构url：

public interface LoadBalancerClient extends ServiceInstanceChooser {
        <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException;
    
        <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException;
    
        URI reconstructURI(ServiceInstance instance, URI original);
    }

**ServiceInstanceChooser**

主要有一个方法，用来根据serviceId来获取ServiceInstance，代码如下:

public interface ServiceInstanceChooser {
        ServiceInstance choose(String serviceId);
    }

**RibbonLoadBalancerClient**

LoadBalancerClient的实现类，这个类是非常重要的一个类，最终的负载均衡的请求处理，由它来执行。它的部分源码如下：

public class RibbonLoadBalancerClient implements LoadBalancerClient {
    
            ...//省略代码
    
            @Override
    	public ServiceInstance choose(String serviceId) {
    		Server server = getServer(serviceId);
    		if (server == null) {
    			return null;
    		}
    		return new RibbonServer(serviceId, server, isSecure(server, serviceId),
    				serverIntrospector(serviceId).getMetadata(server));
    	}
    
    
    
            protected Server getServer(String serviceId) {
    		return getServer(getLoadBalancer(serviceId));
            }
    
    
            protected Server getServer(ILoadBalancer loadBalancer) {
    		if (loadBalancer == null) {
    			return null;
    		}
    		return loadBalancer.chooseServer("default"); // TODO: better handling of key
    	}
    
    
            protected ILoadBalancer getLoadBalancer(String serviceId) {
    		return this.clientFactory.getLoadBalancer(serviceId);
    	}
    	
    	...//省略代码

在RibbonLoadBalancerClient的源码中，choose()方法是选择具体服务实例。该方法通过getServer()方法去获取实例，经过源码跟踪，最终交给了ILoadBalancer类去选择服务实例。ILoadBalancer在ribbon-loadbalancer的jar包下,它是定义了实现软件负载均衡的一个接口，它需要一组可供选择的服务注册列表信息，以及根据特定方法去选择服务，它的源码如下 ：

**ILoadBalancer**

public interface ILoadBalancer {
        void addServers(List<Server> var1);
    
        Server chooseServer(Object var1);
    
        void markServerDown(Server var1);
    
        /** @deprecated */
        @Deprecated
        List<Server> getServerList(boolean var1);
    
        List<Server> getReachableServers();
    
        List<Server> getAllServers();
    }

addServers()方法是添加一个Server集合；chooseServer()方法是根据key去获取Server；markServerDown()方法用来标记某个服务下线；getReachableServers()获取可用的Server集合；getAllServers()获取所有的Server集合。

我们看看和ILoadBalancer接口关联的几个类关系图  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2plcnJ5RHphbg_size_16_color_FFFFFF_t_70][]

**BaseLoadBalancer**

默认实现了以下配置：

*      IClientConfig ribbonClientConfig: DefaultClientConfigImpl配置
 *      IRule ribbonRule: RoundRobinRule 路由策略
 *      IPing ribbonPing: DummyPing         ping策略
 *      ServerList ribbonServerList: ConfigurationBasedServerList   定义获取所有的server的注册列表信息的接口
 *      ServerListFilter ribbonServerListFilter: ZonePreferenceServerListFilter   定于了可根据配置去过滤或者根据特性动态获取符合条件的server列表的方法
 *      ILoadBalancer ribbonLoadBalancer: ZoneAwareLoadBalancer

**IClientConfig** 用于对客户端或者负载均衡的配置，它的默认实现类为DefaultClientConfigImpl。

**IRule**用于复杂均衡的策略，它有三个方法，其中choose()是根据key 来获取server,setLoadBalancer()和getLoadBalancer()是用来设置和获取ILoadBalancer的，它的源码如下：

public interface IRule {
        Server choose(Object var1);
    
        void setLoadBalancer(ILoadBalancer var1);
    
        ILoadBalancer getLoadBalancer();
    }

IRule有很多默认的实现类，这些实现类根据不同的算法和逻辑来处理负载均衡。Ribbon实现的IRule有一下。在大多数情况下，这些默认的实现类是可以满足需求的，如果有特性的需求，可以自己实现。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2plcnJ5RHphbg_size_16_color_FFFFFF_t_70 1][]

*  BestAvailableRule 选择最小请求数
 *  ClientConfigEnabledRoundRobinRule 轮询
 *  RandomRule 随机选择一个server
 *  RoundRobinRule 轮询选择server
 *  RetryRule 根据轮询的方式重试
 *  WeightedResponseTimeRule 根据响应时间去分配一个weight ，weight越低，被选择的可能性就越低
 *  ZoneAvoidanceRule 根据server的zone区域和可用性来轮询选择

**IPing**

是用来想server发生"ping"，来判断该server是否有响应，从而判断该server是否可用。它有一个isAlive()方法

public interface IPing {
        boolean isAlive(Server var1);
    }

IPing的实现类有PingUrl、PingConstant、NoOpPing、DummyPing和NIWSDiscoveryPing。作用如下：

![20190516145853554.png][]

*  PingUrl 真实的去ping 某个url，判断其是否alive
 *  PingConstant 固定返回某服务是否可用，默认返回true，即可用
 *  NoOpPing 不去ping,直接返回true,即可用。
 *  DummyPing 直接返回true，并实现了initWithNiwsConfig方法。
 *  NIWSDiscoveryPing，根据DiscoveryEnabledServer的InstanceInfo的InstanceStatus去判断，如果为InstanceStatus.UP，则为可用，否则不可用。

**DynamicServerListLoadBalancer**

DynamicServerListLoadBalancer的构造函数中有个initWithNiwsConfig()方法。在改方法中，经过一系列的初始化配置，最终执行了restOfInit()方法。其代码如下：

public DynamicServerListLoadBalancer(IClientConfig clientConfig) {
            this.isSecure = false;
            this.useTunnel = false;
            this.serverListUpdateInProgress = new AtomicBoolean(false);
            this.updateAction = new UpdateAction() {
                public void doUpdate() {
                    DynamicServerListLoadBalancer.this.updateListOfServers();
                }
            };
            this.initWithNiwsConfig(clientConfig);
        }
    
        public void initWithNiwsConfig(IClientConfig clientConfig) {
            try {
                super.initWithNiwsConfig(clientConfig);
                String niwsServerListClassName = clientConfig.getPropertyAsString(CommonClientConfigKey.NIWSServerListClassName, "com.netflix.loadbalancer.ConfigurationBasedServerList");
                ServerList<T> niwsServerListImpl = (ServerList)ClientFactory.instantiateInstanceWithClientConfig(niwsServerListClassName, clientConfig);
                this.serverListImpl = niwsServerListImpl;
                if (niwsServerListImpl instanceof AbstractServerList) {
                    AbstractServerListFilter<T> niwsFilter = ((AbstractServerList)niwsServerListImpl).getFilterImpl(clientConfig);
                    niwsFilter.setLoadBalancerStats(this.getLoadBalancerStats());
                    this.filter = niwsFilter;
                }
    
                String serverListUpdaterClassName = clientConfig.getPropertyAsString(CommonClientConfigKey.ServerListUpdaterClassName, "com.netflix.loadbalancer.PollingServerListUpdater");
                this.serverListUpdater = (ServerListUpdater)ClientFactory.instantiateInstanceWithClientConfig(serverListUpdaterClassName, clientConfig);
                this.restOfInit(clientConfig);
            } catch (Exception var5) {
                throw new RuntimeException("Exception while initializing NIWSDiscoveryLoadBalancer:" + clientConfig.getClientName() + ", niwsClientConfig:" + clientConfig, var5);
            }
        }
    
        void restOfInit(IClientConfig clientConfig) {
            boolean primeConnection = this.isEnablePrimingConnections();
            this.setEnablePrimingConnections(false);
            this.enableAndInitLearnNewServersFeature();
            this.updateListOfServers();
            if (primeConnection && this.getPrimeConnections() != null) {
                this.getPrimeConnections().primeConnections(this.getReachableServers());
            }
    
            this.setEnablePrimingConnections(primeConnection);
            LOGGER.info("DynamicServerListLoadBalancer for client {} initialized: {}", clientConfig.getClientName(), this.toString());
        }

在restOfInit()方法上，有一个 updateListOfServers()的方法，该方法是用来获取所有的ServerList的。最终由serverListImpl.getUpdatedListOfServers()获取所有的服务列表的，代码如下：

@VisibleForTesting
    public void updateListOfServers() {
            List<T> servers = new ArrayList();
            if (this.serverListImpl != null) {
                servers = this.serverListImpl.getUpdatedListOfServers();
                LOGGER.debug("List of Servers for {} obtained from Discovery client: {}", this.getIdentifier(), servers);
                if (this.filter != null) {
                    servers = this.filter.getFilteredListOfServers((List)servers);
                    LOGGER.debug("Filtered List of Servers for {} obtained from Discovery client: {}", this.getIdentifier(), servers);
                }
            }
    
            this.updateAllServerList((List)servers);
     }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2plcnJ5RHphbg_size_16_color_FFFFFF_t_70 2][]

而serverListImpl是ServerList接口的具体实现类。跟踪代码，ServerList的实现类为DiscoveryEnabledNIWSServerList，在ribbon-eureka.jar的com.netflix.niws.loadbalancer下。继续跟踪源码，obtainServersViaDiscovery（），是根据eurekaClientProvider.get()获取EurekaClient，再根据EurekaClient来获取注册列表信息

public List<DiscoveryEnabledServer> getUpdatedListOfServers() {
            return this.obtainServersViaDiscovery();
        }
    
        private List<DiscoveryEnabledServer> obtainServersViaDiscovery() {
            List<DiscoveryEnabledServer> serverList = new ArrayList();
            if (this.eurekaClientProvider != null && this.eurekaClientProvider.get() != null) {
                EurekaClient eurekaClient = (EurekaClient)this.eurekaClientProvider.get();
                if (this.vipAddresses != null) {
                    String[] var3 = this.vipAddresses.split(",");
                    int var4 = var3.length;
    
                    for(int var5 = 0; var5 < var4; ++var5) {
                        String vipAddress = var3[var5];
                        List<InstanceInfo> listOfInstanceInfo = eurekaClient.getInstancesByVipAddress(vipAddress, this.isSecure, this.targetRegion);
                        Iterator var8 = listOfInstanceInfo.iterator();
    
                        while(var8.hasNext()) {
                            InstanceInfo ii = (InstanceInfo)var8.next();
                            if (ii.getStatus().equals(InstanceStatus.UP)) {
                                if (this.shouldUseOverridePort) {
                                    if (logger.isDebugEnabled()) {
                                        logger.debug("Overriding port on client name: " + this.clientName + " to " + this.overridePort);
                                    }
    
                                    InstanceInfo copy = new InstanceInfo(ii);
                                    if (this.isSecure) {
                                        ii = (new Builder(copy)).setSecurePort(this.overridePort).build();
                                    } else {
                                        ii = (new Builder(copy)).setPort(this.overridePort).build();
                                    }
                                }
    
                                DiscoveryEnabledServer des = new DiscoveryEnabledServer(ii, this.isSecure, this.shouldUseIpAddr);
                                des.setZone(DiscoveryClient.getZone(ii));
                                serverList.add(des);
                            }
                        }
    
                        if (serverList.size() > 0 && this.prioritizeVipAddressBasedServers) {
                            break;
                        }
                    }
                }
    
                return serverList;
            } else {
                logger.warn("EurekaClient has not been initialized yet, returning an empty list");
                return new ArrayList();
            }
        }

EurekaClient的实现类为DiscoveryClient，具有服务注册、获取服务注册列表等的功能。由此可见，负载均衡器是从EurekaClient获取服务信息，并根据IRule去路由，并且根据IPing去判断服务的可用性。**这里有个问题：负载均衡器每隔多久从Eureka Client获取注册信息？**

在BaseLoadBalancer类下，BaseLoadBalancer的构造函数，调用setupPingTask()开启了一个PingTask任务，它开启了ShutdownEnabledTimer执行PingTask任务，在默认情况下pingIntervalSeconds为10，即每10秒钟，向EurekaClient发送一次"ping"。

void setupPingTask() {
            if (!this.canSkipPing()) {
                if (this.lbTimer != null) {
                    this.lbTimer.cancel();
                }
    
                this.lbTimer = new ShutdownEnabledTimer("NFLoadBalancer-PingTimer-" + this.name, true);
                this.lbTimer.schedule(new BaseLoadBalancer.PingTask(), 0L, (long)(this.pingIntervalSeconds * 1000));
                this.forceQuickPing();
            }
        }
    
    //PingTask源码，即new一个Pinger对象，并执行runPinger()方法。
    
    class PingTask extends TimerTask {
            PingTask() {
            }
    
            public void run() {
                try {
                    (BaseLoadBalancer.this.new Pinger(BaseLoadBalancer.this.pingStrategy)).runPinger();
                } catch (Exception var2) {
                    BaseLoadBalancer.logger.error("LoadBalancer [{}]: Error pinging", BaseLoadBalancer.this.name, var2);
                }
    
            }
        }

查看run()方法中的runPinger()方法，最终根据 pingerStrategy.pingServers(ping, allServers)来获取服务的可用性，如果该返回结果和之前相同，则不去向EurekaClient获取注册列表，如果不同则通知ServerStatusChangeListener或者changeListeners发生了改变，进行更新或者重新拉取。

public void runPinger() throws Exception {
                if (BaseLoadBalancer.this.pingInProgress.compareAndSet(false, true)) {
                    Server[] allServers = null;
                    boolean[] results = null;
                    Lock allLock = null;
                    Lock upLock = null;
    
                    try {
                        allLock = BaseLoadBalancer.this.allServerLock.readLock();
                        allLock.lock();
                        allServers = (Server[])BaseLoadBalancer.this.allServerList.toArray(new Server[BaseLoadBalancer.this.allServerList.size()]);
                        allLock.unlock();
                        int numCandidates = allServers.length;
                        boolean[] resultsx = this.pingerStrategy.pingServers(BaseLoadBalancer.this.ping, allServers);
                        List<Server> newUpList = new ArrayList();
                        List<Server> changedServers = new ArrayList();
    
                        for(int i = 0; i < numCandidates; ++i) {
                            boolean isAlive = resultsx[i];
                            Server svr = allServers[i];
                            boolean oldIsAlive = svr.isAlive();
                            svr.setAlive(isAlive);
                            if (oldIsAlive != isAlive) {
                                changedServers.add(svr);
                                BaseLoadBalancer.logger.debug("LoadBalancer [{}]:  Server [{}] status changed to {}", new Object[]{BaseLoadBalancer.this.name, svr.getId(), isAlive ? "ALIVE" : "DEAD"});
                            }
    
                            if (isAlive) {
                                newUpList.add(svr);
                            }
                        }
    
                        upLock = BaseLoadBalancer.this.upServerLock.writeLock();
                        upLock.lock();
                        BaseLoadBalancer.this.upServerList = newUpList;
                        upLock.unlock();
                        BaseLoadBalancer.this.notifyServerStatusChangeListener(changedServers);
                    } finally {
                        BaseLoadBalancer.this.pingInProgress.set(false);
                    }
                }
            }

### RestTemplate和Ribbon结合 ###

我们回到LoadBalancerAutoConfiguration源码：

@Configuration
    @ConditionalOnClass({RestTemplate.class})
    @ConditionalOnBean({LoadBalancerClient.class})
    @EnableConfigurationProperties({LoadBalancerRetryProperties.class})
    public class LoadBalancerAutoConfiguration {
        
        ...
        @LoadBalanced
        @Autowired(
            required = false
        )
        private List<RestTemplate> restTemplates = Collections.emptyList();
        ...
        @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);
                        }
                    }
    
                });
            };
        }
        ...
        @ConditionalOnMissingClass({"org.springframework.retry.support.RetryTemplate"})
        static class LoadBalancerInterceptorConfig {
            LoadBalancerInterceptorConfig() {
            }
    
            @Bean
            public LoadBalancerInterceptor ribbonInterceptor(LoadBalancerClient loadBalancerClient, LoadBalancerRequestFactory requestFactory) {
                return new LoadBalancerInterceptor(loadBalancerClient, requestFactory);
            }
    
            @Bean
            @ConditionalOnMissingBean
            public RestTemplateCustomizer restTemplateCustomizer(final LoadBalancerInterceptor loadBalancerInterceptor) {
                return (restTemplate) -> {
                    List<ClientHttpRequestInterceptor> list = new ArrayList(restTemplate.getInterceptors());
                    list.add(loadBalancerInterceptor);
                    restTemplate.setInterceptors(list);
                };
            }
        }
                    
    }

在该类中，首先维护了一个被@LoadBalanced修饰的RestTemplate对象的List，在初始化的过程中，通过调用customizer.customize(restTemplate)方法来给RestTemplate增加拦截器LoadBalancerInterceptor。用于实时拦截，并在这里实现来负载均衡。

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));
        }

### **Ribbon工作原理总结** ###

Ribbon的负载均衡，主要通过LoadBalancerClient来实现的，而LoadBalancerClient具体交给了ILoadBalancer来处理，ILoadBalancer通过配置IRule、IPing等信息，并向EurekaClient获取注册列表的信息，并默认10秒一次向EurekaClient发送“ping”,进而检查是否更新服务列表，最后，得到注册列表后，ILoadBalancer根据IRule的策略进行负载均衡。而RestTemplate 被@LoadBalance注解后，能过用负载均衡，主要是维护了一个被@LoadBalance注解的RestTemplate列表，并给列表中的RestTemplate添加拦截器，进而交给负载均衡器去处理。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2plcnJ5RHphbg_size_16_color_FFFFFF_t_70]: /images/20220201/3ba9e2d1ecda49a3a804e41729588ac8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2plcnJ5RHphbg_size_16_color_FFFFFF_t_70 1]: /images/20220201/cc1c02e2167c4e0384870b18ec9b6cd5.png
[20190516145853554.png]: /images/20220201/5d1afe9e31094729abeab6979c541088.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2plcnJ5RHphbg_size_16_color_FFFFFF_t_70 2]: /images/20220201/1a6ab13fd97e4d11904bc65a7699905f.png