微服务生态组件之Spring Cloud LoadBalancer详解和源码分析

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
注册配置 MSE Nacos/ZooKeeper,182元/月
云原生网关 MSE Higress,422元/月
简介: Spring Cloud LoadBalancer详解和源码分析

Spring Cloud LoadBalancer目前Spring官方是放在spring-cloud-commons里,

Spring Cloud最新版本为2021.0.2

Spring Cloud LoadBalancer 官网文档地址 https://docs.spring.io/spring-cloud-commons/docs/3.1.2/reference/html/#spring-cloud-loadbalancer

Spring Cloud官网文档地址 https://docs.spring.io/spring-cloud/docs/current/reference/html/


一方面Netflix Ribbon停止更新,Spring Cloud LoadBalancer是Spring Cloud官方自己提供的客户端负载均衡器,抽象和实现,用来替代Ribbon。

常见负载均衡器分为服务端负载均衡器(如网关层均衡负载)和客户端层均衡负载。

网关层如硬件层面的F5或软件层面的LVS、或者nginx等。

客户端层就如Spring Cloud LoadBalancer,作为一个客户端去发现更新维护服务列表,自定义服务的均衡负载策略(随机、轮询、小流量的金丝雀等等)。

Spring Cloud提供了自己的客户端负载平衡器抽象和实现。对于负载均衡机制,

增加了ReactiveLoadBalancer接口,并提供了基于round-robin轮询和Random随机的实现。

为了从响应式ServiceInstanceListSupplier中选择实例,需要使用ServiceInstanceListSupplier。目前支持ServiceInstanceListSupplier的基于服务发现的实现,该实现使用类路径中的发现客户端从Service Discovery中检索可用的实例。

可以通过如下配置来禁用Spring Cloud LoadBalance

spring:

 cloud:

   loadbalancer:

     enabled: false


入门示例

前面simple-ecommerce项目创建已在父Pom引入三大父依赖,详细可以看下前面的文章<<SpringCloudAlibaba注册中心与配置中心之利器Nacos实战与源码分析>>,其中Spring Cloud的版本为2021.0.1,

前面文章也已说过,Spring Cloud Alibaba整合在spring-cloud-starter-alibaba-nacos-discovery本身就依赖spring-cloud-loadbalancer。

注意如果是Hoxton之前的版本,默认负载均衡器为Ribbon,需要移除Ribbon引用和增加配置spring.cloud.loadbalancer.ribbon.enabled: false。

如果是在Spring Boot项目中添加下面的启动器依赖,该starter也包含了Spring Boot Caching and Evictor.

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


我们使用Spring官方提供了负载均衡的客户端之一RestTemplate,RestTemplate是Spring提供的用于访问Rest服务的客户端,RestTemplate提供了多种便捷访问远程Http服务的方法,能够大大提高客户端的编写效率。默认情况下,RestTemplate默认依赖jdk的HTTP连接工具。创建RestTemplateConfig配置类,

标注 @LoadBalanced注解,默认使用的ReactiveLoadBalancer实现是RoundRobinLoadBalancer。


原理:

RestTemplate

Spring Cloud LoadBalancer源码分析我们先从RestTemplate负载均衡的简单实现来分析入手,除此之外其支持Spring Web Flux响应式编程的实现原理思想也是相同,都是通过客户端添加拦截器,在拦截器中实现负载均衡。从RestTemplate的源码中可以知道其继承自InterceptingHttpAccessor抽象类

而InterceptingHttpAccessor抽象类则提供了一个方法setInterceptors,用于设置拦截器,拦截器需要实现ClientHttpRequestInterceptor接口即可,在实际远程请求服务端接口之前会先调用拦截器的intercept方法。这里的拦截器相当于Servlet技术中的Filter功能

// 代码实现在抽象父类InterceptingHttpAccessor里
// RestTemplate.InterceptingHttpAccessor#setInterceptors
public void setInterceptors(List<ClientHttpRequestInterceptor> interceptors) {
 Assert.noNullElements(interceptors, "'interceptors' must not contain null elements");
 // Take getInterceptors() List as-is when passed in here
 if (this.interceptors != interceptors) {
  this.interceptors.clear();
  this.interceptors.addAll(interceptors);
  AnnotationAwareOrderComparator.sort(this.interceptors);
 }
}



