[spring-cloud: 負載均衡]-源碼分析

獲取服務列表

ServiceInstanceListSupplier

ServiceInstanceListSupplier 接口是一個提供 ServiceInstance 列表的供應者，返回一個響應式流 Flux<List<ServiceInstance>>，用于服務發現。

public interface ServiceInstanceListSupplier extends Supplier<Flux<List<ServiceInstance>>> {String getServiceId();default Flux<List<ServiceInstance>> get(Request request) {return get();}static ServiceInstanceListSupplierBuilder builder() {return new ServiceInstanceListSupplierBuilder();}}

DelegatingServiceInstanceListSupplier

DelegatingServiceInstanceListSupplier 是一個抽象類，繼承自 ServiceInstanceListSupplier，它通過委托給另一個 ServiceInstanceListSupplier 實例來實現其功能，同時支持選定服務實例的回調、初始化和銷毀操作。

public abstract class DelegatingServiceInstanceListSupplier implements ServiceInstanceListSupplier, SelectedInstanceCallback, InitializingBean, DisposableBean {protected final ServiceInstanceListSupplier delegate;public DelegatingServiceInstanceListSupplier(ServiceInstanceListSupplier delegate) {Assert.notNull(delegate, "delegate may not be null");this.delegate = delegate;}public ServiceInstanceListSupplier getDelegate() {return delegate;}@Overridepublic String getServiceId() {return delegate.getServiceId();}@Overridepublic void selectedServiceInstance(ServiceInstance serviceInstance) {if (delegate instanceof SelectedInstanceCallback selectedInstanceCallbackDelegate) {selectedInstanceCallbackDelegate.selectedServiceInstance(serviceInstance);}}@Overridepublic void afterPropertiesSet() throws Exception {if (delegate instanceof InitializingBean) {((InitializingBean) delegate).afterPropertiesSet();}}@Overridepublic void destroy() throws Exception {if (delegate instanceof DisposableBean) {((DisposableBean) delegate).destroy();}}}

負載均衡實現

ReactorLoadBalancer

ReactorLoadBalancer 是基于 Reactor 實現的響應式負載均衡器，通過 Mono<Response<T>> 異步選擇服務實例。

public interface ReactorLoadBalancer<T> extends ReactiveLoadBalancer<T> {@SuppressWarnings("rawtypes")Mono<Response<T>> choose(Request request);default Mono<Response<T>> choose() {return choose(REQUEST);}}

ReactorServiceInstanceLoadBalancer

ReactorServiceInstanceLoadBalancer 是一個標記接口，繼承自 ReactorLoadBalancer，專門用于選擇 ServiceInstance 對象的負載均衡器。

// RandomLoadBalancer, RoundRobinLoadBalancer
public interface ReactorServiceInstanceLoadBalancer extends ReactorLoadBalancer<ServiceInstance> {}

核心代碼邏輯

推薦閱讀：[spring-cloud: @LoadBalanced & @LoadBalancerClient]-源碼分析

1. BlockingLoadBalancerInterceptor

// LoadBalancerInterceptor, RetryLoadBalancerInterceptor 
public interface BlockingLoadBalancerInterceptor extends ClientHttpRequestInterceptor {}

LoadBalancerInterceptor

