SpringCloud之Ribbon源码分析

程序员文章站 2022-06-22 16:20:05

...

@RibbonClient注解

@RibbonClient注解可以实现Ribbon客户端，ribbon需要设置客户端的名称，以及相关的路由配置类

@Configuration
@Import(RibbonClientConfigurationRegistrar.class)
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RibbonClient {
	String value() default "";
	String name() default "";
	Class<?>[] configuration() default {};
}

value和name是等价了，用于设置客户端的实例名称，而configuration用于指定配置类
接下来还导入了个RibbonClientConfigurationRegistrar，他实现了ImportBeanDefinitionRegistrar，他也不用多说，是spring的工具接口，用于spring动态注册BeanDefinition的接口，在这是用于注册Ribbon所需的BeanDefinition（比如Ribbon客户端实例）
org.springframework.cloud.netflix.ribbon.RibbonClientConfigurationRegistrar#registerBeanDefinitions

@Override
	public void registerBeanDefinitions(AnnotationMetadata metadata,
			BeanDefinitionRegistry registry) {
	
//获取@RibbonClient的参数，获取clientName后进行configuraction注册
		Map<String, Object> client = metadata.getAnnotationAttributes(
				RibbonClient.class.getName(), true);
//获取ribbonclient的value或者name属性
		String name = getClientName(client); 
		if (name != null) {
            //注册
			registerClientConfiguration(registry, name, client.get("configuration"));
		}
	}

这里注册会调用

private void registerClientConfiguration(BeanDefinitionRegistry registry,
			Object name, Object configuration) {
		BeanDefinitionBuilder builder = BeanDefinitionBuilder
				.genericBeanDefinition(RibbonClientSpecification.class);
		builder.addConstructorArgValue(name);
		builder.addConstructorArgValue(configuration);
		registry.registerBeanDefinition(name + ".RibbonClientSpecification",
				builder.getBeanDefinition());
	}

registerClientConfiguration会注册一个RibbonClientSpecification的bean,名称是ribbon的名称加上.RibbonClientSpecification
RibbonClientSpecification 实现了 NamedContextFactory.Specification，是提供给SpringClientFactory使用的，他用于初始化ribbon的相关实例使用
SpringClientFactory在哪？
SpringCloud之Ribbon源码分析
org.springframework.cloud.netflix.ribbon.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());
	}

那么最重要的就回到了LoadBalancerClient ，他是Ribbon项目最核心的类
org.springframework.cloud.client.loadbalancer.LoadBalancerClient

public interface LoadBalancerClient extends ServiceInstanceChooser {
//从servericeId 所代表的服务列表中选择一个服务器来发送网络请求
  <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException;
<T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException;
//构建网络请求URI
URI reconstructURI(ServiceInstance instance, URI original);
}

LoadBalancerClient

LoadBalancerClient 还继承了一个接口ServiceInstanceChooser
org.springframework.cloud.client.loadbalancer.ServiceInstanceChooser

public interface ServiceInstanceChooser {
    //根据serviceId从服务器列表中选择一个ServiceInstance
    ServiceInstance choose(String serviceId);
}

org.springframework.cloud.netflix.ribbon.RibbonLoadBalancerClient 是LoadBalancerClient 的实现类

@Override
	public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException {
//每次发送请求都回获取一个ILoadBalancer ，会涉及负载均衡（IRULS），服务器列表集群（ServerList） 和检验服务是否存活（IPing）等细节实现
		ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
		Server server = getServer(loadBalancer);
		if (server == null) {
			throw new IllegalStateException("No instances available for " + serviceId);
		}
		RibbonServer ribbonServer = new RibbonServer(serviceId, server, isSecure(server,
				serviceId), serverIntrospector(serviceId).getMetadata(server));

		return execute(serviceId, ribbonServer, request);
	}

另外
org.springframework.cloud.netflix.ribbon.RibbonLoadBalancerClient#getServer(com.netflix.loadbalancer.ILoadBalancer)

protected Server getServer(ILoadBalancer loadBalancer) {
		if (loadBalancer == null) {
			return null;
		}
		return loadBalancer.chooseServer("default"); 
	}

这方法直接调用了ILoadBalancer 的chooseServer方法来使用负载君和策略，从已知的服务列表中选出一个服务器实例
来接下来重点就到了ILoadBalancer 这接口了
他定义负载君和操作的接口，由前面说过的SpringClientFactory获得而来。
而SpringClientFactory又是再RibbonAutoConfiguration定义
ILoadBalancer
com.netflix.loadbalancer.ZoneAwareLoadBalancer 是ILoadBalancer 的具体实现
看下他构造方法