LoadBalancerAutoConfiguration

从官网可以知道Spring Cloud LoadBalancer放在spring-cloud-commons,因此也作为其核心的@LoadBalanced注解也就是由spring-cloud-commons来实现,依据SpringBoot自动装配的原理先查看依赖包的实现逻辑,不难发现spring-cloud-commons引入了自动配置类LoadBalancerAutoConfiguration和ReactorLoadBalancerClientAutoConfiguration。

当满足上述的条件时(@Conditional为条件注解),将自动创建LoadBalancerInterceptor并注入到RestTemplate中。

LoadBalancerLnterceptor

LoadBalancerInterceptor实现了ClientHttpRequestInterceptor接口,因此也实现intercept方法,用于实现负载均衡的拦截处理。

LoadBalancerClient

LoadBalancerClient用于进行负载均衡逻辑,继承自ServiceInstanceChooser接口,从服务列表中选择出一个服务地址进行调用。在LoadBalancerClient种存在两个execute()方法,均是用来执行请求的,reconstructURI()是用来重构URL。

对于LoadBalancerClient接口Spring Cloud LoadBalancer的提供默认实现为BlockingLoadBalancerClient


@SuppressWarnings({ "unchecked", "rawtypes" })
public class BlockingLoadBalancerClient implements LoadBalancerClient {
private final ReactiveLoadBalancer.Factory<ServiceInstance> loadBalancerClientFactory;
/**
 * @deprecated in favour of
 * {@link BlockingLoadBalancerClient#BlockingLoadBalancerClient(ReactiveLoadBalancer.Factory)}
 */
@Deprecated
public BlockingLoadBalancerClient(LoadBalancerClientFactory loadBalancerClientFactory,
  LoadBalancerProperties properties) {
 this.loadBalancerClientFactory = loadBalancerClientFactory;
}
public BlockingLoadBalancerClient(ReactiveLoadBalancer.Factory<ServiceInstance> loadBalancerClientFactory) {
 this.loadBalancerClientFactory = loadBalancerClientFactory;
}
@Override
public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException {
 String hint = getHint(serviceId);
 LoadBalancerRequestAdapter<T, DefaultRequestContext> lbRequest = new LoadBalancerRequestAdapter<>(request,
   new DefaultRequestContext(request, hint));
 Set<LoadBalancerLifecycle> supportedLifecycleProcessors = getSupportedLifecycleProcessors(serviceId);
 supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onStart(lbRequest));
 ServiceInstance serviceInstance = choose(serviceId, lbRequest);
    // 选择服务
 if (serviceInstance == null) {
  supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onComplete(
    new CompletionContext<>(CompletionContext.Status.DISCARD, lbRequest, new EmptyResponse())));
  throw new IllegalStateException("No instances available for " + serviceId);
 }
 return execute(serviceId, serviceInstance, lbRequest);
}
@Override
public <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request)
  throws IOException {
 DefaultResponse defaultResponse = new DefaultResponse(serviceInstance);
 Set<LoadBalancerLifecycle> supportedLifecycleProcessors = getSupportedLifecycleProcessors(serviceId);
 Request lbRequest = request instanceof Request ? (Request) request : new DefaultRequest<>();
 supportedLifecycleProcessors
   .forEach(lifecycle -> lifecycle.onStartRequest(lbRequest, new DefaultResponse(serviceInstance)));
 try {
  T response = request.apply(serviceInstance);
  Object clientResponse = getClientResponse(response);
  supportedLifecycleProcessors
    .forEach(lifecycle -> lifecycle.onComplete(new CompletionContext<>(CompletionContext.Status.SUCCESS,
      lbRequest, defaultResponse, clientResponse)));
  return response;
 }
 catch (IOException iOException) {
  supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onComplete(
    new CompletionContext<>(CompletionContext.Status.FAILED, iOException, lbRequest, defaultResponse)));
  throw iOException;
 }
 catch (Exception exception) {
  supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onComplete(
    new CompletionContext<>(CompletionContext.Status.FAILED, exception, lbRequest, defaultResponse)));
  ReflectionUtils.rethrowRuntimeException(exception);
 }
 return null;
}
private <T> Object getClientResponse(T response) {
 ClientHttpResponse clientHttpResponse = null;
 if (response instanceof ClientHttpResponse) {
  clientHttpResponse = (ClientHttpResponse) response;
 }
 if (clientHttpResponse != null) {
  try {
   return new ResponseData(clientHttpResponse, null);
  }
  catch (IOException ignored) {
  }
 }
 return response;
}
private Set<LoadBalancerLifecycle> getSupportedLifecycleProcessors(String serviceId) {
 return LoadBalancerLifecycleValidator.getSupportedLifecycleProcessors(
   loadBalancerClientFactory.getInstances(serviceId, LoadBalancerLifecycle.class),
   DefaultRequestContext.class, Object.class, ServiceInstance.class);
}
@Override
public URI reconstructURI(ServiceInstance serviceInstance, URI original) {
 return LoadBalancerUriTools.reconstructURI(serviceInstance, original);
}
@Override
public ServiceInstance choose(String serviceId) {
 return choose(serviceId, REQUEST);
}
// 通过不同的负载均衡客户端实现选择不同的服务
@Override
public <T> ServiceInstance choose(String serviceId, Request<T> request) {
 ReactiveLoadBalancer<ServiceInstance> loadBalancer = loadBalancerClientFactory.getInstance(serviceId);
 if (loadBalancer == null) {
  return null;
 }
 Response<ServiceInstance> loadBalancerResponse = Mono.from(loadBalancer.choose(request)).block();
 if (loadBalancerResponse == null) {
  return null;
 }
 return loadBalancerResponse.getServer();
}
private String getHint(String serviceId) {
 LoadBalancerProperties properties = loadBalancerClientFactory.getProperties(serviceId);
 String defaultHint = properties.getHint().getOrDefault("default", "default");
 String hintPropertyValue = properties.getHint().get(serviceId);
 return hintPropertyValue != null ? hintPropertyValue : defaultHint;
}
}


