Spring Cloud微服务：Loadbalancer 实战

2年前 (2022) 程序员胖胖胖虎阿

288 0 0

nacos维护一个列表，但是我们请求服务不可能一个服务所有的都请求一遍，比如我做一笔转账，我找到其中一个做一次转账就够了，而不是看到有多个转账服务，都去转一次。那这个就需要“选择”，选择这个靠谁来做呢，其实就是客户端负载均衡组件 Spring Cloud Loadbalancer。

均衡负载

首先作为均衡负载的组件，要知道均衡负载是在做什么的？
说白了，我们有对应的服务集群，我们只能把所有的服务指定到对应的一台机或其中的几台机，要是这样集群也没啥特别大的意义。大家都参与进来，别可着一个用就往死里用，一核有难，八核围观的情况。

这里涉及到一个均衡负载的技术，大概就是两种：网关层均衡负载和客户端层均衡负载

网关层想我之前公司有用的F5，就是在网管层做均衡负载，就是你只管请求我，我来负责维护服务地址列表，你也不需要搞什么服务发现这些。（坏处是涉及到网关层维护，网络消耗和复杂度维护角度不简单）

客户端层就好比我们这个LoadBalancer，就是我作为一个客户端我自己去不断去发现更新维护自己的一套服务列表，我自己去定义服务的均衡负载策略，不管是随机还是计数甚至可以搞小流量的金丝雀。

LoadBalancer的工作原理

Nacos 实战部分使用 WebClient 发起服务调用的过程。加了一个注解@Loadbalanced 这个注解就是开启负载均衡功能的玄机

@Bean
@LoadBalanced
public WebClient.Builder register() {
    return WebClient.builder();
}

实际上相当于给这个WebClient创建一个过滤器，帮助WebClient去请求那个主机的服务。


Builder filter(ExchangeFilterFunction filter);

主要的过程分成三步：
第一步，声明负载均衡过滤器
ReactorLoadBalancerClientAutoConfiguration 是一个自动装配器类，我们在项目中引入了 WebClient 和 ReactiveLoadBalancer 类之后，自动装配流程就开始忙活起来了。

它会初始化一个实现了 ExchangeFilterFunction 的实例，在后面的步骤中，该实例将作为过滤器被注入到 WebClient。（看下面的代码，是通过注解来实现的）

@Configuration(proxyBeanMethods = false)
// 只要Path路径上能加载到WebClient和ReactiveLoadBalancer
// 则开启自动装配流程
@ConditionalOnClass(WebClient.class)
@ConditionalOnBean(ReactiveLoadBalancer.Factory.class)
public class ReactorLoadBalancerClientAutoConfiguration {

   // 如果开启了Loadbalancer重试功能(默认开启）
   // 则初始化RetryableLoadBalancerExchangeFilterFunction
   @ConditionalOnMissingBean
   @ConditionalOnProperty(value = "spring.cloud.loadbalancer.retry.enabled", havingValue = "true")
   @Bean
   public RetryableLoadBalancerExchangeFilterFunction retryableLoadBalancerExchangeFilterFunction(
         ReactiveLoadBalancer.Factory<ServiceInstance> loadBalancerFactory, LoadBalancerProperties properties,
         LoadBalancerRetryPolicy retryPolicy) {
      return new RetryableLoadBalancerExchangeFilterFunction(retryPolicy, loadBalancerFactory, properties);
   }
   
    // 如果关闭了Loadbalancer的重试功能
    // 则初始化ReactorLoadBalancerExchangeFilterFunction对象
    @ConditionalOnMissingBean
    @ConditionalOnProperty(value = "spring.cloud.loadbalancer.retry.enabled", havingValue = "false",
        matchIfMissing = true)
    @Bean
    public ReactorLoadBalancerExchangeFilterFunction loadBalancerExchangeFilterFunction(
        ReactiveLoadBalancer.Factory<ServiceInstance> loadBalancerFactory, LoadBalancerProperties properties) {
        return new ReactorLoadBalancerExchangeFilterFunction(loadBalancerFactory, properties);
    }
   // ...省略部分代码
}

第二步，声明后置处理器。LoadBalancerBeanPostProcessorAutoConfiguration 是第二个登场的自动装配器，它的主要作用是将第一步中创建的 ExchangeFilterFunction 拦截器实例添加到一个后置处理器（LoadBalancerWebClientBuilderBeanPostProcessor）中


// 省略部分代码
public class LoadBalancerBeanPostProcessorAutoConfiguration {