public ZoneAwareLoadBalancer(IClientConfig clientConfig, IRule rule, IPing ping, ServerList<T> serverList, ServerListFilter<T> filter) {
        super(clientConfig, rule, ping, serverList, filter);
    }

IClientConfig
client的配置类，具体指的DefaultClientConfigImpl
IRule
负载均衡的策略类,默认轮询 RoundRobinRule
IPing
服务可用性检查，默认DummyPing
ServerList
服务列表获取，ConfigurationBasedServerList
ServerListFilter
服务列表过滤 ZonePreferenceServerListFilter
ZoneAwareLoadBalancer其中一个重要的方法就是
com.netflix.loadbalancer.ZoneAwareLoadBalancer#chooseServer
SpringCloud之Ribbon源码分析

@Override
public Server chooseServer(Object key) {
     //如果就一个zone，直接返回
        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) {
                  //得到zoone对应的BaseLoadBalancer 
                    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);
        }
    }

com.netflix.loadbalancer.BaseLoadBalancer#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;
            }
        }
    }

IRULE负载均衡的实现

具体服务的的负载还是由IRULE实现，前面在入门部分也讲过怎么更换具体的路由实现方法
IRULE具体在RibbonClientConfiguration进行配置，IRule的接口choose 方法负责选区一个具体的服务。

@Bean
	@ConditionalOnMissingBean
	public IRule ribbonRule(IClientConfig config) {
		if (this.propertiesFactory.isSet(IRule.class, name)) {
			return this.propertiesFactory.get(IRule.class, config, name);
		}
		ZoneAvoidanceRule rule = new ZoneAvoidanceRule();
		rule.initWithNiwsConfig(config);
		return rule;
	}

SpringCloud之Ribbon源码分析 BestAvailableRule :选择最小请求数的服务器
RoundRobinRule：轮询
ClientConfigEnabledRoundRobinRule: 使用RoundRobinRule选择服务器
RetryRule：根据选的轮询的方式重试
WeightedResponseTimeRule：根据响应时间去计算一个权重weight ，weight越低，被选择的可能性就越低
ZoneAvoidanceRule：根据server的zone区域和可用性来轮询选择。

ZoneAvoidanceRule 的具体实现

ZoneAvoidanceRule是默认的IRule实例，他使用PredicateBasedRule来根据服务区的运行状况和服务器的可用性来选择服务器
它的父类是PredicateBasedRule
com.netflix.loadbalancer.PredicateBasedRule
具体依次做了以下事情

先使用ILoadBalancer 获取服务器列表
使用AbstractServerPredicate进行服务器过滤
最后轮询从剩余的服务器列表中选择最终的服务器

public abstract class PredicateBasedRule extends ClientConfigEnabledRoundRobinRule {

    public abstract AbstractServerPredicate getPredicate();
        
    @Override
    public Server choose(Object key) {
        ILoadBalancer lb = getLoadBalancer();
        Optional<Server> server = getPredicate().chooseRoundRobinAfterFiltering(lb.getAllServers(), key);
        if (server.isPresent()) {
            return server.get();
        } else {
            return null;
        }       
    }
}

com.netflix.loadbalancer.PredicateBasedRule#getPredicate 又是一个抽象的实现
具体实现
com.netflix.loadbalancer.ZoneAvoidanceRule#getPredicate

public ZoneAvoidanceRule() {
        super();
        ZoneAvoidancePredicate zonePredicate = new ZoneAvoidancePredicate(this);
        AvailabilityPredicate availabilityPredicate = new AvailabilityPredicate(this);
        compositePredicate = createCompositePredicate(zonePredicate, availabilityPredicate);
    }
    //将两个Predicate组合成一个CompositePredicate 
private CompositePredicate createCompositePredicate(ZoneAvoidancePredicate p1, AvailabilityPredicate p2) {
        return CompositePredicate.withPredicates(p1, p2)
                             .addFallbackPredicate(p2)
                             .addFallbackPredicate(AbstractServerPredicate.alwaysTrue())
                             .build();
        
    }

ZoneAvoidancePredicate ：判断一个服务器的运行状况是否可用，去除不可用服务器的所有服务器
AvailabilityPredicate ：用于过滤连接数过多的服务器

在来看下chooseRoundRobinAfterFiltering方法，前面已经说过了它是过滤的方法，然后AvailabilityPredicate 里面并没有这方法，他直接继承了他的父类
com.netflix.loadbalancer.AbstractServerPredicate#chooseRoundRobinAfterFiltering(java.util.List<com.netflix.loadbalancer.Server>)