LoadBalancerClientFactory

BlockingLoadBalancerClient中持有LoadBalancerClientFactory通过调用其getInstance方法获取具体的负载均衡客户端。通过工厂类LoadBalancerClientFactory获取具体的负载均衡器实例,后面的loadBalancer.choose(request)调用其接口choose()方法实现根据负载均衡算法选择下一个服务器完成负载均衡,而ReactiveLoadBalancer getInstance(String serviceId) 有默认实现LoadBalancerClientFactory

LoadBalancerClientFactory客户端实现了不同的负载均衡算法,比如轮询、随机等。LoadBalancerClientFactory继承自NamedContextFactory,NamedContextFactory继承ApplicationContextAware,实现Spring ApplicationContext容器操作。

ReactiveLoadBalancer

ReactiveLoadBalancer负载均衡器实现服务选择,Spring Cloud Balancer中实现了轮询RoundRobinLoadBalancer、随机RandomLoadBalancer、NacosLoadBalancer算法。

LoadBalancerClientConfiguration
如果没有显式指定负载均衡算法,默认缺省值为RoundRobinLoadBalancer


@Bean

@ConditionalOnMissingBean

public ReactorLoadBalancer<ServiceInstance> reactorServiceInstanceLoadBalancer(Environment environment,

LoadBalancerClientFactory loadBalancerClientFactory) {

String name = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME);

return new RoundRobinLoadBalancer(

  loadBalancerClientFactory.getLazyProvider(name, ServiceInstanceListSupplier.class), name);

}


LoadBalancerRequestFactory

LoadBalancerRequest工厂类调用createRequest方法用于创建LoadBalancerRequest。其内部持有LoadBalancerClient对象也即持有BlockingLoadBalancerClient。

