Ribbon源码解析

什么是Ribbon

Ribbon是Netflix公司开源的一个负载均衡的项目,它属于上述的第二种,是一个客户端负载均衡器,运行在客户端上。它是一个经过了云端测试的IPC库,可以很好地控制HTTP和TCP客户端的一些行为。 Feign已经默认使用了Ribbon。

RestTemplate和Ribbon相结合

Ribbon在Netflix组件是非常重要的一个组件,在Zuul中使用Ribbon做负载均衡,以及Feign组件的结合等。在Spring Cloud 中,作为开发中,做的最多的可能是将RestTemplate和Ribbon相结合,你可能会这样写:

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

消费另外一个的服务的接口,差不多是这样的:

@Service
public class RibbonService {
    @Autowired
    RestTemplate restTemplate;
    public String hi(String name) {
        return restTemplate.getForObject(""+name,String.class);
    }
}

深入理解Ribbon

RibbonAutoConfiguration

Ribbon的自动配置类,配置Ribbon负载均衡客户端(LoadBalancerClient)

@Configuration
@Conditional(RibbonAutoConfiguration.RibbonClassesConditions.class)
@RibbonClients
@AutoConfigureAfter(
		name = "org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration")
@AutoConfigureBefore({ LoadBalancerAutoConfiguration.class,
		AsyncLoadBalancerAutoConfiguration.class })
@EnableConfigurationProperties({ RibbonEagerLoadProperties.class,
		ServerIntrospectorProperties.class })
public class RibbonAutoConfiguration {

	@Bean
	public SpringClientFactory springClientFactory() {
			SpringClientFactory factory = new SpringClientFactory();
			factory.setConfigurations(this.configurations);
			return factory;
	}

	@Bean
	@ConditionalOnMissingBean(LoadBalancerClient.class)
	public LoadBalancerClient loadBalancerClient() {
		return new RibbonLoadBalancerClient(springClientFactory());
	}

}

RibbonEurekaAutoConfiguration在RibbonAutoConfiguration后启用

@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties
@ConditionalOnRibbonAndEurekaEnabled
@AutoConfigureAfter(**RibbonAutoConfiguration**.class)
@RibbonClients(defaultConfiguration = **EurekaRibbonClientConfiguration**.class)
public class RibbonEurekaAutoConfiguration {

}

EurekaRibbonClientConfiguration

使用Eureka client去获取Server节点信息

@Configuration(proxyBeanMethods = false)
public class EurekaRibbonClientConfiguration {

	@Bean
	@ConditionalOnMissingBean
	public IPing ribbonPing(IClientConfig config) {
		if (this.propertiesFactory.isSet(IPing.class, serviceId)) {
			return this.propertiesFactory.get(IPing.class, config, serviceId);
		}
		NIWSDiscoveryPing ping = new NIWSDiscoveryPing();
		ping.initWithNiwsConfig(config);
		return ping;
	}

	@Bean
	@ConditionalOnMissingBean
	public ServerList<?> ribbonServerList(IClientConfig config,
			Provider eurekaClientProvider) {
		if (this.propertiesFactory.isSet(ServerList.class, serviceId)) {
			return this.propertiesFactory.get(ServerList.class, config, serviceId);
		}
		**DiscoveryEnabledNIWSServerList** discoveryServerList = new DiscoveryEnabledNIWSServerList(
				config, eurekaClientProvider);
		DomainExtractingServerList serverList = new DomainExtractingServerList(
				discoveryServerList, config, this.approximateZoneFromHostname);
		return serverList;
	}

}

以上配置类在eureka-client的META-INF/Spring.factories文件中配置

org.springframework.boot.autoconfigure.EnableAutoConfiguration=
org.springframework.cloud.netflix.eureka.config.EurekaClientConfigServerAutoConfiguration,
org.springframework.cloud.netflix.eureka.config.EurekaDiscoveryClientConfigServiceAutoConfiguration,
org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration,
org.springframework.cloud.netflix.ribbon.eureka.**RibbonEurekaAutoConfiguration**,
org.springframework.cloud.netflix.eureka.EurekaDiscoveryClientConfiguration,
org.springframework.cloud.netflix.eureka.reactive.EurekaReactiveDiscoveryClientConfiguration

DiscoveryEnabledNIWSServerList

通过Eureka client获取服务节点信息列表,并提供给DynamicServerListLoadBalancer