public Optional<Server> chooseRoundRobinAfterFiltering(List<Server> servers) {
//过滤服务器列表
        List<Server> eligible = getEligibleServers(servers);
        if (eligible.size() == 0) {
            return Optional.absent();
        }
        //（i+1）%n 轮询选择一个服务实例
        return Optional.of(eligible.get(incrementAndGetModulo(eligible.size())));
    }

com.netflix.loadbalancer.AbstractServerPredicate#getEligibleServers(java.util.List<com.netflix.loadbalancer.Server>, java.lang.Object)
由于前面loadBalancerKey直接传入的null, 方法getEligibleServers会使用serverOnlyPredicate来依次过滤

 public List<Server> getEligibleServers(List<Server> servers, Object loadBalancerKey) {
        if (loadBalancerKey == null) {
            return ImmutableList.copyOf(Iterables.filter(servers, this.getServerOnlyPredicate()));            
        } else {
            List<Server> results = Lists.newArrayList();
            for (Server server: servers) {
                if (this.apply(new PredicateKey(loadBalancerKey, server))) {
                    results.add(server);
                }
            }
            return results;            
        }
    }

serverOnlyPredicate 则会调用apply方法，并将Server 对象分装PredicateKey当作参数传入

private final Predicate<Server> serverOnlyPredicate =  new Predicate<Server>() {
        @Override
        public boolean apply(@Nullable Server input) {                    
            return AbstractServerPredicate.this.apply(new PredicateKey(input));
        }
    };

AbstractServerPredicate并没有实现apply方法，具体的实现又回到了子类CompositePredicate的apply方法
会依次调用ZoneAvoidancePredicate与AvailabilityPredicate的apply方法

public class ZoneSnapshot {
    //实例数
final int instanceCount;
//平均负载
final double loadPerServer;
//断路器端口数量
final int circuitTrippedCount;
//活动请求数量
final int activeRequestsCount;
}

com.netflix.loadbalancer.ZoneAvoidancePredicate#apply

@Override
    public boolean apply(@Nullable PredicateKey input) {
        if (!ENABLED.get()) {
            return true;
        }
        String serverZone = input.getServer().getZone();
        if (serverZone == null) {
            //如果服务器没有zone的相关信息，直接返回
            return true;
        }
        LoadBalancerStats lbStats = getLBStats();
//LoadBalancerStats 存储每个服务器节点的执行特征和运行记录，这些信息可供动态负责均衡使用
        if (lbStats == null) {
            //如果没有服务器的记录，直接返回
            return true;
        }
        if (lbStats.getAvailableZones().size() <= 1) {
            //如果根本就一个服务器，直接返回
            return true;
        }
        //PredicateKey 封装了Server的信息，判断下服务器区的记录是否用当前区的信息
        Map<String, ZoneSnapshot> zoneSnapshot = ZoneAvoidanceRule.createSnapshot(lbStats);
         //如果没有直接返回
        if (!zoneSnapshot.keySet().contains(serverZone)) {
            // The server zone is unknown to the load balancer, do not filter it out 
            return true;
        }
        logger.debug("Zone snapshots: {}", zoneSnapshot);
        //获取可用的服务器列表
        Set<String> availableZones = ZoneAvoidanceRule.getAvailableZones(zoneSnapshot, triggeringLoad.get(), triggeringBlackoutPercentage.get());
        logger.debug("Available zones: {}", availableZones);
   //判断当前服务器是否在可用的服务器列表中
        if (availableZones != null) {
            return availableZones.contains(input.getServer().getZone());
        } else {
            return false;
        }
    }

com.netflix.loadbalancer.ZoneAvoidanceRule#getAvailableZones(java.util.Map<java.lang.String,com.netflix.loadbalancer.ZoneSnapshot>, double, double)
最后落到这方法上
这方法是用来筛选服务区列表的