在日常项目中,一般负载均衡都是结合Feign使用,后续我们有时间再来分析Feign整合LoadBalancer的自动配置类FeignLoadBalancerAutoConfiguration的实现



ReactorLoadBalancerClientAutoConfiguration

我们也抛一下基于WebClient的@Loadbalanced的流程的引入,首先声明负载均衡过滤器ReactorLoadBalancerClientAutoConfiguration是一个自动装配器类,在项目中引入了 WebClient 和 ReactiveLoadBalancer 类之后,自动装配流程就开始运行,它会初始化一个实现了 ExchangeFilterFunction 的实例,在后面该实例将作为过滤器被注入到WebClient。后续流程有兴趣再自行研究

自定义负载均衡器
从上面可以知道LoadBalancerClientFactory是创建客户机、负载均衡器和客户机配置实例的工厂。它根据客户端名称创建一个Spring ApplicationContext,并从中提取所需的bean。因此进入到LoadBalancerClientFactory类中,需要去实现它的子接口ReactorServiceInstanceLoadBalancer,因为去获取负载均衡器实例的时候,是通过去容器中查找ReactorServiceInstanceLoadBalancer类型的bean来实现的,可以参照RandomLoadBalancer实现代码

自定义负载均衡器

从上面可以知道LoadBalancerClientFactory是创建客户机、负载均衡器和客户机配置实例的工厂。它根据客户端名称创建一个Spring ApplicationContext,并从中提取所需的bean。因此进入到LoadBalancerClientFactory类中,需要去实现它的子接口ReactorServiceInstanceLoadBalancer,因为去获取负载均衡器实例的时候,是通过去容器中查找ReactorServiceInstanceLoadBalancer类型的bean来实现的,可以参照RandomLoadBalancer实现代码

