Ribbon源码

主要功能分析：

　　Ribbon的负载均衡主要通过LoadBalancerClient来实现的，而LoadBalanceClient具体交给了ILoadBalancer来处理，ILoadBalancer通过配置IRule，IPing等，向EurekaClient获取注册列表的信息，默认每10秒向EurekaClient发送一次，进而检查是否需要更新服务的注册列表信息，最后，在得到服务注册列表后，ILoadBalancer根据IRule的策略进行负载均衡。

　　而RestTemplate加上@LoadBalance注解后，在远程调用时能负载均衡，主要是维护了一个被@LoadBalance注解的RestTemplate列表，并给该列表中的RestTemplate对象添加了拦截器，在拦截器的方法中，将远程调度方法交给了Ribbon的负载器LoadBalanceClient去处理，从而实现。

LoadBalanceClient继承了ServiceInstanceChooser，具体实现类为RibbonLoadBalanceClient。

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

}

public interface ServiceInstanceChooser {

    /**

     * Chooses a ServiceInstance from the LoadBalancer for the specified service.

     * @param serviceId The service ID to look up the LoadBalancer.

     * @return A ServiceInstance that matches the serviceId.

     */

    ServiceInstance choose(String serviceId);

}

reconstructURI用于重构URL，choose用于根据serviceID获取ServiceInstance，具体实现

    public ServiceInstance choose(String serviceId, Object hint) {

        Server server = getServer(getLoadBalancer(serviceId), hint);

        if (server == null) {

            return null;

        }

        return new RibbonServer(serviceId, server, isSecure(server, serviceId),

                serverIntrospector(serviceId).getMetadata(server));

    }

    protected ILoadBalancer getLoadBalancer(String serviceId) {

        return this.clientFactory.getLoadBalancer(serviceId);

    }

在RibbonLoadBalancerClient的源码中，交给了ILoadBalancer类去执行，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方法用于标记服务下线

getReachableServer用于获取可用的Server集合

getAllServers用于获取所有server集合

接口继承关系：ILoadBalancer——AbstractLoadBalancer——BaseLoadBalancer——DynamicServerListLoadBalancer

    public DynamicServerListLoadBalancer(IClientConfig clientConfig, IRule rule, IPing ping,

                                         ServerList<T> serverList, ServerListFilter<T> filter,

                                         ServerListUpdater serverListUpdater) {

        super(clientConfig, rule, ping);

        this.serverListImpl = serverList;

        this.filter = filter;

        this.serverListUpdater = serverListUpdater;

        if (filter instanceof AbstractServerListFilter) {

            ((AbstractServerListFilter) filter).setLoadBalancerStats(getLoadBalancerStats());

        }

        restOfInit(clientConfig);

    }

DynamicServerListLoadBalancer类实例化的时候需要参数

IClientConfig：用于配置负载均衡的客户端，默认实现类DefaultClientConfigImpl
IRule：用于配置负载均衡的策略，默认是RoundRobinRule，提供了7种实现类

　　　　BestAvailableRule：选择最小请求数

　　　　ClientConfigEnableRoundRobinRule：轮询

　　　　RandomRule：随机选择一个server

　　　　RoundRobinRule：轮询选择server

　　　　RetryRule：根据轮询的方式重试

　　　　WeightedResponseTimeRule：根据响应时间分配权重，权重越低，被选择的可能性越低

　　　　ZoneAvoidanceRule：根据server的zone区域和可用性来轮询选择

IPing：DummyPing
ServerList：ConfigurationBasedServerList
ServerListFilter：ZonePreferenceServerListFilter

DynamicServerListLoadBalancer初始化中有一个restOfInit方法

    void restOfInit(IClientConfig clientConfig) {

        boolean primeConnection = this.isEnablePrimingConnections();

        // turn this off to avoid duplicated asynchronous priming done in BaseLoadBalancer.setServerList()

        this.setEnablePrimingConnections(false);

        enableAndInitLearnNewServersFeature();

        updateListOfServers();

        if (primeConnection && this.getPrimeConnections() != null) {

            this.getPrimeConnections()

                    .primeConnections(getReachableServers());

        }

        this.setEnablePrimingConnections(primeConnection);

        LOGGER.info("DynamicServerListLoadBalancer for client {} initialized: {}", clientConfig.getClientName(), this.toString());

    }

中间有一个updateListOfServers方法，该方法用来获取所有的serverList的

    @VisibleForTesting

    public void updateListOfServers() {

        List<T> servers = new ArrayList<T>();

        if (serverListImpl != null) {

            servers = serverListImpl.getUpdatedListOfServers();

            LOGGER.debug("List of Servers for {} obtained from Discovery client: {}",

                    getIdentifier(), servers);

            if (filter != null) {

                servers = filter.getFilteredListOfServers(servers);

                LOGGER.debug("Filtered List of Servers for {} obtained from Discovery client: {}",

                        getIdentifier(), servers);

            }

        }

        updateAllServerList(servers);

    }

由serverListImpl的实现类DiscoveryEnabledNIWSServerList的getUpdatedListOfServers

    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 = this.createServer(ii, this.isSecure, this.shouldUseIpAddr);

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

        }

    }

obtainServersViaDiscovery()方法是根据EurekaClientProvider.get来获取EurekaClient，再根据EurekaClient来获取注册列表信息

Ribbon就是从EurekaClient获取服务列表信息的，并根据IRule的策略去路由，根据Iping去判断服务可用性

@LoadBalanced注解

　　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 -> {

            for (RestTemplate restTemplate : LoadBalancerAutoConfiguration.this.restTemplates) {

                for (RestTemplateCustomizer customizer : customizers) {

                    customizer.customize(restTemplate);

                }

            }

        });

    }

    @Autowired(required = false)

    private List<LoadBalancerRequestTransformer> transformers = Collections.emptyList();

    @Bean

    @ConditionalOnMissingBean

    public LoadBalancerRequestFactory loadBalancerRequestFactory(

            LoadBalancerClient loadBalancerClient) {

        return new LoadBalancerRequestFactory(loadBalancerClient, transformers);

    }

    @Configuration

    @ConditionalOnMissingClass("org.springframework.retry.support.RetryTemplate")

    static class 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。

    @Override

    public ClientHttpResponse intercept(final HttpRequest request, final byte[] body,

            final ClientHttpRequestExecution execution) throws IOException {

        final URI originalUri = request.getURI();

        String serviceName = originalUri.getHost();

        Assert.state(serviceName != null, "Request URI does not contain a valid hostname: " + originalUri);

        return this.loadBalancer.execute(serviceName, requestFactory.createRequest(request, body, execution));

    }

巴特西

Ribbon源码

最新文章

热门文章