spring cloud自定义loaderbalancer实现流量就近访问

spring cloud自定义loaderbalancer实现流量就近访问

背景

文旅项目处于稳定需要，目前每个服务部署了多个实例，分布在不同的结点；同时为了应对甲方的稳定性需要，接下来会上云

这样，每个服务的多个节点会跨网络，甚至是不同的运营商，这样会带来一些问题

基于现有的负载均衡策略，会出现跨网络节点互相访问，这样带来的超时会比较明显，因此需要解决这个问题

目标

实现流量的就近访问

方案

基于spring cloud loadbalancer

代码分析

先看spring-cloud-loadbalancer下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.LoadBalancerStatsAutoConfiguration

重点关注org.springframework.cloud.loadbalancer.config.LoadBalancerAutoConfiguration

@Configuration(
    proxyBeanMethods = false
)
@LoadBalancerClients // 重点看这里
@EnableConfigurationProperties({LoadBalancerClientsProperties.class})
@AutoConfigureBefore({ReactorLoadBalancerClientAutoConfiguration.class, LoadBalancerBeanPostProcessorAutoConfiguration.class})
@ConditionalOnProperty(
    value = {"spring.cloud.loadbalancer.enabled"},
    havingValue = "true",
    matchIfMissing = true
)
public class LoadBalancerAutoConfiguration {
}

接着看org.springframework.cloud.loadbalancer.annotation.LoadBalancerClients

@Configuration(
    proxyBeanMethods = false
)
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE})
@Documented
@Import({LoadBalancerClientConfigurationRegistrar.class})
public @interface LoadBalancerClients {
    LoadBalancerClient[] value() default {};
 
    Class<?>[] defaultConfiguration() default {};
}

通过defaultConfiguration可以指定默认配置，如果不指定，会用到org.springframework.cloud.loadbalancer.annotation.LoadBalancerClientConfiguration

@Configuration(
    proxyBeanMethods = false
)
@ConditionalOnDiscoveryEnabled
public class LoadBalancerClientConfiguration {
    private static final int REACTIVE_SERVICE_INSTANCE_SUPPLIER_ORDER = 193827465;
 
    public LoadBalancerClientConfiguration() {
    }
 
    @Bean
    @ConditionalOnMissingBean  // 默认走这里，实际走的是轮询
    public ReactorLoadBalancer<ServiceInstance> reactorServiceInstanceLoadBalancer(Environment environment, LoadBalancerClientFactory loadBalancerClientFactory) {
        String name = environment.getProperty("loadbalancer.client.name");
        return new RoundRobinLoadBalancer(loadBalancerClientFactory.getLazyProvider(name, ServiceInstanceListSupplier.class), name);
    }
}

为什么默认会走到org.springframework.cloud.loadbalancer.annotation.LoadBalancerClientConfiguration

看org.springframework.cloud.loadbalancer.support.LoadBalancerClientFactory

public class LoadBalancerClientFactory extends NamedContextFactory<LoadBalancerClientSpecification> implements Factory<ServiceInstance> {
    private static final Log log = LogFactory.getLog(LoadBalancerClientFactory.class);
    public static final String NAMESPACE = "loadbalancer";
    public static final String PROPERTY_NAME = "loadbalancer.client.name";
    private final LoadBalancerClientsProperties properties;
 
    /** @deprecated */
    @Deprecated
    public LoadBalancerClientFactory() {
        this((LoadBalancerClientsProperties)null);
    }
 
    public LoadBalancerClientFactory(LoadBalancerClientsProperties properties) {
        super(LoadBalancerClientConfiguration.class, "loadbalancer", "loadbalancer.client.name"); // 主要是这里
        this.properties = properties;
    }
    // ...
}

如果还想继续往下跟，可以看org.springframework.cloud.gateway.filter.ReactiveLoadBalancerClientFilter

public class ReactiveLoadBalancerClientFilter implements GlobalFilter, Ordered {
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        URI url = (URI)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
        String schemePrefix = (String)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_SCHEME_PREFIX_ATTR);
        if (url != null && ("lb".equals(url.getScheme()) || "lb".equals(schemePrefix))) {
            ServerWebExchangeUtils.addOriginalRequestUrl(exchange, url);
            if (log.isTraceEnabled()) {
                log.trace(ReactiveLoadBalancerClientFilter.class.getSimpleName() + " url before: " + url);
            }
 
            URI requestUri = (URI)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
            String serviceId = requestUri.getHost();
            Set<LoadBalancerLifecycle> supportedLifecycleProcessors = LoadBalancerLifecycleValidator.getSupportedLifecycleProcessors(this.clientFactory.getInstances(serviceId, LoadBalancerLifecycle.class), RequestDataContext.class, ResponseData.class, ServiceInstance.class);   // 重点在这里
            DefaultRequest<RequestDataContext> lbRequest = new DefaultRequest(new RequestDataContext(new RequestData(exchange.getRequest()), this.getHint(serviceId)));
            // ...
    }
    
    // ...
 
    private Mono<Response<ServiceInstance>> choose(Request<RequestDataContext> lbRequest, String serviceId, Set<LoadBalancerLifecycle> supportedLifecycleProcessors) {
        ReactorLoadBalancer<ServiceInstance> loadBalancer = (ReactorLoadBalancer)this.clientFactory.getInstance(serviceId, ReactorServiceInstanceLoadBalancer.class);  // 以及这里
        if (loadBalancer == null) {
            throw new NotFoundException("No loadbalancer available for " + serviceId);
        } else {
            supportedLifecycleProcessors.forEach((lifecycle) -> {
                lifecycle.onStart(lbRequest);
            });
            return loadBalancer.choose(lbRequest);
        }
     }
     // ...
}