/**
 * The server list class that fetches the server information from Eureka client. 
 * ServerList is used by DynamicServerListLoadBalancer to get server list dynamically.
*/ 
public class DiscoveryEnabledNIWSServerList extends AbstractServerList{
	private List obtainServersViaDiscovery() {
        List serverList = new ArrayList();
				EurekaClient eurekaClient = eurekaClientProvider.get();
        if (vipAddresses!=null){
            for (String vipAddress : vipAddresses.split(",")) {
                // if targetRegion is null, it will be interpreted as the same region of client
                List listOfInstanceInfo = eurekaClient.getInstancesByVipAddress(vipAddress, isSecure, targetRegion);
                for (InstanceInfo ii : listOfInstanceInfo) {
                    if (ii.getStatus().equals(InstanceStatus.UP)) {
												//  ......
												DiscoveryEnabledServer des = createServer(ii, isSecure, shouldUseIpAddr);
                        serverList.add(des);
										}

						}
				}
	}						

}

LoadBalancerClient

在Ribbon中一个非常重要的组件为LoadBalancerClient,它作为负载均衡的一个客户端。它在spring-cloud-commons包下:的LoadBalancerClient是一个接口,它继承ServiceInstanceChooser,它的实现类是RibbonLoadBalancerClient,这三者之间的关系如下图:

其中LoadBalancerClient接口,有如下三个方法,其中excute()为执行请求,reconstructURI()用来重构url:

package org.springframework.cloud.client.loadbalancer;

public interface LoadBalancerClient extends ServiceInstanceChooser {

   T execute(String serviceId, LoadBalancerRequest request) throws IOException;
   T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest request) throws IOException;
  URI reconstructURI(ServiceInstance instance, URI original);

}

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

public interface ServiceInstanceChooser {

    ServiceInstance choose(String serviceId);
}

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

public class RibbonLoadBalancerClient implements LoadBalancerClient {

  //省略代码

	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 Server getServer(ILoadBalancer loadBalancer, Object hint) {
		if (loadBalancer == null) {
			return null;
		}
		// Use 'default' on a null hint, or just pass it on?
		return loadBalancer.chooseServer(hint != null ? hint : "default");
	}

	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包下,它是定义了实现软件负载均衡的一个接口,它需要一组可供选择的服务注册列表信息,以及根据特定方法去选择服务,它的源码如下 :

public interface ILoadBalancer {
		// 添加服务节点
    public void addServers(List newServers);
		// 根据key选择服务节点
    public Server chooseServer(Object key);
    public void markServerDown(Server server);
		// 可达服务列表
    public List getReachableServers();
		// 所有服务节点
    public List getAllServers();
}

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

DynamicServerListLoadBalancer

它的继承类为BaseLoadBalancer,它的实现类为DynamicServerListLoadBalancer,这三者之间的关系如下:

查看上述三个类的源码,可用发现,配置以下信息,IClientConfig、IRule、IPing、ServerList、ServerListFilter和ILoadBalancer,查看BaseLoadBalancer类,它默认的情况下,实现了以下配置:

IClientConfig

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

IRule

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

public interface IRule{

    public Server choose(Object key);
    
    public void setLoadBalancer(ILoadBalancer lb);
    
    public ILoadBalancer getLoadBalancer();    
}

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

**RoundRobinRule:轮询负载规则实现**

public class RoundRobinRule extends AbstractLoadBalancerRule {
		// 原子整数,下一次获取Server节点的index
		private AtomicInteger nextServerCyclicCounter;

		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 reachableServers = lb.getReachableServers();
            List 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) {
                /* 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;
    }

    /**
     * Inspired by the implementation of {@link AtomicInteger#incrementAndGet()}.
     *
     * @param modulo The modulo to bound the value of the counter.
     * @return The next value.
     */
    private int incrementAndGetModulo(int modulo) {
        for (;;) {
            int current = nextServerCyclicCounter.get();
            int next = (current + 1) % modulo;
            if (nextServerCyclicCounter.compareAndSet(current, next))
                return next;
        }
    }

)

BestAvailableRule:最少并发请求负载规则

public class BestAvailableRule extends ClientConfigEnabledRoundRobinRule {

    private LoadBalancerStats loadBalancerStats;
    