   // 内部配置类
   @Configuration(proxyBeanMethods = false)
   @ConditionalOnBean(ReactiveLoadBalancer.Factory.class)
   protected static class ReactorDeferringLoadBalancerFilterConfig {
      
      // 将第一步中创建的ExchangeFilterFunction实例封装到另一个名为
      // DeferringLoadBalancerExchangeFilterFunction的过滤器中
      @Bean
      @Primary
      DeferringLoadBalancerExchangeFilterFunction<LoadBalancedExchangeFilterFunction> reactorDeferringLoadBalancerExchangeFilterFunction(
            ObjectProvider<LoadBalancedExchangeFilterFunction> exchangeFilterFunctionProvider) {
         return new DeferringLoadBalancerExchangeFilterFunction<>(exchangeFilterFunctionProvider);
      }
   }
   
   // 将过滤器打包到后置处理器中
   @Bean
   public LoadBalancerWebClientBuilderBeanPostProcessor loadBalancerWebClientBuilderBeanPostProcessor(
         DeferringLoadBalancerExchangeFilterFunction deferringExchangeFilterFunction, ApplicationContext context) {
      return new LoadBalancerWebClientBuilderBeanPostProcessor(deferringExchangeFilterFunction, context);
   }
}

第三步，添加过滤器到 WebClient。LoadBalancerWebClientBuilderBeanPostProcessor 后置处理器开始发挥作用，将过滤器添加到 WebClient 中。注意不是所有的 WebClient 都会被注入过滤器，只有被 @Loadbalanced 注解修饰的 WebClient 实例才能享受这个待遇

通过LoadBalancer实现金丝雀测试

金丝雀测试是灰度测试的一种，就是我要在线上测试一下，但是我发的请求是给指定的或者有限范围，影响小的几台机上，其他照样是正常的业务。

参与测试的几台机都是金丝雀，只有打过标的，带有“测试流量标记”才能到这些机子上，不影响其他服务器的日常服务

Spring Cloud微服务：Loadbalancer 实战
用LoadBalancer来实现金丝雀测试，默认的两种均衡负载策略：一个是随机拼概率，一个是计数器大家都参与雨露均沾。这两个肯定是不行的，要自己定义均衡负载的策略。

编写 CanaryRule 负载均衡
创建了一个叫 CanaryRule 的负载均衡规则类，它继承自 Loadbalancer 项目的标准接口 ReactorServiceInstanceLoadBalancer。

CanaryRule 借助 Http Header 中的属性和 Nacos 服务节点的 metadata 完成测试流量的负载均衡。在这个过程里，它需要准确识别哪些请求是测试流量，并且把测试流量导向到正确的目标服务。

CanaryRule 如何识别测试流量：
识别那些是测试流量，这是金丝雀测试的一个要点。

如果 WebClient 发出一个请求，其 Header 的 key-value 列表中包含了特定的流量 Key：traffic-version，那么这个请求就被识别为一个测试请求，只能发送到特定的金丝雀服务器上。

CanaryRule 如何对测试流量做负载均衡：
包含了新的代码改动的服务器就是这个金丝雀，我会在这台服务器的 Nacos 元数据中插入同样的流量密码：traffic-version。如果 Nacos 元数据中的 traffic-version 值与测试流量 Header 中的一样，那么这个 Instance 就是我们要找的那只金丝雀。


// 可以将这个负载均衡策略单独拎出来，作为一个公共组件提供服务
@Slf4j
public class CanaryRule implements ReactorServiceInstanceLoadBalancer {
    private ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;
    private String serviceId;
    // 定义一个轮询策略的种子
    final AtomicInteger position;
    
    // ...省略构造器代码
    
    // 这个服务是Loadbalancer的标准接口，也是负载均衡策略选择服务器的入口方法
    @Override
    public Mono<Response<ServiceInstance>> choose(Request request) {
        ServiceInstanceListSupplier supplier = serviceInstanceListSupplierProvider
                .getIfAvailable(NoopServiceInstanceListSupplier::new);
        return supplier.get(request).next()
                .map(serviceInstances -> processInstanceResponse(supplier, serviceInstances, request));
    }
    
    // 省略该方法内容，本方法主要完成了对getInstanceResponse的调用
    private Response<ServiceInstance> processInstanceResponse(
    }
    
