Spring Cloud Loadbalancer (三) 如何实现 ReactorLoadBalancer

逃离我推掉我的手 2024-03-30 13:53 52阅读 0赞

前面一个章节，我们知道 RestTemplate 负载均衡，最终是委托给了 LoadBalancerClient 去执行的，而 LoadBalancerClient 的负载均衡策略，则是交给了 ReactorLoadBalancer，是由 LoadBalancerClientConfiguration 进行配置的。

@Configuration(proxyBeanMethods = false)
    @ConditionalOnDiscoveryEnabled
    public class LoadBalancerClientConfiguration {
        
    
    	private static final int REACTIVE_SERVICE_INSTANCE_SUPPLIER_ORDER = 193827465;
    
    	@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);
    	}
    
    	@Configuration(proxyBeanMethods = false)
    	@ConditionalOnBlockingDiscoveryEnabled
    	@Order(REACTIVE_SERVICE_INSTANCE_SUPPLIER_ORDER + 1)
    	public static class BlockingSupportConfiguration {
        
    
    		@Bean
    		@ConditionalOnBean(DiscoveryClient.class)
    		@ConditionalOnMissingBean
    		@Conditional(DefaultConfigurationCondition.class)
    		public ServiceInstanceListSupplier discoveryClientServiceInstanceListSupplier(
    				ConfigurableApplicationContext context) {
        
    			return ServiceInstanceListSupplier.builder().withBlockingDiscoveryClient().withCaching().build(context);
    		}
    
    		// 忽略其他配置
    
    	}
    
    	// 忽略其他配置
    
    }

在 DiscoveryClient 条件下，默认使用的是 ServiceInstanceListSupplier.builder().withBlockingDiscoveryClient().withCaching().build(context); 构建的 ServiceInstanceListSupplier 。我们看下

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

public final class ServiceInstanceListSupplierBuilder {
        
    
    	private static final Log LOG = LogFactory.getLog(ServiceInstanceListSupplierBuilder.class);
    
    	private Creator baseCreator;
    
    	private DelegateCreator cachingCreator;
    
    	private final List<DelegateCreator> creators = new ArrayList<>();
    
    	ServiceInstanceListSupplierBuilder() {
        
    	}
    
    	
    	public ServiceInstanceListSupplierBuilder withBlockingDiscoveryClient() {
        
    		if (baseCreator != null && LOG.isWarnEnabled()) {
        
    			LOG.warn("Overriding a previously set baseCreator with a blocking DiscoveryClient baseCreator.");
    		}
    		this.baseCreator = context -> {
        
    			DiscoveryClient discoveryClient = context.getBean(DiscoveryClient.class);
    
    			return new DiscoveryClientServiceInstanceListSupplier(discoveryClient, context.getEnvironment());
    		};
    		return this;
    	}
    
    	
    	public ServiceInstanceListSupplier build(ConfigurableApplicationContext context) {
        
    		Assert.notNull(baseCreator, "A baseCreator must not be null");
    
    		ServiceInstanceListSupplier supplier = baseCreator.apply(context);
    
    		for (DelegateCreator creator : creators) {
        
    			supplier = creator.apply(context, supplier);
    		}
    
    		if (this.cachingCreator != null) {
        
    			supplier = this.cachingCreator.apply(context, supplier);
    		}
    		return supplier;
    	}
    
    	
    	public interface DelegateCreator extends
    			BiFunction<ConfigurableApplicationContext, ServiceInstanceListSupplier, ServiceInstanceListSupplier> {
        
    	}
    
    }

public class DiscoveryClientServiceInstanceListSupplier implements ServiceInstanceListSupplier {
        
    
    	/**
    	 * Property that establishes the timeout for calls to service discovery.
    	 */
    	public static final String SERVICE_DISCOVERY_TIMEOUT = "spring.cloud.loadbalancer.service-discovery.timeout";
    
    	private static final Log LOG = LogFactory.getLog(DiscoveryClientServiceInstanceListSupplier.class);
    
    	private Duration timeout = Duration.ofSeconds(30);
    
    	private final String serviceId;
    
    	private final Flux<List<ServiceInstance>> serviceInstances;
    
    	public DiscoveryClientServiceInstanceListSupplier(DiscoveryClient delegate, Environment environment) {
        
    		this.serviceId = environment.getProperty(PROPERTY_NAME);
    		resolveTimeout(environment);
    		this.serviceInstances = Flux.defer(() -> Mono.fromCallable(() -> delegate.getInstances(serviceId)))
    				.timeout(timeout, Flux.defer(() -> {
        
    					logTimeout();
    					return Flux.just(new ArrayList<>());
    				}), Schedulers.boundedElastic()).onErrorResume(error -> {
        
    					logException(error);
    					return Flux.just(new ArrayList<>());
    				});
    	}
    
    	
    	@Override
    	public String getServiceId() {
        
    		return serviceId;
    	}
    
    	@Override
    	public Flux<List<ServiceInstance>> get() {
        
    		return serviceInstances;
    	}
        
        // 忽略其他代码
    
    }

最终还是交给了 DiscoveryClient 的 List getInstances(String serviceId) 方法获取 List。

本文只展示了 ServiceInstanceListSupplier 的基本逻辑，实际上 ServiceInstanceListSupplier 提供了多种场景的使用方法，可以参考 [Spring Cloud LoadBalancer][]

[Spring Cloud LoadBalancer]: https://zhuanlan.zhihu.com/p/456948287