Spring Cloud升级之路 - Hoxton - 3. 负载均衡从ribbon替换成spring-cloud-loadbalancer

谁践踏了优雅 2023-02-15 03:33 31阅读 0赞

> 本系列示例与胶水代码地址： https://github.com/HashZhang/spring-cloud-scaffold

## 负载均衡Ribbon替换成Spring Cloud Load Balancer ##

`Spring Cloud Load Balancer`并不是一个独立的项目，而是`spring-cloud-commons`其中的一个模块。 项目中用了`Eureka`以及相关的 starter，想完全剔除`Ribbon`的相关依赖基本是不可能的，Spring 社区的人也是看到了这一点，通过配置去关闭`Ribbon`启用`Spring-Cloud-LoadBalancer`。

spring.cloud.loadbalancer.ribbon.enabled=false

关闭ribbon之后，Spring Cloud LoadBalancer就会加载成为默认的负载均衡器。

Spring Cloud LoadBalancer 结构如下所示：

![image][]

其中：

1.  全局只有一个 `BlockingLoadBalancerClient`，负责执行所有的负载均衡请求。
2.  `BlockingLoadBalancerClient`从`LoadBalancerClientFactory`里面加载对应微服务的负载均衡配置。
3.  每个微服务下有独自的`LoadBalancer`，`LoadBalancer`里面包含负载均衡的算法，例如`RoundRobin`.根据算法，从`ServiceInstanceListSupplier`返回的实例列表中选择一个实例返回。

### 1. 实现`zone`隔离 ###

要想实现`zone`隔离，应该从`ServiceInstanceListSupplier`里面做手脚。默认的实现里面有关于`zone`隔离的`ServiceInstanceListSupplier` \-> `org.springframework.cloud.loadbalancer.core.ZonePreferenceServiceInstanceListSupplier`:

private List<ServiceInstance> filteredByZone(List<ServiceInstance> serviceInstances) {
    	if (zone == null) {
    		zone = zoneConfig.getZone();
    	}
    	//如果zone不为null，并且该zone下有存活实例，则返回这个实例列表
    	//否则，返回所有的实例
    	if (zone != null) {
    		List<ServiceInstance> filteredInstances = new ArrayList<>();
    		for (ServiceInstance serviceInstance : serviceInstances) {
    			String instanceZone = getZone(serviceInstance);
    			if (zone.equalsIgnoreCase(instanceZone)) {
    				filteredInstances.add(serviceInstance);
    			}
    		}
    		if (filteredInstances.size() > 0) {
    			return filteredInstances;
    		}
    	}
    	// If the zone is not set or there are no zone-specific instances available,
    	// we return all instances retrieved for given service id.
    	return serviceInstances;
    }

这里对于没指定`zone`或者该`zone`下没有存活实例的情况下，会返回所有查到的实例，不区分`zone`。这个不符合我们的要求，所以我们修改并实现下我们自己的[com.github.hashjang.hoxton.service.consumer.config.SameZoneOnlyServiceInstanceListSupplier][]:

private List<ServiceInstance> filteredByZone(List<ServiceInstance> serviceInstances) {
        if (zone == null) {
            zone = zoneConfig.getZone();
        }
        if (zone != null) {
            List<ServiceInstance> filteredInstances = new ArrayList<>();
            for (ServiceInstance serviceInstance : serviceInstances) {
                String instanceZone = getZone(serviceInstance);
                if (zone.equalsIgnoreCase(instanceZone)) {
                    filteredInstances.add(serviceInstance);
                }
            }
            if (filteredInstances.size() > 0) {
                return filteredInstances;
            }
        }
        //如果没找到就返回空列表，绝不返回其他集群的实例
        return List.of();
    }

然后我们来看一下默认的 `Spring Cloud LoadBalancer` 提供的 `LoadBalancer` ，它是带缓存的：

`org.springframework.cloud.loadbalancer.annotation.LoadBalancerClientConfiguration`

@Bean
    @ConditionalOnBean(ReactiveDiscoveryClient.class)
    @ConditionalOnMissingBean
    public ServiceInstanceListSupplier discoveryClientServiceInstanceListSupplier(
    		ReactiveDiscoveryClient discoveryClient, Environment env,
    		ApplicationContext context) {
    	DiscoveryClientServiceInstanceListSupplier delegate = new DiscoveryClientServiceInstanceListSupplier(
    			discoveryClient, env);
    	ObjectProvider<LoadBalancerCacheManager> cacheManagerProvider = context
    			.getBeanProvider(LoadBalancerCacheManager.class);
    	if (cacheManagerProvider.getIfAvailable() != null) {
    		return new CachingServiceInstanceListSupplier(delegate,
    				cacheManagerProvider.getIfAvailable());
    	}
    	return delegate;
    }

