Ribbon源码解析

• 1 客户端负载均衡
• • 1.1 服务端负载均衡
  • 1.2 客户端负载均衡
• 2 Ribbon架构
• • 2.1 主要组件
  • 2.2 BaseLoadBalancer
  • • 2.2.1 LoadBalancer负载均衡器
    • 2.2.2 Ping心跳检测
    • 2.2.3 Rule负载均衡策略
    • • 自定义负载均衡策略
  • 2.3 DynamicServerListLoadBalancer
  • • 2.3.1 ServerList服务列表
    • 2.3.2 ServerListUpdater服务列表更新
    • 2.3.3 ServerListFilter服务列表过滤器
• 3 服务调用流程
• • 3.1 自动装配
  • 3.2 负载均衡
  • 3.3 获取服务
  • 3.4 调用服务
• 参考

1 客户端负载均衡 Spring Cloud Ribbon是一个基于HTTP和TCP的客户端负载均衡工具，基于Netflix Ribbon实现。通过Spring Cloud的封装，可以让我们轻松地将面向服务的REST模板请求自动转换成客户端负载均衡的调用。
1.1 服务端负载均衡 负载均衡是对系统的高可用、网络压力的环节和处理能力扩容的重要手段之一。通常所说的负载均衡都是指服务端的负载均衡，其中分为硬件负载均衡和软件负载均衡。

• 硬件负载均衡：在服务器节点之间安装专门用于负载均衡的设备，例如F5。
• 软件负载均衡：通过在服务器上安装一些具有负载均衡功能或模块的软件来完成请求的分发工作。例如Nginx（工作在7层）、LVS（工作在4层）。

不论是硬件负载均衡还是软件负载均衡，只要是服务器端负载均衡都能以类似下图的架构搭建：

文章图片

服务端负载均衡，在负载均衡设备处需要维护一个下挂可用的服务端清单，通过心跳检测来剔除故障的服务端节点以保证清单中都是可以正常访问的服务端节点。当客户端发起请求到负载均衡设备时，该设备按照某种算法从维护的可用服务端清单中取出一个服务端的地址进行转发。
1.2 客户端负载均衡 客户端负载均衡和服务端负载均衡的最大差别在于服务清单的存储位置。在客户端负载均衡中，所有客户端节点都维护着自己要访问的服务清单，而这些服务清单来自于服务注册中心。
通过Spring Cloud的封装，在微服务中使用客户端负载均衡十分简单，主要包含一下两部：

• 服务提供者启动多个服务实例并注册到一个注册中心或是多个相关联的注册中心。
• 服务消费者直接通过调用被@LoadBalanced注解修饰过得RestTemplate来实现面向服务的接口调用。

2 Ribbon架构 2.1 主要组件 Ribbon主要包含6个组件，分别为：

1. LoadBalancer（负载均衡器）。
2. Rule（负载均衡策略）。
3. Ping（心跳检测）。
4. ServerList（服务列表）。
5. ServerListFilter（服务列表过滤）。
6. ServerListUpdater（服务列表更新）。

2.2 BaseLoadBalancer 2.2.1 LoadBalancer负载均衡器
负载均衡器是核心组件，其他组件均围绕它实现特定的功能。负载均衡器的核心接口是ILoadBalancer。

/** * 定义软件负载平衡器操作的接口。一个典型的负载平衡器至少需要一组服务器来 * 进行负载平衡，一个方法来标记一个特定的服务器停止运行，以及一个从现有服 * 务器列表中选择一个服务器的调用。 */ public interface ILoadBalancer { /** * 初始服务器列表。 * 此API还可以在以后添加其他API。 * 同一个逻辑服务器(端口：主机)基本上可以被多次添加（在您希望给出更多“权重”的情况下很有用） */ public void addServers(List> newServers); /** * 从负载平衡器中选择服务器。 */ public Server chooseServer(Object key); /** * 下线负载均衡器中的某个具体实例。 */ public void markServerDown(Server server); @Deprecated public List> getServerList(boolean availableOnly); /** * 返回只有已启动且可访问的服务器。 */ public List> getReachableServers(); /** * 返回所有已知的服务器，包括可访问和不可访问。 */ public List> getAllServers(); }

接口的UML图

文章图片

AbstractLoadBalancer
AbstractLoadBalancer包含大多数负载平衡实现所需的特性。典型负载平衡器的解剖结构包括:

• 可能基于特定条件进行分段的服务器（节点）列表。
• 一个通过IRule定义和实现负载平衡策略的类。
• 一个类定义并实现一种机制来确定列表中节点/服务器的适用性/可用性。