    // 根据金丝雀的规则返回目标节点
    Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> instances, Request request) {
        // 注册中心无可用实例 返回空
        if (CollectionUtils.isEmpty(instances)) {
            log.warn("No instance available {}", serviceId);
            return new EmptyResponse();
        }
        // 从WebClient请求的Header中获取特定的流量打标值
        // 注意：以下代码仅适用于WebClient调用，使用RestTemplate或者Feign则需要额外适配
        DefaultRequestContext context = (DefaultRequestContext) request.getContext();
        RequestData requestData = (RequestData) context.getClientRequest();
        HttpHeaders headers = requestData.getHeaders();
        // 获取到header中的流量标记
        String trafficVersion = headers.getFirst(TRAFFIC_VERSION);
        
        // 如果没有找到打标标记，或者标记为空，则使用RoundRobin规则进行查找
        if (StringUtils.isBlank(trafficVersion)) {
            // 过滤掉所有金丝雀测试的节点，即Nacos Metadaba中包含流量标记的节点
            // 从剩余的节点中进行RoundRobin查找
            List<ServiceInstance> noneCanaryInstances = instances.stream()
                    .filter(e -> !e.getMetadata().containsKey(TRAFFIC_VERSION))
                    .collect(Collectors.toList());
            return getRoundRobinInstance(noneCanaryInstances);
        }
        
        // 如果WelClient的Header里包含流量标记
        // 循环每个Nacos服务节点，过滤出metadata值相同的instance，再使用RoundRobin查找
        List<ServiceInstance> canaryInstances = instances.stream().filter(e -> {
            String trafficVersionInMetadata = e.getMetadata().get(TRAFFIC_VERSION);
            return StringUtils.equalsIgnoreCase(trafficVersionInMetadata, trafficVersion);
        }).collect(Collectors.toList());
        return getRoundRobinInstance(canaryInstances);
    }
    
    // 使用RoundRobin机制获取节点
    private Response<ServiceInstance> getRoundRobinInstance(List<ServiceInstance> instances) {
        // 如果没有可用节点，则返回空
        if (instances.isEmpty()) {
            log.warn("No servers available for service: " + serviceId);
            return new EmptyResponse();
        }
        
        // 每一次计数器都自动+1，实现轮询的效果
        int pos = Math.abs(this.position.incrementAndGet());
        ServiceInstance instance = instances.get(pos % instances.size());
        return new DefaultResponse(instance);
    }
}

确定测试流量，例如通过查看header里面有没有定义的TRAFFIC_VERSION标记 headers.getFirst(TRAFFIC_VERSION)
测试流量的均衡负载，就是找到那些nacos上medata有TRAFFIC_VERSION的节点。

配置负载均衡策略
我们写好了金丝雀的论衡负载的策略代码，但是不能@Configuration注解，因为这样或部署到全局


// 注意这里不要写上@Configuration注解
public class CanaryRuleConfiguration {

    @Bean
    public ReactorLoadBalancer<ServiceInstance> reactorServiceInstanceLoadBalancer(
            Environment environment,
            LoadBalancerClientFactory loadBalancerClientFactory) {
        String name = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME);
        // 在Spring上下文中声明了一个CanaryRule规则
        return new CanaryRule(loadBalancerClientFactory.getLazyProvider(name,
                ServiceInstanceListSupplier.class), name);
    }
}

写好配置类之后，我们需要在 coupon-customer-serv 的启动类上添加一个 @LoadBalancerClient 注解，将 Configuration 类和目标服务关联起来


// 发到coupon-template-serv的调用，使用CanaryRuleConfiguration中定义的负载均衡Rule
@LoadBalancerClient(value = "coupon-template-serv", configuration = CanaryRuleConfiguration.class)
public class Application {
   // xxx省略方法
}

测试流量打标
掐你按方案定好了，要给数据打标了

测试流量打标的方法有很多种，比如添加一个特殊的 key-value 到 Http header，或者塞一个值到 RPC Context 中。为了方便演示，我这里采用了一种更为简单的方式，直接在用户领券接口的请求参数对象 RequestCoupon 中添加了一个 trafficVersion 成员变量，用来标识测试流量。


public class RequestCoupon {
  //.. 省略其他成员变量

  // Loadbalancer - 用作测试流量打标
  private String trafficVersion;
}


@Override
public Coupon requestCoupon(RequestCoupon request) {
    CouponTemplateInfo templateInfo = webClientBuilder.build().get()
            .uri("http://coupon-template-serv/template/getTemplate?id=" + request.getCouponTemplateId())
            // 将流量标记传入WebClient请求的Header中
            .header(TRAFFIC_VERSION, request.getTrafficVersion())
            .retrieve()
            .bodyToMono(CouponTemplateInfo.class)
            .block();
 
    // xxx 省略以下代码           
}

nacos也要设置medata
Spring Cloud微服务：Loadbalancer 实战
设置traffic-version