Spring Cloud LoadBalancer 源码解析
前言
LoadBalancer（负载均衡器）：一种网络设备或软件机制，用于分发传入的网络流量负载到多个后端目标服务器上，依次来提高系统的可用性和性能，Spring Cloud 2020 版本以后，移除了对 Netflix 的依赖，也就移除了负载均衡器 Ribbon，Spring Cloud 官方推荐使用 Loadbalancer 替换 Ribbon，并成为了Spring Cloud负载均衡器的唯一实现。LoadBalancer也可以看做是一种进程级的负载均衡器。
LoadBalancer 引入
LoadBalancer 的引入十分简单，有封装好的 starter ，只需在 pom.xml 中引入依赖即可，如下：
<!--引入 LoadBalancer 支持-->
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-loadbalancer</artifactId><version>3.0.1</version>
</dependency>
LoadBalancerAutoConfiguration 源码分析
学习 LoadBalancer 源码我们还是从 spring-cloud-starter-loadbalancer 的 spring.factories 文件入手，spring.factories 文件内容如下：
# AutoConfiguration
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.loadbalancer.config.LoadBalancerAutoConfiguration,\
org.springframework.cloud.loadbalancer.config.BlockingLoadBalancerClientAutoConfiguration,\
org.springframework.cloud.loadbalancer.config.LoadBalancerCacheAutoConfiguration,\
org.springframework.cloud.loadbalancer.security.OAuth2LoadBalancerClientAutoConfiguration,\
org.springframework.cloud.loadbalancer.config.LoadBalancerStatsAutoConfigurationspring.factories 文件中有一个 LoadBalancerAutoConfiguration 类，我们来看看这个类做了什么。
LoadBalancerInterceptorConfig 类源码分析
LoadBalancerAutoConfiguration 类中有一个内部类 LoadBalancerInterceptorConfig，LoadBalancerInterceptorConfig 类是一个负载均衡拦截器配置类，通过 LoadBalancerInterceptorConfig 类完成负载均衡拦截器的配置。
@Configuration(proxyBeanMethods = false
)
@Conditional({LoadBalancerAutoConfiguration.RetryMissingOrDisabledCondition.class})
static class LoadBalancerInterceptorConfig {LoadBalancerInterceptorConfig() {}// 创建默认的LB拦截器@Beanpublic LoadBalancerInterceptor loadBalancerInterceptor(LoadBalancerClient loadBalancerClient, LoadBalancerRequestFactory requestFactory) {return new LoadBalancerInterceptor(loadBalancerClient, requestFactory);}//在 restTemplate 中添加 LoadBalancerInterceptor//org.springframework.cloud.client.loadbalancer.LoadBalancerAutoConfiguration.LoadBalancerInterceptorConfig#restTemplateCustomizer@Bean@ConditionalOnMissingBeanpublic RestTemplateCustomizer restTemplateCustomizer(final LoadBalancerInterceptor loadBalancerInterceptor) {return (restTemplate) -> {List<ClientHttpRequestInterceptor> list = new ArrayList(restTemplate.getInterceptors());list.add(loadBalancerInterceptor);restTemplate.setInterceptors(list);};}
}有了拦截器，接下来我们来分析拦截器的拦截方法。
LoadBalancerInterceptor#intercept 方法源码分析
LoadBalancerInterceptor#intercept 方法将 request、body、execution 包装成 LoadBalancerRequest，LoadBalancerRequest 做为调用LoadBancerClient#execute 方法的参数。
//org.springframework.cloud.client.loadbalancer.LoadBalancerInterceptor#intercept
public ClientHttpResponse intercept(final HttpRequest request, final byte[] body, final ClientHttpRequestExecution execution) throws IOException {//获取请求 uriURI originalUri = request.getURI();//获取 ServiceNameString serviceName = originalUri.getHost();Assert.state(serviceName != null, "Request URI does not contain a valid hostname: " + originalUri);//将 request body execution 包装成 LoadBalancerRequest 交给 LoadBalancerClient 执行return (ClientHttpResponse)this.loadBalancer.execute(serviceName, this.requestFactory.createRequest(request, body, execution));
}BlockingLoadBalancerClient#choose 方法源码分析
BlockingLoadBalancerClient#choose 方法主要是获取负载均衡策略，然后获取服务实例，目前负载均衡策略有两种，分别是 RandomLoadBalancer、RoundRobinLoadBalancer，默认是轮训 RoundRobinLoadBalancer。
//org.springframework.cloud.loadbalancer.blocking.client.BlockingLoadBalancerClient#choose(java.lang.String, org.springframework.cloud.client.loadbalancer.Request<T>)
public <T> ServiceInstance choose(String serviceId, Request<T> request) {//根据 serviceId 获取 ReactiveLoadBalancerReactiveLoadBalancer<ServiceInstance> loadBalancer = this.loadBalancerClientFactory.getInstance(serviceId);//loadBalancer为空判断if (loadBalancer == null) {return null;} else {//不为空 根据负载均衡策略获取服务实例 目前有两种负载均衡策略 RandomLoadBalancer RoundRobinLoadBalancer 默认是轮训 RoundRobinLoadBalancerResponse<ServiceInstance> loadBalancerResponse = (Response)Mono.from(loadBalancer.choose(request)).block();return loadBalancerResponse == null ? null : (ServiceInstance)loadBalancerResponse.getServer();}
}RandomLoadBalancer#choose 方法源码分析
RandomLoadBalancer#choose 方法是负载均衡策略随机算法的实现，随机算法实现很简单，就是根据根据服务实例列表的个数产生随机数，根据随机数获取指定的服务实例返回。
//org.springframework.cloud.loadbalancer.core.RandomLoadBalancer#choose
public Mono<Response<ServiceInstance>> choose(Request request) {//获取服务实例列表ServiceInstanceListSupplier supplier = (ServiceInstanceListSupplier)this.serviceInstanceListSupplierProvider.getIfAvailable(NoopServiceInstanceListSupplier::new);//遍历使用随机的负载均衡算法得到服务实例return supplier.get(request).next().map((serviceInstances) -> {return this.processInstanceResponse(supplier, serviceInstances);});
}//org.springframework.cloud.loadbalancer.core.RandomLoadBalancer#processInstanceResponse
private Response<ServiceInstance> processInstanceResponse(ServiceInstanceListSupplier supplier, List<ServiceInstance> serviceInstances) {//使用随机的负载算法获取服务实例Response<ServiceInstance> serviceInstanceResponse = this.getInstanceResponse(serviceInstances);if (supplier instanceof SelectedInstanceCallback && serviceInstanceResponse.hasServer()) {((SelectedInstanceCallback)supplier).selectedServiceInstance((ServiceInstance)serviceInstanceResponse.getServer());}return serviceInstanceResponse;
}//org.springframework.cloud.loadbalancer.core.getInstanceResponse#choose
private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> instances) {//服务实例为空判断if (instances.isEmpty()) {if (log.isWarnEnabled()) {log.warn("No servers available for service: " + this.serviceId);}return new EmptyResponse();} else {//以服务实例的个数来获取随机数int index = ThreadLocalRandom.current().nextInt(instances.size());//根据随机数获取服务实例ServiceInstance instance = (ServiceInstance)instances.get(index);//返回服务实例return new DefaultResponse(instance);}
}RoundRobinLoadBalancer#choose 方法源码分析
RoundRobinLoadBalancer#choose 方法是负载均衡策略轮训算法的实现，轮训算法的实现也很简单，维护了一个 position 的原子类，每次获取服务实例的时候就对 position 进行加一操作，然后使用 position 和服务实例个数进行取模，根据取模后的结果来获取服务实例返回。
//org.springframework.cloud.loadbalancer.core.RoundRobinLoadBalancer#choose
public Mono<Response<ServiceInstance>> choose(Request request) {//获取服务实例列表ServiceInstanceListSupplier supplier = (ServiceInstanceListSupplier)this.serviceInstanceListSupplierProvider.getIfAvailable(NoopServiceInstanceListSupplier::new);//使用轮训算法获取服务实例return supplier.get(request).next().map((serviceInstances) -> {return this.processInstanceResponse(supplier, serviceInstances);});
}//org.springframework.cloud.loadbalancer.core.RoundRobinLoadBalancer#processInstanceResponse
private Response<ServiceInstance> processInstanceResponse(ServiceInstanceListSupplier supplier, List<ServiceInstance> serviceInstances) {//使用轮训算法获取服务实例Response<ServiceInstance> serviceInstanceResponse = this.getInstanceResponse(serviceInstances);if (supplier instanceof SelectedInstanceCallback && serviceInstanceResponse.hasServer()) {((SelectedInstanceCallback)supplier).selectedServiceInstance((ServiceInstance)serviceInstanceResponse.getServer());}//返回服务实例return serviceInstanceResponse;
}//org.springframework.cloud.loadbalancer.core.RoundRobinLoadBalancer#getInstanceResponse
private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> instances) {//服务实例为空判断if (instances.isEmpty()) {if (log.isWarnEnabled()) {log.warn("No servers available for service: " + this.serviceId);}return new EmptyResponse();} else {//position +1 获取绝对值 int pos = Math.abs(this.position.incrementAndGet());//根据 pos 和服务实例数取模后从服务列表中获取服务实例ServiceInstance instance = (ServiceInstance)instances.get(pos % instances.size());return new DefaultResponse(instance);}
}BlockingLoadBalancerClient#execute 方法源码分析
BlockingLoadBalancerClient#execute 方法是执行 HTTP 请求的方法，该方法的主要作用就是发起 HTTP 请求和对 HTTP 请求结果的处理。
//org.springframework.cloud.loadbalancer.blocking.client.BlockingLoadBalancerClient#execute(java.lang.String, org.springframework.cloud.client.ServiceInstance, org.springframework.cloud.client.loadbalancer.LoadBalancerRequest<T>)
public <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException {//创建 DefaultResponseDefaultResponse defaultResponse = new DefaultResponse(serviceInstance);//获取当前服务的负载均衡生命周期管理实例Set<LoadBalancerLifecycle> supportedLifecycleProcessors = LoadBalancerLifecycleValidator.getSupportedLifecycleProcessors(this.loadBalancerClientFactory.getInstances(serviceId, LoadBalancerLifecycle.class), DefaultRequestContext.class, Object.class, ServiceInstance.class);//request 转换Request lbRequest = request instanceof Request ? (Request)request : new DefaultRequest();//执行 onStartRequest 方法supportedLifecycleProcessors.forEach((lifecycle) -> {lifecycle.onStartRequest(lbRequest, new DefaultResponse(serviceInstance));});try {//发起 http 请求T response = request.apply(serviceInstance);//获取响应Object clientResponse = this.getClientResponse(response);//执行 onCompletesupportedLifecycleProcessors.forEach((lifecycle) -> {lifecycle.onComplete(new CompletionContext(Status.SUCCESS, lbRequest, defaultResponse, clientResponse));});//返回return response;} catch (IOException var9) {supportedLifecycleProcessors.forEach((lifecycle) -> {lifecycle.onComplete(new CompletionContext(Status.FAILED, var9, lbRequest, defaultResponse));});throw var9;} catch (Exception var10) {supportedLifecycleProcessors.forEach((lifecycle) -> {lifecycle.onComplete(new CompletionContext(Status.FAILED, var10, lbRequest, defaultResponse));});ReflectionUtils.rethrowRuntimeException(var10);return null;}
}如有不正确的地方请各位指出纠正。