`DiscoveryClientServiceInstanceListSupplier`每次从`Eureka`上面拉取实例列表，`CachingServiceInstanceListSupplier`提供了缓存，这样不必每次从`Eureka`上面拉取。可以看出`CachingServiceInstanceListSupplier`是一种代理模式的实现，和`SameZoneOnlyServiceInstanceListSupplier`的模式是一样的。

我们来组装我们的`ServiceInstanceListSupplier`，由于我们是同步的环境，只用实现同步的`ServiceInstanceListSupplier`就行了。

public class CommonLoadBalancerConfig {
    
        /**
         * 同步环境下的ServiceInstanceListSupplier
         * SameZoneOnlyServiceInstanceListSupplier限制仅返回同一个zone下的实例（注意）
         * CachingServiceInstanceListSupplier启用缓存，不每次访问eureka请求实例列表
         *
         * @param discoveryClient
         * @param env
         * @param zoneConfig
         * @param context
         * @return
         */
        @Bean
        @Order(Integer.MIN_VALUE)
        public ServiceInstanceListSupplier discoveryClientServiceInstanceListSupplier(
                DiscoveryClient discoveryClient, Environment env,
                LoadBalancerZoneConfig zoneConfig,
                ApplicationContext context) {
            ServiceInstanceListSupplier delegate = new SameZoneOnlyServiceInstanceListSupplier(
                    new DiscoveryClientServiceInstanceListSupplier(discoveryClient, env),
                    zoneConfig
            );
            ObjectProvider<LoadBalancerCacheManager> cacheManagerProvider = context
                    .getBeanProvider(LoadBalancerCacheManager.class);
            if (cacheManagerProvider.getIfAvailable() != null) {
                return new CachingServiceInstanceListSupplier(
                        delegate,
                        cacheManagerProvider.getIfAvailable()
                );
            }
            return delegate;
        }
    }

### 2. 实现下一次重试的时候，如果存在其他实例，则一定会重试与本次不同的其他实例 ###

默认的`RoundRobinLoadBalancer`，其中的轮询`position`，是一个`Atomic`类型的，在某个微服务的调用请求下，所有线程，所有请求共用（调用其他的每个微服务会创建一个`RoundRobinLoadBalancer`）。在使用的时候，会有这样的一个问题：

*  假设某个微服务有两个实例，实例 A 和实例 B
 *  某次请求 X 发往实例 A，position = position + 1
 *  在请求没有返回时，请求 Y 到达，发往实例 B，position = position + 1
 *  请求 A 失败，重试，重试的实例还是实例 A

这样在重试的情况下，某个请求的重试可能会发送到上一次的实例进行重试，这不是我们想要的。针对这个，我提了个Issue：[Enhance RoundRoubinLoadBalancer position][]。我修改的思路是，我们需要一个单次请求隔离的`position`，这个`position`对于实例个数取余得出请求要发往的实例。那么如何进行请求隔离呢？

首先想到的是线程隔离，但是这个是不行的。Spring Cloud LoadBalancer 底层运用了 reactor 框架，导致实际承载选择实例的线程，不是业务线程，而是 reactor 里面的线程池，如图所示：  
![image][image 1]  
所以，不能用`ThreadLocal`的方式实现`position`。

由于我们用到了`sleuth`，一般请求的`context`会传递其中的`traceId`，我们根据这个`traceId`区分不同的请求，实现我们的 `LoadBalancer`：

[`RoundRobinBaseOnTraceIdLoadBalancer`][RoundRobinBaseOnTraceIdLoadBalancer]

//这个超时时间，需要设置的比你的请求的 connectTimeout + readTimeout 长
    private final LoadingCache<Long, AtomicInteger> positionCache = Caffeine.newBuilder().expireAfterWrite(3, TimeUnit.SECONDS).build(k -> new AtomicInteger(ThreadLocalRandom.current().nextInt(0, 1000)));
    
    private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> serviceInstances) {
        if (serviceInstances.isEmpty()) {
            log.warn("No servers available for service: " + this.serviceId);
            return new EmptyResponse();
        }
        //如果没有 traceId，就生成一个新的，但是最好检查下为啥会没有
        //是不是 MQ 消费这种没有主动生成 traceId 的情况，最好主动生成下。
        Span currentSpan = tracer.currentSpan();
        if (currentSpan == null) {
            currentSpan = tracer.newTrace();
        }
        long l = currentSpan.context().traceId();
        int seed = positionCache.get(l).getAndIncrement();
        return new DefaultResponse(serviceInstances.get(seed % serviceInstances.size()));
    }