public abstract class AbstractLoadBalancer implements ILoadBalancer { /** * 表示服务实例状态的枚举 */ public enum ServerGroup{ ALL, STATUS_UP, STATUS_NOT_UP }public Server chooseServer() { return chooseServer(null); }/** * 此Loadbalancer知道的服务器的列表 */ public abstract List> getServerList(ServerGroup serverGroup); /** * 获取与负载平衡器相关的统计信息 */ public abstract LoadBalancerStats getLoadBalancerStats(); }

AbstractLoadBalancer共有两个实现类，右侧的NoOpLoadBalancer是一个空的实现类，这里可以忽略不计。这里接着继续看BaseLoadBalancer。
BaseLoadBalancer
BaseLoadBalancer是负载均衡器的基础实现类，这个类对于接口ILoadBalancer的所有方法都给予了基础的实现，除此之外还保护了很多重要的对象。
两个存储当前服务实例对象的列表，一个是包含所有服务、一个是包含正常服务。

@Monitor(name = PREFIX + "AllServerList", type = DataSourceType.INFORMATIONAL) protected volatile List> allServerList = Collections .synchronizedList(new ArrayList>()); @Monitor(name = PREFIX + "UpServerList", type = DataSourceType.INFORMATIONAL) protected volatile List> upServerList = Collections .synchronizedList(new ArrayList>());

存储负载系统器各服务实例属性和统计信息的对象

protected LoadBalancerStats lbStats;

心跳检测对象和心跳检测策略对象

protected IPing ping = null; private final static SerialPingStrategy DEFAULT_PING_STRATEGY = new SerialPingStrategy(); protected IPingStrategy pingStrategy = DEFAULT_PING_STRATEGY;

负载均衡策略对象

private final static IRule DEFAULT_RULE = new RoundRobinRule(); protected IRule rule = DEFAULT_RULE;

2.2.2 Ping心跳检测
IPing是用来向服务发起心跳检测的，通过心跳检测来判断该服务是否可用。IPing的实现类有以下几种：

1. PingUrl：使用HttpClient去get请求某个url，判断其是否alive
2. PingConstant：固定返回某服务是否可用，默认返回true
3. NoOpPing：没有任何操作，直接返回true
4. DummyPing：同样是直接返回true
5. NIWSDiscoveryPing：根据服务的实例对象InstanceInfo的InstanceStatus去判断，如果为InstanceStatus.UP，则为可用

文章图片

IPingStrategy为心跳检测策略接口，头部注释中提示若向自定义一个并行的策略，实现类必须是不可变的。

public interface IPingStrategy { boolean[] pingServers(IPing ping, Server[] servers); }

只有一个在BaseLoadBalancer中的内部实现类SerialPingStrategy，是串行策略

private static class SerialPingStrategy implements IPingStrategy { @Override public boolean[] pingServers(IPing ping, Server[] servers) { int numCandidates = servers.length; boolean[] results = new boolean[numCandidates]; logger.debug("LoadBalancer:PingTask executing [{}] servers configured", numCandidates); for (int i = 0; i < numCandidates; i++) { results[i] = false; /* Default answer is DEAD. */ try { if (ping != null) { results[i] = ping.isAlive(servers[i]); } } catch (Exception e) { logger.error("Exception while pinging Server: '{}'", servers[i], e); } } return results; } }

2.2.3 Rule负载均衡策略
IRule是在选择实例的时候的负载均衡策略对象，包含实现如下：

1. RoundRobinRule：线性轮询
2. BestAvailableRule：选择最小请求数
3. RandomRule：随机选择
4. RetryRule：轮询重试
5. WeightedResponseTimeRule：根据响应时间分配权重
6. ZoneAvoidanceRule：根据服务的分区可用性轮询

默认使用的是RoundRobinRule线性轮询。

文章图片

RandomRule实现