    @Override
    public Server choose(Object key) {
        if (loadBalancerStats == null) {
            return super.choose(key);
        }
        List serverList = getLoadBalancer().getAllServers();
        int minimalConcurrentConnections = Integer.MAX_VALUE;
        long currentTime = System.currentTimeMillis();
        Server chosen = null;
        for (Server server: serverList) {
						// 获取服务节点的状态信息,保存在Guava缓存中
            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;
        }
    }

    @Override
    public void setLoadBalancer(ILoadBalancer lb) {
        super.setLoadBalancer(lb);
        if (lb instanceof AbstractLoadBalancer) {
            loadBalancerStats = ((AbstractLoadBalancer) lb).getLoadBalancerStats();            
        }
    }
}

IPing

IPing是用来想server发生”ping”,来判断该server是否有响应,从而判断该server是否可用。它有一个isAlive()方法,它的源码如下:

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

IPing的实现类有PingUrl、PingConstant、NoOpPing、DummyPing和NIWSDiscoveryPing。它门之间的关系如下:

ServerList

ServerList是定义获取所有的server的注册列表信息的接口,它的代码如下:

public interface ServerList {

    public List getInitialListOfServers();
    public List getUpdatedListOfServers();   

}

ServerListFilter

ServerListFilter接口,定于了可根据配置去过滤或者根据特性动态获取符合条件的server列表的方法,代码如下:

public interface ServerListFilter {

    public List getFilteredListOfServers(List servers);

}

DynamicServerListLoadBalancer

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

public DynamicServerListLoadBalancer(IClientConfig clientConfig) {
        initWithNiwsConfig(clientConfig);
    }
    
    @Override
    public void initWithNiwsConfig(IClientConfig clientConfig) {
        try {
            super.initWithNiwsConfig(clientConfig);
            String niwsServerListClassName = clientConfig.getPropertyAsString(
                    CommonClientConfigKey.NIWSServerListClassName,
                    DefaultClientConfigImpl.DEFAULT_SEVER_LIST_CLASS);

            ServerList niwsServerListImpl = (ServerList) ClientFactory
                    .instantiateInstanceWithClientConfig(niwsServerListClassName, clientConfig);
            this.serverListImpl = niwsServerListImpl;

            if (niwsServerListImpl instanceof AbstractServerList) {
                AbstractServerListFilter niwsFilter = ((AbstractServerList) niwsServerListImpl)
                        .getFilterImpl(clientConfig);
                niwsFilter.setLoadBalancerStats(getLoadBalancerStats());
                this.filter = niwsFilter;
            }

            String serverListUpdaterClassName = clientConfig.getPropertyAsString(
                    CommonClientConfigKey.ServerListUpdaterClassName,
                    DefaultClientConfigImpl.DEFAULT_SERVER_LIST_UPDATER_CLASS
            );

            this.serverListUpdater = (ServerListUpdater) ClientFactory
                    .instantiateInstanceWithClientConfig(serverListUpdaterClassName, clientConfig);

            **restOfInit**(clientConfig);
        } catch (Exception e) {
            throw new RuntimeException(
                    "Exception while initializing NIWSDiscoveryLoadBalancer:"
                            + clientConfig.getClientName()
                            + ", niwsClientConfig:" + clientConfig, e);
        }
    }

在restOfInit()方法上,有一个 updateListOfServers()的方法,该方法是用来获取所有的ServerList的。

 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()方法的源码,最终由serverListImpl.getUpdatedListOfServers()获取所有的服务列表的,代码如下:

