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

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

// 代码实现在抽象父类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);
 }
}

@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;
}
}