### 3. 替换默认的负载均衡相关 Bean 实现 ###

我们要用上面的两个类替换默认的实现，先编写一个配置类：

public class CommonLoadBalancerConfig {
    
        private volatile boolean isValid = false;
    
        /**
         * 同步环境下的ServiceInstanceListSupplier
         * SameZoneOnlyServiceInstanceListSupplier限制仅返回同一个zone下的实例（注意）
         * CachingServiceInstanceListSupplier启用缓存，不每次访问eureka请求实例列表
         *
         * @param discoveryClient
         * @param env
         * @param zoneConfig
         * @param context
         * @return
         */
        @Bean
        @Order(Integer.MIN_VALUE)
        public ServiceInstanceListSupplier discoveryClientServiceInstanceListSupplier(
                DiscoveryClient discoveryClient, Environment env,
                LoadBalancerZoneConfig zoneConfig,
                ApplicationContext context) {
            isValid = true;
            ServiceInstanceListSupplier delegate = new SameZoneOnlyServiceInstanceListSupplier(
                    new DiscoveryClientServiceInstanceListSupplier(discoveryClient, env),
                    zoneConfig
            );
            ObjectProvider<LoadBalancerCacheManager> cacheManagerProvider = context
                    .getBeanProvider(LoadBalancerCacheManager.class);
            if (cacheManagerProvider.getIfAvailable() != null) {
                return new CachingServiceInstanceListSupplier(
                        delegate,
                        cacheManagerProvider.getIfAvailable()
                );
            }
            return delegate;
        }
    
        @Bean
        public ReactorLoadBalancer<ServiceInstance> reactorServiceInstanceLoadBalancer(
                Environment environment,
                ServiceInstanceListSupplier serviceInstanceListSupplier,
                Tracer tracer) {
            if (!isValid) {
                throw new IllegalStateException("should use the ServiceInstanceListSupplier in this configuration, please check config");
            }
            String name = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME);
            return new RoundRobinBaseOnTraceIdLoadBalancer(
                    name,
                    serviceInstanceListSupplier,
                    tracer
            );
        }
    }

然后，**指定默认的负载均衡配置采取这个配置**, 通过注解：

@LoadBalancerClients(defaultConfiguration = {CommonLoadBalancerConfig.class})

[image]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly96aHhoYXNoLWJsb2cub3NzLWNuLWJlaWppbmcuYWxpeXVuY3MuY29tL3NwcmluZy1jbG91ZCVFNSU4RCU4NyVFNyVCQSVBNyVFNCVCOSU4QiVFOCVCNyVBRi9ob3h0b24vU3ByaW5nQ2xvdWRMb2FkQmFsYW5jZXIlRTclQkIlOTMlRTYlOUUlODQucG5n?x-oss-process=image/format,png
[com.github.hashjang.hoxton.service.consumer.config.SameZoneOnlyServiceInstanceListSupplier]: https://github.com/HashZhang/spring-cloud-scaffold/blob/master/spring-cloud-hoxton/spring-cloud-hoxton-service-consumer/src/main/java/com/github/hashjang/hoxton/service/consumer/config/SameZoneOnlyServiceInstanceListSupplier.java
[Enhance RoundRoubinLoadBalancer position]: https://github.com/spring-cloud/spring-cloud-commons/issues/747
[image 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly96aHhoYXNoLWJsb2cub3NzLWNuLWJlaWppbmcuYWxpeXVuY3MuY29tL3NwcmluZy1jbG91ZCVFNSU4RCU4NyVFNyVCQSVBNyVFNCVCOSU4QiVFOCVCNyVBRi9ob3h0b24vJUU4JUFGJUI3JUU2JUIxJTgyJUU5JTgwJTg5JUU2JThCJUE5JUU1JUFFJTlFJUU0JUJFJThCJUU3JUJBJUJGJUU3JUE4JThCLnBuZw?x-oss-process=image/format,png
[RoundRobinBaseOnTraceIdLoadBalancer]: https://github.com/HashZhang/spring-cloud-scaffold/blob/master/spring-cloud-hoxton/spring-cloud-hoxton-service-consumer/src/main/java/com/github/hashjang/hoxton/service/consumer/config/RoundRobinBaseOnTraceIdLoadBalancer.java