   @VisibleForTesting
    public void updateListOfServers() {
        List servers = new ArrayList();
        if (serverListImpl != null) {
						// serverList接口的实现类
            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是ServerList接口的具体实现类。跟踪代码,ServerList的实现类为DiscoveryEnabledNIWSServerList,在ribbon-eureka.jar的com.netflix.niws.loadbalancer下。其中DiscoveryEnabledNIWSServerList有 getInitialListOfServers()和getUpdatedListOfServers()方法,具体代码如下:

		@Override
    public List getInitialListOfServers(){
        return obtainServersViaDiscovery();
    }

    @Override
    public List getUpdatedListOfServers(){
        return obtainServersViaDiscovery();
    }

继续跟踪源码,obtainServersViaDiscovery(),是根据eurekaClientProvider.get()来回去EurekaClient,再根据EurekaClient来获取注册列表信息,代码如下:

private List obtainServersViaDiscovery() {
        List serverList = new ArrayList();

        if (eurekaClientProvider == null || eurekaClientProvider.get() == null) {
            logger.warn("EurekaClient has not been initialized yet, returning an empty list");
            return new ArrayList();
        }

        EurekaClient eurekaClient = eurekaClientProvider.get();
        if (vipAddresses!=null){
            for (String vipAddress : vipAddresses.split(",")) {
                // if targetRegion is null, it will be interpreted as the same region of client
                List listOfInstanceInfo = eurekaClient.getInstancesByVipAddress(vipAddress, isSecure, targetRegion);
                for (InstanceInfo ii : listOfInstanceInfo) {
                    if (ii.getStatus().equals(InstanceStatus.UP)) {

                        if(shouldUseOverridePort){
                            if(logger.isDebugEnabled()){
                                logger.debug("Overriding port on client name: " + clientName + " to " + overridePort);
                            }

                            // copy is necessary since the InstanceInfo builder just uses the original reference,
                            // and we don't want to corrupt the global eureka copy of the object which may be
                            // used by other clients in our system
                            InstanceInfo copy = new InstanceInfo(ii);

                            if(isSecure){
                                ii = new InstanceInfo.Builder(copy).setSecurePort(overridePort).build();
                            }else{
                                ii = new InstanceInfo.Builder(copy).setPort(overridePort).build();
                            }
                        }

                        DiscoveryEnabledServer des = new DiscoveryEnabledServer(ii, isSecure, shouldUseIpAddr);
                        des.setZone(DiscoveryClient.getZone(ii));
                        serverList.add(des);
                    }
                }
                if (serverList.size()>0 && prioritizeVipAddressBasedServers){
                    break; // if the current vipAddress has servers, we dont use subsequent vipAddress based servers
                }
            }
        }
        return serverList;
    }

其中eurekaClientProvider的实现类是LegacyEurekaClientProvider,它是一个获取eurekaClient类,通过静态的方法去获取eurekaClient,其代码如下:

class LegacyEurekaClientProvider implements Provider {

    private volatile EurekaClient eurekaClient;

    @Override
    public synchronized EurekaClient get() {
        if (eurekaClient == null) {
            eurekaClient = DiscoveryManager.getInstance().getDiscoveryClient();
        }

        return eurekaClient;
    }
}

EurekaClient的实现类为DiscoveryClient,在之前已经分析了它具有服务注册、获取服务注册列表等的全部功能。

由此可见,负载均衡器是从EurekaClient获取服务信息,并根据IRule去路由,并且根据IPing去判断服务的可用性。

那么现在还有个问题,负载均衡器多久一次去获取一次从Eureka Client获取注册信息呢。

在BaseLoadBalancer类下,BaseLoadBalancer的构造函数,该构造函数开启了一个PingTask任务,代码如下:

   public BaseLoadBalancer(String name, IRule rule, LoadBalancerStats stats,
            IPing ping, IPingStrategy pingStrategy) {
	    ...//代码省略
        setupPingTask();
         ...//代码省略
    }

setupPingTask()的具体代码逻辑,它开启了ShutdownEnabledTimer执行PingTask任务,在默认情况下pingIntervalSeconds为10,即每10秒钟,想EurekaClient发送一次”ping”。

    void setupPingTask() {
        if (canSkipPing()) {
            return;
        }
        if (lbTimer != null) {
            lbTimer.cancel();
        }
        lbTimer = new ShutdownEnabledTimer("NFLoadBalancer-PingTimer-" + name,
                true);
        lbTimer.schedule(new PingTask(), 0, pingIntervalSeconds * 1000);
        forceQuickPing();
    }

PingTask源码,即new一个Pinger对象,并执行runPinger()方法。

class PingTask extends TimerTask {
        public void run() {
            try {
            	new Pinger(pingStrategy).runPinger();
            } catch (Exception e) {
                logger.error("LoadBalancer [{}]: Error pinging", name, e);
            }
        }
    }

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

  public void runPinger() throws Exception {
            if (!pingInProgress.compareAndSet(false, true)) { 
                return; // Ping in progress - nothing to do
            }
            
            // we are "in" - we get to Ping

            Server[] allServers = null;
            boolean[] results = null;

            Lock allLock = null;
            Lock upLock = null;

            try {
                /*
                 * The readLock should be free unless an addServer operation is
                 * going on...
                 */
                allLock = allServerLock.readLock();
                allLock.lock();
                allServers = allServerList.toArray(new Server[allServerList.size()]);
                allLock.unlock();

                int numCandidates = allServers.length;
                results = pingerStrategy.pingServers(ping, allServers);

                final List newUpList = new ArrayList();
                final List changedServers = new ArrayList();

                for (int i = 0; i < numCandidates; i++) {
                    boolean isAlive = results[i];
                    Server svr = allServers[i];
                    boolean oldIsAlive = svr.isAlive();

                    svr.setAlive(isAlive);

                    if (oldIsAlive != isAlive) {
                        changedServers.add(svr);
                        logger.debug("LoadBalancer [{}]:  Server [{}] status changed to {}", 
                    		name, svr.getId(), (isAlive ? "ALIVE" : "DEAD"));
                    }

                    if (isAlive) {
                        newUpList.add(svr);
                    }
                }
                upLock = upServerLock.writeLock();
                upLock.lock();
                upServerList = newUpList;
                upLock.unlock();

                notifyServerStatusChangeListener(changedServers);
            } finally {
                pingInProgress.set(false);
            }
        }

由此可见,LoadBalancerClient是在初始化的时候,会向Eureka回去服务注册列表,并且向通过10s一次向EurekaClient发送“ping”,来判断服务的可用性,如果服务的可用性发生了改变或者服务数量和之前的不一致,则更新或者重新拉取。LoadBalancerClient有了这些服务注册列表,就可以根据具体的IRule来进行负载均衡。

RestTemplate是如何和Ribbon结合的

最后,回答问题的本质,为什么在RestTemplate加一个@LoadBalance注解就可可以开启负载均衡呢?

 @LoadBalanced
    RestTemplate restTemplate() {
        return new RestTemplate();
    }

全局搜索ctr+shift+f @LoadBalanced有哪些类用到了LoadBalanced有哪些类用到了, 发现LoadBalancerAutoConfiguration类,即LoadBalancer自动配置类。

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

@LoadBalanced
	@Autowired(required = false)
	private List restTemplates = Collections.emptyList();
}
	@Bean
	public SmartInitializingSingleton loadBalancedRestTemplateInitializer(
			final List customizers) {
		return new SmartInitializingSingleton() {
			@Override
			public void afterSingletonsInstantiated() {
				for (RestTemplate restTemplate : LoadBalancerAutoConfiguration.this.restTemplates) {
					for (RestTemplateCustomizer customizer : customizers) {
						customizer.customize(restTemplate);
					}
				}
			}
		};
	}
	
	
	@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 new RestTemplateCustomizer() {
				@Override
				public void customize(RestTemplate restTemplate) {
					List list = new ArrayList<>(
							restTemplate.getInterceptors());
					list.add(loadBalancerInterceptor);
					restTemplate.setInterceptors(list);
				}
			};
		}
	}

}