public class LoadBalancerInterceptor implements BlockingLoadBalancerInterceptor {private final LoadBalancerClient loadBalancer;private final LoadBalancerRequestFactory requestFactory;public LoadBalancerInterceptor(LoadBalancerClient loadBalancer, LoadBalancerRequestFactory requestFactory) {this.loadBalancer = loadBalancer;this.requestFactory = requestFactory;}public LoadBalancerInterceptor(LoadBalancerClient loadBalancer) {// for backwards compatibilitythis(loadBalancer, new LoadBalancerRequestFactory(loadBalancer));}// 重點!@Overridepublic ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution)throws IOException {URI originalUri = request.getURI();String serviceName = originalUri.getHost();Assert.state(serviceName != null, "Request URI does not contain a valid hostname: " + originalUri);return loadBalancer.execute(serviceName, requestFactory.createRequest(request, body, execution));}}

2. BlockingLoadBalancerClient

ServiceInstanceChooser

ServiceInstanceChooser 接口用于通過負載均衡器選擇與指定服務ID對應的服務實例，支持帶請求上下文的選擇。

public interface ServiceInstanceChooser {ServiceInstance choose(String serviceId);<T> ServiceInstance choose(String serviceId, Request<T> request);}

LoadBalancerClient

LoadBalancerClient 接口用于客戶端負載均衡，選擇服務實例并執行請求，同時提供將邏輯服務名重構為實際服務實例的 URI 的功能。

public interface LoadBalancerClient extends ServiceInstanceChooser {<T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException;<T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException;URI reconstructURI(ServiceInstance instance, URI original);}

// BlockingLoadBalancerClientAutoConfiguration
@SuppressWarnings({ "unchecked", "rawtypes" })
public class BlockingLoadBalancerClient implements LoadBalancerClient {// org.springframework.cloud.loadbalancer.config.LoadBalancerAutoConfiguration// LoadBalancerClientFactoryprivate final ReactiveLoadBalancer.Factory<ServiceInstance> loadBalancerClientFactory;public BlockingLoadBalancerClient(ReactiveLoadBalancer.Factory<ServiceInstance> loadBalancerClientFactory) {this.loadBalancerClientFactory = loadBalancerClientFactory;}// 重點！@Overridepublic <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException {String hint = getHint(serviceId);LoadBalancerRequestAdapter<T, TimedRequestContext> lbRequest = new LoadBalancerRequestAdapter<>(request, buildRequestContext(request, hint));Set<LoadBalancerLifecycle> supportedLifecycleProcessors = getSupportedLifecycleProcessors(serviceId);supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onStart(lbRequest));// 通過 choose 方法來選擇一個合適的 ServiceInstanceServiceInstance 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);}private <T> TimedRequestContext buildRequestContext(LoadBalancerRequest<T> delegate, String hint) {if (delegate instanceof HttpRequestLoadBalancerRequest) {HttpRequest request = ((HttpRequestLoadBalancerRequest) delegate).getHttpRequest();if (request != null) {RequestData requestData = new RequestData(request);return new RequestDataContext(requestData, hint);}}return new DefaultRequestContext(delegate, hint);}// 通過生命周期鉤子函數來管理負載均衡請求的開始與結束，并處理可能的異常，確保負載均衡的執行過程有序@Overridepublic <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException {if (serviceInstance == null) {throw new IllegalArgumentException("Service Instance cannot be null, serviceId: " + serviceId);}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);}@Overridepublic URI reconstructURI(ServiceInstance serviceInstance, URI original) {return LoadBalancerUriTools.reconstructURI(serviceInstance, original);}@Overridepublic ServiceInstance choose(String serviceId) {return choose(serviceId, REQUEST);}// 重點！通過負載均衡器同步選擇一個服務實例并返回@Overridepublic <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;}}

3. LoadBalancerRequestFactory

LoadBalancerRequestFactory 類用于創建封裝負載均衡請求的 LoadBalancerRequest 實例，支持請求轉換器和負載均衡客戶端的配置。

public class LoadBalancerRequestFactory {private final LoadBalancerClient loadBalancer;private final List<LoadBalancerRequestTransformer> transformers;public LoadBalancerRequestFactory(LoadBalancerClient loadBalancer,List<LoadBalancerRequestTransformer> transformers) {this.loadBalancer = loadBalancer;this.transformers = transformers;}public LoadBalancerRequestFactory(LoadBalancerClient loadBalancer) {this.loadBalancer = loadBalancer;transformers = new ArrayList<>();}public LoadBalancerRequest<ClientHttpResponse> createRequest(final HttpRequest request, final byte[] body,final ClientHttpRequestExecution execution) {return new BlockingLoadBalancerRequest(loadBalancer, transformers,new BlockingLoadBalancerRequest.ClientHttpRequestData(request, body, execution));}}

LoadBalancerRequestTransformer

LoadBalancerRequestTransformer 接口允許在負載均衡過程中根據不同的服務實例自定義轉換 HttpRequest，如修改請求頭、URL 等，同時通過 @Order 注解控制其執行順序。

@Order(LoadBalancerRequestTransformer.DEFAULT_ORDER)
public interface LoadBalancerRequestTransformer {int DEFAULT_ORDER = 0;HttpRequest transformRequest(HttpRequest request, ServiceInstance instance);
}

4. BlockingLoadBalancerRequest

BlockingLoadBalancerRequest 類實現了負載均衡請求接口，負責將原始 HTTP 請求封裝為負載均衡請求，并支持應用請求轉換器和執行負載均衡操作。

class BlockingLoadBalancerRequest implements HttpRequestLoadBalancerRequest<ClientHttpResponse> {private final LoadBalancerClient loadBalancer;private final List<LoadBalancerRequestTransformer> transformers;private final ClientHttpRequestData clientHttpRequestData;BlockingLoadBalancerRequest(LoadBalancerClient loadBalancer, List<LoadBalancerRequestTransformer> transformers,ClientHttpRequestData clientHttpRequestData) {this.loadBalancer = loadBalancer;this.transformers = transformers;this.clientHttpRequestData = clientHttpRequestData;}@Overridepublic ClientHttpResponse apply(ServiceInstance instance) throws Exception {HttpRequest serviceRequest = new ServiceRequestWrapper(clientHttpRequestData.request, instance, loadBalancer);if (this.transformers != null) {for (LoadBalancerRequestTransformer transformer : this.transformers) {serviceRequest = transformer.transformRequest(serviceRequest, instance);}}return clientHttpRequestData.execution.execute(serviceRequest, clientHttpRequestData.body);}@Overridepublic HttpRequest getHttpRequest() {return clientHttpRequestData.request;}static class ClientHttpRequestData {private final HttpRequest request;private final byte[] body;private final ClientHttpRequestExecution execution;ClientHttpRequestData(HttpRequest request, byte[] body, ClientHttpRequestExecution execution) {this.request = request;this.body = body;this.execution = execution;}}}