相关实践学习
每个IT人都想学的“Web应用上云经典架构”实战
本实验从Web应用上云这个最基本的、最普遍的需求出发,帮助IT从业者们通过“阿里云Web应用上云解决方案”,了解一个企业级Web应用上云的常见架构,了解如何构建一个高可用、可扩展的企业级应用架构。
相关文章
|
15天前
|
数据可视化 Java BI
将 Spring 微服务与 BI 工具集成:最佳实践
本文探讨了 Spring 微服务与商业智能(BI)工具集成的潜力与实践。随着微服务架构和数据分析需求的增长,Spring Boot 和 Spring Cloud 提供了构建可扩展、弹性服务的框架,而 BI 工具则增强了数据可视化与实时分析能力。文章介绍了 Spring 微服务的核心概念、BI 工具在企业中的作用,并深入分析了两者集成带来的优势,如实时数据处理、个性化报告、数据聚合与安全保障。同时,文中还总结了集成过程中的最佳实践,包括事件驱动架构、集中配置管理、数据安全控制、模块化设计与持续优化策略,旨在帮助企业构建高效、智能的数据驱动系统。
将 Spring 微服务与 BI 工具集成:最佳实践
|
19天前
|
Java 数据库 数据安全/隐私保护
Spring 微服务和多租户:处理多个客户端
本文介绍了如何在 Spring Boot 微服务架构中实现多租户。多租户允许单个应用实例为多个客户提供独立服务,尤其适用于 SaaS 应用。文章探讨了多租户的类型、优势与挑战,并详细说明了如何通过 Spring Boot 的灵活配置实现租户隔离、动态租户管理及数据源路由,同时确保数据安全与系统可扩展性。结合微服务的优势,开发者可以构建高效、可维护的多租户系统。
246 127
|
15天前
|
存储 安全 Java
管理 Spring 微服务中的分布式会话
在微服务架构中,管理分布式会话是确保用户体验一致性和系统可扩展性的关键挑战。本文探讨了在 Spring 框架下实现分布式会话管理的多种方法,包括集中式会话存储和客户端会话存储(如 Cookie),并分析了它们的优缺点。同时,文章还涵盖了与分布式会话相关的安全考虑,如数据加密、令牌验证、安全 Cookie 政策以及服务间身份验证。此外,文中强调了分布式会话在提升系统可扩展性、增强可用性、实现数据一致性及优化资源利用方面的显著优势。通过合理选择会话管理策略,结合 Spring 提供的强大工具,开发人员可以在保证系统鲁棒性的同时,提供无缝的用户体验。
|
15天前
|
消息中间件 Java 数据库
Spring 微服务中的数据一致性:最终一致性与强一致性
本文探讨了在Spring微服务中实现数据一致性的策略,重点分析了最终一致性和强一致性的定义、优缺点及适用场景。结合Spring Boot与Spring Cloud框架,介绍了如何根据业务需求选择合适的一致性模型,并提供了实现建议,帮助开发者在分布式系统中确保数据的可靠性与同步性。
|
1月前
|
监控 Java API
Spring Boot 3.2 结合 Spring Cloud 微服务架构实操指南 现代分布式应用系统构建实战教程
Spring Boot 3.2 + Spring Cloud 2023.0 微服务架构实践摘要 本文基于Spring Boot 3.2.5和Spring Cloud 2023.0.1最新稳定版本,演示现代微服务架构的构建过程。主要内容包括: 技术栈选择:采用Spring Cloud Netflix Eureka 4.1.0作为服务注册中心,Resilience4j 2.1.0替代Hystrix实现熔断机制,配合OpenFeign和Gateway等组件。 核心实操步骤: 搭建Eureka注册中心服务 构建商品
349 3
|
3月前
|
JavaScript 前端开发 Java
垃圾分类管理系统基于 Spring Boot Vue 3 微服务架构实操指南
本文介绍了基于Java技术的垃圾分类管理系统开发方案与实施案例。系统采用前后端分离架构,后端使用Spring Boot框架搭配MySQL数据库,前端可选择Vue.js或Java Swing实现。核心功能模块包括垃圾分类查询、科普教育、回收预约等。文中提供了两个典型应用案例:彭湖花园小区使用的Swing桌面系统和基于Spring Boot+Vue的城市管理系统,分别满足不同场景需求。最新技术方案升级为微服务架构,整合Spring Cloud、Redis、Elasticsearch等技术,并采用Docker容器
203 0
|
16天前
|
消息中间件 Java Kafka
消息队列比较:Spring 微服务中的 Kafka 与 RabbitMQ
本文深入解析了 Kafka 和 RabbitMQ 两大主流消息队列在 Spring 微服务中的应用与对比。内容涵盖消息队列的基本原理、Kafka 与 RabbitMQ 的核心概念、各自优势及典型用例,并结合 Spring 生态的集成方式,帮助开发者根据实际需求选择合适的消息中间件,提升系统解耦、可扩展性与可靠性。
消息队列比较:Spring 微服务中的 Kafka 与 RabbitMQ
|
16天前
|
Prometheus 监控 Java
日志收集和Spring 微服务监控的最佳实践
在微服务架构中,日志记录与监控对系统稳定性、问题排查和性能优化至关重要。本文介绍了在 Spring 微服务中实现高效日志记录与监控的最佳实践,涵盖日志级别选择、结构化日志、集中记录、服务ID跟踪、上下文信息添加、日志轮转,以及使用 Spring Boot Actuator、Micrometer、Prometheus、Grafana、ELK 堆栈等工具进行监控与可视化。通过这些方法,可提升系统的可观测性与运维效率。
日志收集和Spring 微服务监控的最佳实践
|
9天前
|
监控 安全 Java
Spring Cloud 微服务治理技术详解与实践指南
本文档全面介绍 Spring Cloud 微服务治理框架的核心组件、架构设计和实践应用。作为 Spring 生态系统中构建分布式系统的标准工具箱,Spring Cloud 提供了一套完整的微服务解决方案,涵盖服务发现、配置管理、负载均衡、熔断器等关键功能。本文将深入探讨其核心组件的工作原理、集成方式以及在实际项目中的最佳实践,帮助开发者构建高可用、可扩展的分布式系统。
41 1