在该类中,首先维护了一个被@LoadBalanced修饰的RestTemplate对象的List,在初始化的过程中,通过调用customizer.customize(restTemplate)方法来给RestTemplate增加拦截器LoadBalancerInterceptor。

而LoadBalancerInterceptor,用于实时拦截,在LoadBalancerInterceptor这里实现来负载均衡。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的负载均衡,主要通过LoadBalancerClient来实现的,而LoadBalancerClient具体交给了ILoadBalancer来处理,ILoadBalancer通过配置IRule、IPing等信息,并向EurekaClient获取注册列表的信息,并默认10秒一次向EurekaClient发送“ping”,进而检查是否更新服务列表,最后,得到注册列表后,ILoadBalancer根据IRule的策略进行负载均衡。

而RestTemplate 被@LoadBalance注解后,能过用负载均衡,主要是维护了一个被@LoadBalance注解的RestTemplate列表,并给列表中的RestTemplate添加拦截器,进而交给负载均衡器去处理。

展开阅读全文

页面更新:2024-05-02

标签:均衡器   源码   可用性   节点   负载   客户端   代码   方法   列表   信息

1 2 3 4 5

上滑加载更多 ↓
推荐阅读:
友情链接:
更多:

本站资料均由网友自行发布提供,仅用于学习交流。如有版权问题,请与我联系,QQ:4156828  

© CopyRight 2020-2024 All Rights Reserved. Powered By 71396.com 闽ICP备11008920号-4
闽公网安备35020302034903号

Top