public class RandomRule extends AbstractLoadBalancerRule {@edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "https://www.it610.com/article/RCN_REDUNDANT_NULLCHECK_OF_NULL_VALUE") public Server choose(ILoadBalancer lb, Object key) { if (lb == null) { return null; } Server server = null; while (server == null) { if (Thread.interrupted()) { return null; } List> upList = lb.getReachableServers(); List> allList = lb.getAllServers(); int serverCount = allList.size(); if (serverCount == 0) { return null; }int index = chooseRandomInt(serverCount); server = upList.get(index); if (server == null) { Thread.yield(); continue; }if (server.isAlive()) { return (server); }server = null; Thread.yield(); }return server; }protected int chooseRandomInt(int serverCount) { return ThreadLocalRandom.current().nextInt(serverCount); } @Override public Server choose(Object key) { return choose(getLoadBalancer(), key); } @Override public void initWithNiwsConfig(IClientConfig clientConfig) { // TODO Auto-generated method stub } }

自定义负载均衡策略 若想实现自定义负载均衡策略，需要做一下几点：

1. 继承AbstractLoadBalancerRule，参考已有实现类实现自定义策略。

public class CustomeRule extends AbstractLoadBalancerRule { /* total = 0 // 当total==5以后，我们指针才能往下走， index = 0 // 当前对外提供服务的服务器地址， total需要重新置为零，但是已经达到过一个5次，我们的index = 1 */private int total = 0; // 总共被调用的次数，目前要求每台被调用5次 private int currentIndex = 0; // 当前提供服务的机器号public Server choose(ILoadBalancer lb, Object key) { if (lb == null) { return null; } Server server = null; while (server == null) { if (Thread.interrupted()) { return null; } List> upList = lb.getReachableServers(); //当前存活的服务 List> allList = lb.getAllServers(); //获取全部的服务int serverCount = allList.size(); if (serverCount == 0) { return null; }//int index = rand.nextInt(serverCount); //server = upList.get(index); if(total < 5) { server = upList.get(currentIndex); total++; }else { total = 0; currentIndex++; if(currentIndex >= upList.size()) { currentIndex = 0; } }if (server == null) { Thread.yield(); continue; }if (server.isAlive()) { return (server); }// Shouldn't actually happen.. but must be transient or a bug. server = null; Thread.yield(); } return server; }@Override public Server choose(Object key) { return choose(getLoadBalancer(), key); }@Override public void initWithNiwsConfig(IClientConfig clientConfig) { } }

2. 配置类中增加自定义规则。

@Configuration public class ConfigBean { @Bean @LoadBalanced //Ribbon 是客户端负载均衡的工具； public RestTemplate getRestTemplate() { return new RestTemplate(); }@Bean public IRule myRule() { return new CustomeRule(); //自定义负载均衡规则 } }

3. 主启动类添加 @RibbonClient 注解，name和configuration参数很重要；
   @RibbonClient(name=“microservicecloud-dept”, configuration=ConfigBean.class)

启动该消费者服务，可以看到先访问生产者1服务5次，然后访问生产者2服务5次
2.3 DynamicServerListLoadBalancer DynamicServerListLoadBalancer是BaseLoadBalancer的子类，这个类对基础负载均衡器做了扩展。扩展的功能如下：

1. 服务实例运行期间的动态更新
2. 服务实例的过滤

添加了以下3个对象

volatile ServerList serverListImpl; volatile ServerListFilter filter; protected volatile ServerListUpdater serverListUpdater;

2.3.1 ServerList服务列表
ServerList是获取服务器列表的接口。

public interface ServerList { /** * 获取初始化的服务列表 */ public List getInitialListOfServers(); /** * 获取更新后的服务列表 */ public List getUpdatedListOfServers(); }

文章图片

在DynamicServerListLoadBalancer中默认使用的服务列表实现类DomainExtractingServerList，只不过该服务列表内部还定义了一个服务列表，这个服务列表的实现类则是DiscoveryEnabledNIWSServerList。

public class DomainExtractingServerList implements ServerList { private ServerList list; // .. }

这个最终的服务列表的数据来源则主要依靠Eureka Client从注册中心获取。
2.3.2 ServerListUpdater服务列表更新
ServerListUpdater接口定义了动态服务器列表更新的策略。

public interface ServerListUpdater { //内部接口 public interface UpdateAction { //实现对服务列表的更新操作 void doUpdate(); } //启动服务更新器 void start(UpdateAction updateAction); //停止服务更新器 void stop(); //返回最近的更新时间戳 String getLastUpdate(); //返回上一次更新到现在的时间间隔（ms） long getDurationSinceLastUpdateMs(); //返回错过的更新周期数 int getNumberMissedCycles(); //返回核心线程数 int getCoreThreads(); }

它的实现类有两个：

1. PollingServerListUpdater：通过定时任务进行更新
2. EurekaNotificationServerListUpdater：利用Eureka的事件监听器来更新

文章图片

2.3.3 ServerListFilter服务列表过滤器
该接口主要用于根据一些规则过滤传入的服务实例列表。

public interface ServerListFilter { public List getFilteredListOfServers(List servers); }

该接口的实现类如下：

1. ZoneAffinityServerListFilter：基于Eureka的分区规则对服务实例的过滤
2. DefaultNIWSServerListFilter：ZoneAffinityServerListFilter的子类且没有做特殊的更新
3. ServerListSubsetFilter：通过比较服务实例的通信失败数和并发连接数来剔除那些相对不够健康的实例
4. ZonePreferenceServerListFilter：使用SpringCloud整合eureka和ribbon时默认使用的该过滤器。它实现了通过配置或eureka实例无数据的所属区域(Zone)来过滤出同区域的服务实例。

ZoneAwareLoadBalancer则是对DynamicServerListLoadBalancer的扩展，它主要增加了区域过滤的功能

文章图片

3 服务调用流程 普通项目中Ribbon的使用

@SpringBootApplication @RibbonClient(name = "provider-demo", configuration = cn.org.config.LoadBalanced.class) public class CloudDemoConsumerApplication { @Bean @LoadBalanced public RestTemplate restTemplate(){ return new RestTemplate(); } public static void main(String[] args) { SpringApplication.run(CloudDemoConsumerApplication.class, args); } }

@RibbonClient注解中通过@Import导入了配置类RibbonClientConfigurationRegistrar
观察RibbonClientConfigurationRegistrar中的registerBeanDefinitions方法：

@Override public void registerBeanDefinitions(AnnotationMetadata metadata, BeanDefinitionRegistry registry) { Map, Object> attrs = metadata .getAnnotationAttributes(RibbonClients.class.getName(), true); if (attrs != null && attrs.containsKey("value")) { AnnotationAttributes[] clients = (AnnotationAttributes[]) attrs.get("value"); for (AnnotationAttributes client : clients) { registerClientConfiguration(registry, getClientName(client), client.get("configuration")); } } if (attrs != null && attrs.containsKey("defaultConfiguration")) { String name; if (metadata.hasEnclosingClass()) { name = "default." + metadata.getEnclosingClassName(); } else { name = "default." + metadata.getClassName(); } registerClientConfiguration(registry, name, attrs.get("defaultConfiguration")); } Map, Object> client = metadata .getAnnotationAttributes(RibbonClient.class.getName(), true); String name = getClientName(client); if (name != null) { registerClientConfiguration(registry, name, client.get("configuration")); } }

分别判断是否存在注解@RibbonClients和@RibbonClient，若有则调用registerClientConfiguration方法。

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

3.1 自动装配 查看RibbonAutoConfiguration配置类。
先决条件

1. @ConditionalOnClass，当前环境必须存在这几个类： IClient, RestTemplate, AsyncRestTemplate, Ribbon
2. @RibbonClients
3. @AutoConfigureAfter，负载均衡肯定是要基于注册中心来做的，所以自动装配是在Eureka初始化完毕之后初始化的
4. @AutoConfigureBefore
5. @EnableConfigurationProperties，两个配置类，其中：
   RibbonEagerLoadProperties类中是关于Ribbon的饥饿加载模式的属性
   ServerIntrospectorProperties类中是关于安全端口的属性

装配bean

@Autowired(required = false) private List configurations = new ArrayList<>(); @Bean public SpringClientFactory springClientFactory() { SpringClientFactory factory = new SpringClientFactory(); factory.setConfigurations(this.configurations); return factory; }

3.2 负载均衡 SpringCloud把负载均衡相关的自动配置放在了spring-cloud-commons包下，负载均衡的配置类是LoadBalancerAutoConfiguration。
这个类里注册的几个核心的bean：

• LoadBalancerInterceptor
• RestTemplateCustomizer，客户端请求拦截器

用于给所有的RestTemplate增加拦截器

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

LoadBalancerInterceptor拦截器核心方法为intercept

public class LoadBalancerInterceptor implements ClientHttpRequestInterceptor { private LoadBalancerClient loadBalancer; private LoadBalancerRequestFactory requestFactory; public LoadBalancerInterceptor(LoadBalancerClient loadBalancer, LoadBalancerRequestFactory requestFactory) { this.loadBalancer = loadBalancer; this.requestFactory = requestFactory; } public LoadBalancerInterceptor(LoadBalancerClient loadBalancer) { // for backwards compatibility this(loadBalancer, new LoadBalancerRequestFactory(loadBalancer)); } @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, this.requestFactory.createRequest(request, body, execution)); } }

其中requestFactory.createRequest(request, body, execution)方法是为了把请求参数封装为request
重点关注execute方法

public T execute(String serviceId, LoadBalancerRequest request, Object hint) throws IOException { ILoadBalancer loadBalancer = getLoadBalancer(serviceId); Server server = getServer(loadBalancer, hint); 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); }

创建负载均衡器
每个Ribbon客户端的负载均衡器都是唯一的，第一行getLoadBalancer就会去创建这个负载均衡器

protected ILoadBalancer getLoadBalancer(String serviceId) { return this.clientFactory.getLoadBalancer(serviceId); }public ILoadBalancer getLoadBalancer(String name) { return getInstance(name, ILoadBalancer.class); }@Override public C getInstance(String name, Class type) { C instance = super.getInstance(name, type); if (instance != null) { return instance; } IClientConfig config = getInstance(name, IClientConfig.class); return instantiateWithConfig(getContext(name), type, config); }

如果存在缓存则从缓存中获取，如果不存在创建

static C instantiateWithConfig(AnnotationConfigApplicationContext context, Class clazz, IClientConfig config) { C result = null; try { Constructor constructor = clazz.getConstructor(IClientConfig.class); result = constructor.newInstance(config); } catch (Throwable e) { // Ignored } if (result == null) { result = BeanUtils.instantiate(clazz); if (result instanceof IClientConfigAware) { ((IClientConfigAware) result).initWithNiwsConfig(config); }if (context != null) { context.getAutowireCapableBeanFactory().autowireBean(result); } } return result; }

3.3 获取服务 RibbonLoadBalancerClient中的getServer方法

protected Server getServer(String serviceId) { return getServer(getLoadBalancer(serviceId), null); }protected Server getServer(ILoadBalancer loadBalancer) { return getServer(loadBalancer, null); }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"); }

使用具体的负载均衡器loadBalancer结合相应的负载均衡算法再加上服务列表过滤、服务健康检测等操作最后会获取的一个可用服务。
3.4 调用服务 这里在调用之前把服务封装成了RibbonServer，RibbonServer为RibbonLoadBalancerClient的内部类：

public static class RibbonServer implements ServiceInstance { private final String serviceId; private final Server server; private final boolean secure; private Map, String> metadata; public RibbonServer(String serviceId, Server server) { this(serviceId, server, false, Collections.emptyMap()); } public RibbonServer(String serviceId, Server server, boolean secure, Map, String> metadata) { this.serviceId = serviceId; this.server = server; this.secure = secure; this.metadata = https://www.it610.com/article/metadata; } // .. }

除了这几个属性外，RibbonServer还有一个方法

@Override public URI getUri() { return DefaultServiceInstance.getUri(this); }

这个方法就把服务从实例id转化为一个可调用的url了

public static URI getUri(ServiceInstance instance) { String scheme = (instance.isSecure()) ? "https" : "http"; String uri = String.format("%s://%s:%s", scheme, instance.getHost(), instance.getPort()); return URI.create(uri); }

然后就是发送http请求
参考 【Ribbon源码解析】Ribbon源码解析
《Spring Cloud微服务实战》

推荐阅读

• 

  公主连接re毛二力怎么样 毛二力属性被动技能介绍
• 

  冉闵的简介资料，冉闵的主要成就，力挽狂澜的华夏民族英雄
• 

  免疫|免疫治疗，抗癌领域升起的明星！它是如何发挥作用的？
• 

  如何看待那些想当网红的女孩子呢？
• 

  九刃,九丑在八字中代表什么意思
• 

  p30是安卓系统吗
• 

  民族风情园设计主题是什么？民族风情园景区
• 

  maxwell涡流场分析视频教程
• 

  蔗糖素是糖，蔗糖素是什么
• 

  法律对员工的辞退年龄有哪些规定？
• 

  上古王冠英雄如何升星
• 

  全款购买一套二手房,合同已经签了,现在房主反悔,该怎么办？
• 

  彭放|彭放| 感悟《论语》的人文情怀之【5·23】——不念旧恶，怨是用希
• 

  母猪菜的功效与作用
• 

  ［青春］情窦初开的少年（1）
• 

  苹果id密码在哪里输入
• 

  合同履行地不明怎么确定 合同履行不明确怎么办
• 

  像素|摩托罗拉Edge 20系列手机官宣，8月5日发布，采用1亿像素主摄
• 

  Linux下C++|Linux下C++ EPOLL TCP服务器的设计与实现
• 

  室内白天都算散射光吗

• SpringCloudAlibaba-Nacos服务注册与配置中心
• Spring|Spring Cloud Alibaba——Nacos进行多配置文件的加载
• 分布式微服务|Spring Cloud Alibaba Nacos服务注册与配置中心
• springCloud|nacos简单使用
• #|一、【SpringCloud】微服务与分布式架构
• 这篇SpringCloud GateWay 详解，你用的到
• 使用restTemplate(httpClient)实现微服务之间的调用
• 面试八股文|Spring Cloud知识
• SpringCloud之Eureka注册中心与Robbin负载均衡