按权重分配流量负载均衡

SpringCloud新版本（2021.x.x）中负载均衡器用LoadBalancer替代了Ribbon，默认只提供了2种负载均衡策略：RandomLoadBalancer 和 RoundRobinLoadBalancer。
SpringCloud Alibaba Nacos 2021.1版本提供了基于Nacos注册中心的轮询策略 NacosLoadBalancer 是基于权重的策略。

• NacosLoadBalancer的权重策略默认是关闭的。仅识别流量值为0（不引入流量）和非0（引入流量），不支持按照Nacos实例中的流量值进行流量负载均衡。 如果要使用基于权重的负载策略要手动开启。

# 引入依赖loadbalancer依赖
<dependency>
    <groupId>org.springframework.cloud</groupId>
     <artifactId>spring-cloud-starter-loadbalancer</artifactId>
 </dependency>

#开启nacos的负载均衡策略
spring.cloud.loadbalancer.nacos.enabled=true

• nacos中设置服务对应的权重，就可以生效了。

实例级别的配置。权重为浮点数。权重越大，分配给该实例的流量越大。

• 具体实现，可以查看nacos源码的 NacosLoadBalancer类相关信息。

自定义负载均衡策略

@Configuration
@LoadBalancerClients(defaultConfiguration = {SpringBeanConfiguration.class})
public class SpringBeanConfiguration {

    @Bean
    @LoadBalanced
    public RestTemplate restTemplate(){
        return new RestTemplate();
    }


    @Bean
    public ReactorLoadBalancer<ServiceInstance> nacosLoadBalancer(Environment environment,
                                                                  LoadBalancerClientFactory loadBalancerClientFactory,
                                                                  NacosDiscoveryProperties nacosDiscoveryProperties) {
        String name = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME);
        return new NacosSameClusterWeightedRule(
                loadBalancerClientFactory.getLazyProvider(name,
                        ServiceInstanceListSupplier.class),
                name,nacosDiscoveryProperties);
    }
}

@Slf4j
// 自定义负载均衡实现需要实现 ReactorServiceInstanceLoadBalancer 接口 以及重写choose方法
public class NacosSameClusterWeightedRule implements ReactorServiceInstanceLoadBalancer {

    // 注入当前服务的nacos的配置信息
    private final NacosDiscoveryProperties nacosDiscoveryProperties;

    // loadbalancer 提供的访问当前服务的名称
    final String serviceId;
    
    // loadbalancer 提供的访问的服务列表
    ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;

    public NacosSameClusterWeightedRule(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider, String serviceId, NacosDiscoveryProperties nacosDiscoveryProperties) {
        this.serviceId = serviceId;
        this.serviceInstanceListSupplierProvider = serviceInstanceListSupplierProvider;
        this.nacosDiscoveryProperties = nacosDiscoveryProperties;
    }

/**
     * 服务器调用负载均衡时调的放啊
     * 此处代码内容与 RandomLoadBalancer 一致
     */
    public Mono<Response<ServiceInstance>> choose(Request request) {
        System.out.println("choose.request:"+ request);
        ServiceInstanceListSupplier supplier = this.serviceInstanceListSupplierProvider.getIfAvailable(NoopServiceInstanceListSupplier::new);
        return supplier.get(request).next().map((serviceInstances) -> this.processInstanceResponse(supplier, serviceInstances));
    }
    
    /**
     * 对负载均衡的服务进行筛选的方法
     * 此处代码内容与 RandomLoadBalancer 一致
     */
    private Response<ServiceInstance> processInstanceResponse(ServiceInstanceListSupplier supplier, List<ServiceInstance> serviceInstances) {
        System.out.println("processInstanceResponse.serviceInstances:"+ serviceInstances);
        Response<ServiceInstance> serviceInstanceResponse = this.getInstanceResponse(serviceInstances);
        if (supplier instanceof SelectedInstanceCallback && serviceInstanceResponse.hasServer()) {
            ((SelectedInstanceCallback)supplier).selectedServiceInstance(serviceInstanceResponse.getServer());
        }

        return serviceInstanceResponse;
    }

    /**
     * 对负载均衡的服务进行筛选的方法
     * 自定义
     * 此处的 instances 实例列表  只会提供健康的实例  所以不需要担心如果实例无法访问的情况
     */
    private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> instances) {
        if (instances.isEmpty()) {
            return new EmptyResponse();
        }
        // 获取当前服务所在的集群名称
        String currentServiceName = nacosDiscoveryProperties.getService();
        System.out.println("currentServiceName:"+currentServiceName);
        // 过滤在同一集群下注册的服务 根据集群名称筛选的集合
        List<ServiceInstance> sameClusterNameInstList  = instances.stream().filter(i-> StringUtils.equals(i.getMetadata().get("nacos.service"),currentServiceName)).collect(Collectors.toList());
        ServiceInstance sameClusterNameInst;
        if (sameClusterNameInstList.isEmpty()) {
            // 如果为空，则根据权重直接过滤所有服务列表
            sameClusterNameInst = getHostByRandomWeight(instances);
        } else {
            // 如果不为空，则根据权重直接过滤所在集群下的服务列表
            sameClusterNameInst = getHostByRandomWeight(sameClusterNameInstList);
        }
        return new DefaultResponse(sameClusterNameInst);
    }

    private ServiceInstance getHostByRandomWeight(List<ServiceInstance> sameClusterNameInstList){

        List<Instance> list = new ArrayList<>();
        Map<String,ServiceInstance> dataMap = new HashMap<>();
        // 此处将 ServiceInstance 转化为 Instance 是为了接下来调用nacos中的权重算法，由于入参不同，所以需要转换，此处建议打断电进行参数调试，以下是我目前为止所用到的参数，转化为map是为了最终方便获取取值到的服务对象
        sameClusterNameInstList.forEach(i->{
            Instance ins = new Instance();
            Map<String, String> metadata = i.getMetadata();

            ins.setInstanceId(metadata.get("nacos.instanceId"));
            ins.setWeight(new BigDecimal(metadata.get("nacos.weight")).doubleValue());
            ins.setClusterName(metadata.get("nacos.cluster"));
            ins.setEphemeral(Boolean.parseBoolean(metadata.get("nacos.ephemeral")));
            ins.setHealthy(Boolean.parseBoolean(metadata.get("nacos.healthy")));
            ins.setPort(i.getPort());
            ins.setIp(i.getHost());
            ins.setServiceName(i.getServiceId());

            ins.setMetadata(metadata);

            list.add(ins);
            // key为服务ID，值为服务对象
            dataMap.put(metadata.get("nacos.instanceId"),i);
        });
        // 调用nacos官方提供的负载均衡权重算法
        System.out.println("main list:{  }" + list);
        Instance hostByRandomWeightCopy = ExtendBalancer.getHostByRandomWeightCopy(list);
        
        // 根据最终ID获取需要返回的实例对象
        return dataMap.get(hostByRandomWeightCopy.getInstanceId());
    }

}

class ExtendBalancer extends Balancer {
    /**
     * 根据自定义规则选择
     */
    public static Instance getHostByRandomWeightCopy(List<Instance> hosts) {

        for (Instance host : hosts) {
            if (host.getServiceName().equals("wms-outbound")) {
                System.out.println("main host:" + host);
                if (host.getPort() == 8906) {
                    continue;
                }
                return host;
            }
        }
        return getHostByRandomWeight(hosts);
    }

}

你知道的越多，你不知道的越多。