具体实施

替换默认配置

@Slf4j
@ConditionalOnDiscoveryEnabled
public class NearbyLoadBalancerConfig {
 
    public NearbyLoadBalancerConfig() {
    }
 
    @PostConstruct
    public void init() {
        log.info("NearbyLoadBalancerConfig inited");
    }
 
    @Bean
    @ConditionalOnMissingBean
    public ReactorLoadBalancer<ServiceInstance> reactorServiceInstanceLoadBalancer(LoadBalancerClientFactory loadBalancerClientFactory, Environment environment, InetUtils inetUtils) {
        String name = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME);
        return new NearbyServiceLoadBalancer(loadBalancerClientFactory.getLazyProvider(name, ServiceInstanceListSupplier.class), name, environment, inetUtils);
    }
}

应用端通过@LoadBalancerClients 注解修改默认配置

@LoadBalancerClients(defaultConfiguration = NearbyLoadBalancerConfig.class)
@SpringBootApplication
public class Application {
 
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
        log.info("......服务启动成功!");
    }
 
}

负载策略

1、优先访问通ip节点，也可以访问同网段节点（此处未做实现）

2、如果没有通ip节点，轮询选择一个

3、开关控制，关闭之后继续走轮询

代码如下

@Slf4j
public class NearbyServiceLoadBalancer implements ReactorServiceInstanceLoadBalancer {
 
    public static final String SWITCH_KEY = "discovery.nearby.enable"; // 开关配置key
  
    private ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;
    private final String serviceId;
    private Environment environment;
 
    private final AtomicInteger position;
    private final String nearbyIp; // 本机ip
 
    private RoundRobinLoadBalancer roundRobinLoadBalancer;
 
 
    public NearbyServiceLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider, String serviceId, Environment environment, InetUtils inetUtils) {
        this.serviceInstanceListSupplierProvider = serviceInstanceListSupplierProvider;
        this.serviceId = serviceId;
        this.environment = environment;
 
        this.position = new AtomicInteger(1000);
        this.nearbyIp = inetUtils.findFirstNonLoopbackHostInfo().getIpAddress();
 
        this.roundRobinLoadBalancer = new RoundRobinLoadBalancer(serviceInstanceListSupplierProvider, serviceId);
 
        log.info("NearbyServiceLoadBalancer.construct, serviceId={}, nearbyIp={}", this.serviceId, this.nearbyIp);
    }
 
    /**
     * 判断是否走就近访问策略
     **/
    private boolean disableNearby() {
        return StringUtils.isBlank(this.nearbyIp)
                || !Boolean.TRUE.toString().equals(environment.getProperty(SWITCH_KEY));
    }
 
    @Override
    public Mono<Response<ServiceInstance>> choose(Request request) {
        if (disableNearby()) {
            return this.roundRobinLoadBalancer.choose(request);
        }
 
        ServiceInstanceListSupplier supplier = this.serviceInstanceListSupplierProvider.getIfAvailable(NoopServiceInstanceListSupplier::new);
        return supplier.get(request).next().map((serviceInstances) -> this.processInstanceResponse(supplier, serviceInstances));
    }
 
    protected Response<ServiceInstance> processInstanceResponse(ServiceInstanceListSupplier supplier, List<ServiceInstance> serviceInstances) {
        Response<ServiceInstance> serviceInstanceResponse = this.getInstanceResponse(serviceInstances);
        if (supplier instanceof SelectedInstanceCallback && serviceInstanceResponse.hasServer()) {
            ((SelectedInstanceCallback) supplier).selectedServiceInstance(serviceInstanceResponse.getServer());
        }
 
        return serviceInstanceResponse;
    }
 
    private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> instances) {
        ServiceInstance instance = selectInstance(instances);
        return instance == null ? new EmptyResponse() : new DefaultResponse(instance);
    }
 
    /**
     * 选择最合适的节点
     **/
    protected ServiceInstance selectInstance(List<ServiceInstance> instances) {
        if (log.isDebugEnabled()) {
            log.debug("NearbyServiceLoadBalancer.selectInstance, instances={}", formatServiceInstances(instances));
        }
 
        if (CollectionUtils.isEmpty(instances)) {
            return null;
        }
 
        // 只有一个候选节点，不用走策略，直接使用
        if (instances.size() == 1) {
            return instances.get(0);
        }
 
        ServiceInstance instance = instances.stream()
                .filter(o -> this.nearbyIp.equals(o.getHost()))
                .findFirst()
                .orElse(null);
        if (instance == null) {
            int pos = Math.abs(this.position.incrementAndGet());
            instance = instances.get(pos % instances.size());
        }
 
        if (log.isDebugEnabled()) {
            log.debug("NearbyServiceLoadBalancer.selectInstance, instance={}", formatServiceInstance(instance));
        }
 
        return instance;
    }
 
    private String formatServiceInstances(List<ServiceInstance> serviceInstances) {
        if (serviceInstances == null) {
            return null;
        }
 
        if (serviceInstances.size() == 0) {
            return "[]";
        }
 
        return "[" + serviceInstances.stream().map(this::formatServiceInstance).reduce((s1, s2) -> s1 + "," + s2).get() + "]";
    }
 
    private String formatServiceInstance(ServiceInstance serviceInstance) {
        if (serviceInstance == null) {
            return null;
        }
 
        return "{instanceId=" + serviceInstance.getInstanceId() + "}";
    }
}

功能开启

discovery:
  nearby:
    enable: true