public static Set<String> getAvailableZones(
            Map<String, ZoneSnapshot> snapshot, double triggeringLoad,
            double triggeringBlackoutPercentage) {
        if (snapshot.isEmpty()) {
            return null;
        }
        Set<String> availableZones = new HashSet<String>(snapshot.keySet());
        if (availableZones.size() == 1) {
            return availableZones;
        }
        Set<String> worstZones = new HashSet<String>();
        double maxLoadPerServer = 0;
        boolean limitedZoneAvailability = false;
//遍历所有的服务区
        for (Map.Entry<String, ZoneSnapshot> zoneEntry : snapshot.entrySet()) {
            String zone = zoneEntry.getKey();
            ZoneSnapshot zoneSnapshot = zoneEntry.getValue();
//获取服务器中的服务实例数
            int instanceCount = zoneSnapshot.getInstanceCount();
            if (instanceCount == 0) {
            //如果服务器中没有服务实例，那么移除该服务区
                availableZones.remove(zone);
                limitedZoneAvailability = true;
            } else {
                double loadPerServer = zoneSnapshot.getLoadPerServer();
                 //服务区的实例平均负载小于0，或者实例故障率（断路器端口次数/实例数）大于等于阈值（默认0.99999），则去掉该服务区
                if (((double) zoneSnapshot.getCircuitTrippedCount())
                        / instanceCount >= triggeringBlackoutPercentage
                        || loadPerServer < 0) {
                    availableZones.remove(zone);
                    limitedZoneAvailability = true;
                } else {
//如果该服务区的平均负载和最大负载的差小于一定的两，则将该服务器加入到最坏服务区集合
                    if (Math.abs(loadPerServer - maxLoadPerServer) < 0.000001d) {
                        // they are the same considering double calculation
                        // round error
                        worstZones.add(zone);
                    } else if (loadPerServer > maxLoadPerServer) {
                       //否则，如果该zone的平均负载还大于最大负载
                        maxLoadPerServer = loadPerServer;
                        worstZones.clear();
                        worstZones.add(zone);
                    }
                }
            }
        }

    //如果最大的平均负载小于设定的阈值则直接返回   
 if (maxLoadPerServer < triggeringLoad && !limitedZoneAvailability) {
            // zone override is not needed here
            return availableZones;
        }
         //否则，从最好的服务器集合里面随机挑选一个
        String zoneToAvoid = randomChooseZone(snapshot, worstZones);
        if (zoneToAvoid != null) {
            availableZones.remove(zoneToAvoid);
        }
        return availableZones;

    }

CompositePredicate的apply方法
会依次调用ZoneAvoidancePredicate与AvailabilityPredicate的apply方法
那接下来就是AvailabilityPredicate的apply方法了

@Override
    public boolean apply(@Nullable PredicateKey input) {
        LoadBalancerStats stats = getLBStats();
        if (stats == null) {
            return true;
        }
        //获得关于该服务器的记录
        return !shouldSkipServer(stats.getSingleServerStat(input.getServer()));
    }
    
    
private boolean shouldSkipServer(ServerStats stats) {  
   //如果该服务器的断路器已经打开，或者他的连接数大于设定的阈值，那么就需要将服务器过滤掉      
        if ((CIRCUIT_BREAKER_FILTERING.get() && stats.isCircuitBreakerTripped()) 
                || stats.getActiveRequestsCount() >= activeConnectionsLimit.get()) {
            return true;
        }
        return false;
    }

SpringCloud之Ribbon源码分析

@RibbonClient注解

LoadBalancerClient

IRULE负载均衡的实现

ZoneAvoidanceRule 的具体实现

九、Spring之BeanFactory源码分析(一)

Spring源码分析之IoC容器初始化

ThinkPHP6源码分析之应用初始化

并发编程（五）——AbstractQueuedSynchronizer 之 ReentrantLock源码分析

Android8.1 SystemUI源码分析之电池时钟刷新

Netty源码分析（十）----- 拆包器之LineBasedFrameDecoder

MapReduce之Job提交流程源码和切片源码分析

JDK源码分析（6）之 LinkedHashMap 相关

JDK源码分析（10）之 Hashtable 相关

Django-restframework 源码分析之认证

SpringCloud之Ribbon源码分析

@RibbonClient注解

LoadBalancerClient

IRULE负载均衡的实现

ZoneAvoidanceRule 的具体实现

九、Spring之BeanFactory源码分析(一)

Spring源码分析之IoC容器初始化

ThinkPHP6源码分析之应用初始化

并发编程（五）——AbstractQueuedSynchronizer 之 ReentrantLock源码分析

Android8.1 SystemUI源码分析之 电池时钟刷新

Netty源码分析 （十）----- 拆包器之LineBasedFrameDecoder

MapReduce之Job提交流程源码和切片源码分析

JDK源码分析（6）之 LinkedHashMap 相关

JDK源码分析（10）之 Hashtable 相关

Django-restframework 源码分析之认证

Android8.1 SystemUI源码分析之电池时钟刷新

Netty源码分析（十）----- 拆包器之LineBasedFrameDecoder