前言

RestTemplate是spring的http客户端，底层使用HttpURLConnection，不过也支持使用apache httpclient、okhttp、netty等方式

loadbalancer和ribbon介绍

  1. loadbalancer：
    1. spring cloud commons下面有个loadbalance包，是spring cloud提供的负载均衡通用包，这个通用基础包，定义了许多spring cloud的负载均衡接口，客户端只需要实现这些接口，就可以整合到spring cloud中，符合面向对象七大设计原则
    2. spring cloud loadbalancer是spring cloud实现这个上面通用包而产生的负载均衡包，所以这里需要区别一下，有一个loadbalance是spring cloud commons通用包，定义了一系列通用接口，有一个loadbalance是spring cloud专门实现的负载均衡包。并且loadbalance支持spring后面提出的WebClient，ribbon太早不再维护而不支持WebClient
  2. ribbon：ribbon是Netflix实现spring cloud commons下面loadbalancer包的负载均衡组件，和spring cloud loadbalancer属于同一类的产品，只不过ribbon后面不再维护，在spring cloud后续版本中，只支持spring cloud loadbalancer

RestTemplate整合Ribbon

依赖

1
2
3
4
5
6
<!--spring-cloud-starter-netflix-ribbon已停止维护，最新版本为2.2.10.RELEASE（2021年11月），并且在spring cloud 2020.0.0之后不再引入（包括2020.0.0），故需要单独定义版本-->
<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-starter-netflix-ribbon</artifactId>
  <version>2.2.10.RELEASE</version>
</dependency>

这里需要注意，如果项目引入的是spring-cloud-dependencies：2020.0.0及以后的版本，将需要单独定义ribbon版本，并且可能因为不适配而存在一些问题，所以并不推荐使用，推荐使用spring cloud官方提供的spring-cloud-starter-loadbalancer，这里只做演示

代码

如果需要在RestTemplate请求过程中加入ribbon负载均衡，只需要在注册RestTemplate的时候加上@LoadBalanced即可

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/**
     * springboot注入RestTemplate
     * @return
     */
@Bean
@LoadBalanced
public RestTemplate restTemplate(){
    //配置使用apache httpclient作为RestTemplate的Http客户端
    RestTemplate restTemplate = new RestTemplate(new HttpComponentsClientHttpRequestFactory());
    //设置自定义拦截器
    restTemplate.setInterceptors(
        Collections.singletonList(
            new RestTemplateInterceptor()
        )
    );
    return restTemplate;
}

如何生效

@LoadBalanced生效的方法是通过为RestTemplate添加一个拦截器，拦截器会将RestTemplate提交的请求中的服务名通过负载均衡替换为实际的地址，例如http://user:6060/会被替换成http://192.168.0.1:6060/，这个拦截器是org.springframework.cloud.client.loadbalancer.RetryLoadBalancerInterceptor，RetryLoadBalancerInterceptor实现了ClientHttpRequestInterceptor，但是ClientHttpRequestInterceptor不只有这一个实现类，只是这一个实现类整合了spring重试的机制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
public ClientHttpResponse intercept(final HttpRequest request, final byte[] body, final ClientHttpRequestExecution execution) throws IOException {
    URI originalUri = request.getURI();
    String serviceName = originalUri.getHost();
    Assert.state(serviceName != null, "Request URI does not contain a valid hostname: " + originalUri);
    LoadBalancedRetryPolicy retryPolicy = this.lbRetryFactory.createRetryPolicy(serviceName, this.loadBalancer);
    RetryTemplate template = this.createRetryTemplate(serviceName, request, retryPolicy);
    return (ClientHttpResponse)template.execute((context) -> {
        ServiceInstance serviceInstance = null;
        if (context instanceof LoadBalancedRetryContext) {
            LoadBalancedRetryContext lbContextxx = (LoadBalancedRetryContext)context;
            serviceInstance = lbContextxx.getServiceInstance();
            if (LOG.isDebugEnabled()) {
                LOG.debug(String.format("Retrieved service instance from LoadBalancedRetryContext: %s", serviceInstance));
            }
        }

        Set<LoadBalancerLifecycle> supportedLifecycleProcessors = LoadBalancerLifecycleValidator.getSupportedLifecycleProcessors(this.loadBalancerFactory.getInstances(serviceName, LoadBalancerLifecycle.class), RetryableRequestContext.class, ResponseData.class, ServiceInstance.class);
        String hint = this.getHint(serviceName);
        if (serviceInstance == null) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Service instance retrieved from LoadBalancedRetryContext: was null. Reattempting service instance selection");
            }

            ServiceInstance previousServiceInstance = null;
            if (context instanceof LoadBalancedRetryContext) {
                LoadBalancedRetryContext lbContext = (LoadBalancedRetryContext)context;
                previousServiceInstance = lbContext.getPreviousServiceInstance();
            }

            DefaultRequest<RetryableRequestContext> lbRequestx = new DefaultRequest(new RetryableRequestContext(previousServiceInstance, new RequestData(request), hint));
            supportedLifecycleProcessors.forEach((lifecycle) -> {
                lifecycle.onStart(lbRequestx);
            });
            serviceInstance = this.loadBalancer.choose(serviceName, lbRequestx);
            if (LOG.isDebugEnabled()) {
                LOG.debug(String.format("Selected service instance: %s", serviceInstance));
            }

            if (context instanceof LoadBalancedRetryContext) {
                LoadBalancedRetryContext lbContextx = (LoadBalancedRetryContext)context;
                lbContextx.setServiceInstance(serviceInstance);
            }

            Response<ServiceInstance> lbResponse = new DefaultResponse(serviceInstance);
            if (serviceInstance == null) {
                supportedLifecycleProcessors.forEach((lifecycle) -> {
                    lifecycle.onComplete(new CompletionContext(Status.DISCARD, new DefaultRequest(new RetryableRequestContext((ServiceInstance)null, new RequestData(request), hint)), lbResponse));
                });
            }
        }

        LoadBalancerRequestAdapter<ClientHttpResponse, RetryableRequestContext> lbRequest = new LoadBalancerRequestAdapter(this.requestFactory.createRequest(request, body, execution), new RetryableRequestContext((ServiceInstance)null, new RequestData(request), hint));
        ClientHttpResponse response = (ClientHttpResponse)this.loadBalancer.execute(serviceName, serviceInstance, lbRequest);
        int statusCode = response.getRawStatusCode();
        if (retryPolicy != null && retryPolicy.retryableStatusCode(statusCode)) {
            if (LOG.isDebugEnabled()) {
                LOG.debug(String.format("Retrying on status code: %d", statusCode));
            }

            byte[] bodyCopy = StreamUtils.copyToByteArray(response.getBody());
            response.close();
            throw new ClientHttpResponseStatusCodeException(serviceName, response, bodyCopy);
        } else {
            return response;
        }
    }, new LoadBalancedRecoveryCallback<ClientHttpResponse, ClientHttpResponse>() {
        protected ClientHttpResponse createResponse(ClientHttpResponse response, URI uri) {
            return response;
        }
    });
}

RetryLoadBalancerInterceptor的loadBalancer对象就是负载均衡客户端，LoadBalancerClient属于spring cloud基础包的内容

1
2
3
4
5
6
7
8
9
public interface LoadBalancerClient extends ServiceInstanceChooser {
    <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException;
    <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException;
    URI reconstructURI(ServiceInstance instance, URI original);
}
public interface ServiceInstanceChooser {
    ServiceInstance choose(String serviceId);
    <T> ServiceInstance choose(String serviceId, Request<T> request);
}

ribbon和loadbanlancer分别实现了这个接口，serviceInstance = this.loadBalancer.choose(serviceName, lbRequestx);，方法中的这一行代码就是负载均衡获取服务实例，但是ribbon因为没有维护的原因，实现的是之前的一个参数choose方法，和现在两个参数的方法有冲突，所以我们需要自己实现一版

1
2
3
4
5
6
7
8
9
10
public class MyRibbonLoadBalancerClient extends RibbonLoadBalancerClient {
    public MyRibbonLoadBalancerClient(SpringClientFactory clientFactory) {
        super(clientFactory);
    }

    @Override
    public <T> ServiceInstance choose(String serviceId, Request<T> request) {
        return choose(serviceId);
    }
}

我们继承RibbonLoadBalancerClient并实现ServiceInstanceChooser的choose(String serviceId, Request request)方法，但是调用的还是RibbonLoadBalancerClient的choose(String serviceId)方法，从而去除版本不一致带来的影响，最后还要将我们继承的类注册到spring

1
2
3
4
5
@Bean
public LoadBalancerClient loadBalancerClient(){
    return new MyRibbonLoadBalancerClient(springClientFactory);
}

自定义负载均衡规则

1
2
3
protected Server getServer(ILoadBalancer loadBalancer, Object hint) {
    return loadBalancer == null ? null : loadBalancer.chooseServer(hint != null ? hint : "default");
}

从这个方法可以看出，具体的负载规则是由ILoadBalancer负责的，我们可以实现ILoadBalancer并且注册到spring中，但是ribbon对于ILoadBalancer有了默认的实现，BaseLoadBalancer，并且BaseLoadBalancer还有很多子类，实现了多种多样的功能，一般情况不通过实现ILoadBalancer来自定义负载均衡规则，BaseLoadBalancer默认实现了获取服务的通用方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public Server chooseServer(Object key) {
    if (counter == null) {
        counter = createCounter();
    }
    counter.increment();
    if (rule == null) {
        return null;
    } else {
        try {
            return rule.choose(key);
        } catch (Exception e) {
            logger.warn("LoadBalancer [{}]:  Error choosing server for key {}", name, key, e);
            return null;
        }
    }
}

其中我们需要查看的就是rule.choose(key)这行代码，这就是BaseLoadBalancer具体的负载均衡规则，也就是说BaseLoadBalancer对于具体负载均衡并不是强实现，而是给了扩展的空间，并且在此基础上，封装了许多功能，所以我们一般实现IRule（或者AbstractLoadBalancerRule（IRule的子类））接口即可，并且注册到spring中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@Bean
public IRule iRule() {
    return new NacosSameClusterWeightedRule();
}

public class NacosSameClusterWeightedRule extends AbstractLoadBalancerRule {

    @Autowired
    private NacosDiscoveryProperties nacosDiscoveryProperties;

    @Override
    public void initWithNiwsConfig(IClientConfig iClientConfig) {
    }

    @Override
    public Server choose(Object o) {
        String clusterName = nacosDiscoveryProperties.getClusterName();
        BaseLoadBalancer loadBalancer = (BaseLoadBalancer) this.getLoadBalancer();
        String name = loadBalancer.getName();
        NamingService namingService;
        try {
            namingService = NacosFactory.createNamingService(nacosDiscoveryProperties.getNacosProperties());
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }

        List<Instance> instances;
        try {
            instances = namingService.selectInstances(name, true);
        } catch (NacosException e) {
            System.out.println("集群负载均衡算法出现问题");
            e.printStackTrace();
            return null;
        }

        List<Instance> collect = instances.stream()
        .filter(instance -> Objects.equals(instance.getClusterName(), clusterName))
        .collect(Collectors.toList());

        if (CollUtil.isNotEmpty(collect)) {
            Instance instance = ExtendsBalancer.getHostByRandomWeight2(instances);
            return new Server(instance.getIp(), instance.getPort());
        } else {
            return null;
        }
    }
}

public class NacosVersionRule extends AbstractLoadBalancerRule {

    @Autowired
    private NacosDiscoveryProperties nacosDiscoveryProperties;

    @Override
    public void initWithNiwsConfig(IClientConfig iClientConfig) {
    }

    @Override
    public Server choose(Object o) {
        String clusterName = nacosDiscoveryProperties.getClusterName();
        String targetVersion = nacosDiscoveryProperties.getMetadata().get("target-version");
        NamingService namingService;
        try {
            namingService = NacosFactory.createNamingService(nacosDiscoveryProperties.getNacosProperties());
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }

        BaseLoadBalancer loadBalancer = (BaseLoadBalancer) getLoadBalancer();
        String name = loadBalancer.getName();

        List<Instance> instances;
        try {
            instances = namingService.selectInstances(name, true);
        } catch (NacosException e) {
            log.error("通过服务名获取实例失败，可能不存在该服务");
            return null;
        }

        if (StringUtils.isNotBlank(targetVersion)) {
            instances = instances.stream()
            .filter(instance -> Objects.equals(targetVersion, instance.getMetadata().get("version")))
            .collect(Collectors.toList());
            if (CollectionUtils.isEmpty(instances)) {
                log.error("未找到符合servicename={}且version={}的实例", name, targetVersion);
                return null;
            }
        }

        if (StringUtils.isNotBlank(clusterName)) {
            List<Instance> collect = instances.stream()
                    .filter(instance -> Objects.equals(clusterName, instance.getClusterName()))
                    .collect(Collectors.toList());
            if (CollUtil.isNotEmpty(collect)) {
                Instance instance = ExtendsBalancer.getHostByRandomWeight2(instances);
                return new Server(instance.getIp(), instance.getPort());
            } else {
                return null;
            }
        }else {
            return null;
        }
    }
}

/**
 * 根据权重的负载均衡算法
 */
public class NacosWeightedRule extends AbstractLoadBalancerRule {

    @Autowired
    private NacosDiscoveryProperties nacosDiscoveryProperties;

    @Override
    public void initWithNiwsConfig(IClientConfig iClientConfig) {
    }

    @Override
    public Server choose(Object o) {
        //Ribbon的入口
        BaseLoadBalancer loadBalancer = (BaseLoadBalancer) this.getLoadBalancer();
        //指定服务的名称
        String name = loadBalancer.getName();
        NamingService namingService;
        try {
            namingService = NacosFactory.createNamingService(nacosDiscoveryProperties.getNacosProperties());
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }
        Instance instance;
        try {
            //根据基于权重的业务方法得到一个实例
            instance = namingService.selectOneHealthyInstance(name);
        } catch (NacosException e) {
            System.out.println("基于权重的负载均衡算法出现问题！");
            return null;
        }
        return new Server(instance.getIp(), instance.getPort());
    }
}

static class ExtendsBalancer extends Balancer {
    static Instance getHostByRandomWeight2(List<Instance> instanceList){
        return getHostByRandomWeight(instanceList);
    }
}

上面实现了三种不同的负载均衡算法，有只返回同clusterName的服务，只返回同target-version的服务，最为主要的方法就是com.alibaba.nacos.client.naming.core.Balancer#getHostByRandomWeight，它是nacos实现的一个针对nacos服务的负载均衡算法，由于这个方法是protected（只允许子类调用），所以我们实现一个它的子类，并且调用这个方法，让我们的程序可以简介调用getHostByRandomWeight，namingService.selectOneHealthyInstance(name)的底层也是调用的它

Ribbon获取服务

Ribbon获取服务比较重要的几个类

  1. com.netflix.loadbalancer.ServerList：ServerList是获取服务列表的接口，在不同注册中心下有不同的实现，例如2020之前版本的spring-cloud-starter-alibaba-nacos-discovery中就有实现com.alibaba.cloud.nacos.ribbon.NacosServerList
  2. com.netflix.loadbalancer.ServerListFilter：ServerListFilter接口根据配置的规则对服务实例进行过滤
  3. com.netflix.loadbalancer.BaseLoadBalancer：BaseLoadBalancer的allServerList为上面IRule获取服务过程中默认获取的服务列表，初始会通过ServerList的getInitialListOfServers方法获取初始服务列表
  4. com.netflix.loadbalancer.DynamicServerListLoadBalancer#updateListOfServers：updateListOfServers方法通过调用ServerList的getUpdatedListOfServers方法获取服务列表并通过调用ServerListFilter的getFilteredListOfServers方法过滤服务，并且通过调用父类BaseLoadBalancer的setServersList方法，更新BaseLoadBalancer的allServerList
1
2
3
4
5
6
7
8
9
10
11
12
public interface ServerList<T extends Server> {

    public List<T> getInitialListOfServers();

    /**
     * Return updated list of servers. This is called say every 30 secs
     * (configurable) by the Loadbalancer's Ping cycle
     * 
     */
    public List<T> getUpdatedListOfServers();   

}
1
2
3
4
5
public interface ServerListFilter<T extends Server> {

    public List<T> getFilteredListOfServers(List<T> servers);

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public void updateListOfServers() {
    List<T> servers = new ArrayList<T>();
    if (serverListImpl != null) {
        servers = serverListImpl.getUpdatedListOfServers();
        LOGGER.debug("List of Servers for {} obtained from Discovery client: {}",
                     getIdentifier(), servers);

        if (filter != null) {
            servers = filter.getFilteredListOfServers(servers);
            LOGGER.debug("Filtered List of Servers for {} obtained from Discovery client: {}",
                         getIdentifier(), servers);
        }
    }
    updateAllServerList(servers);
}

上述获取服务的流程，是在IRule负载过程中获取服务的默认方式，如果我们自定义了负载均衡规则，那么获取服务的操作就在我们自己实现的负载均衡规则代码中实现。在2020之后版本的spring-cloud-starter-alibaba-nacos-discovery已经不提供ServerList的实现类了，所以在nacos作为注册中心的情况下，只能需要自定义负载均衡规则获取nacos服务，或者实现ServerList

RestTemplate整合LoadBalancer

依赖

1
2
3
4
5
<!--基本所有spring-cloud-dependencies版本都会引入spring-cloud-starter-loadbalancer，所以一般不需要单独定义版本--->
<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-starter-loadbalancer</artifactId>
</dependency>

代码

代码和RestTemplate整合Ribbon的一致，也是通过@LoadBalanced注解

如何生效

生效过程和ribbion一致，也是通过RetryLoadBalancerInterceptor类，和ribbon唯一的区别就是对于LoadBalancerClient的实现，相对于ribbon实现的RibbonLoadBalancerClient，loadBalancer实现了BlockingLoadBalancerClient，如果项目引入了loadbalancer没有引入ribbon，那么默认会使用BlockingLoadBalancerClient，但是如果引入了ribbon，会默认使用RibbonLoadBalancerClient，所以不可避免的引入ribbon的情况下，需要手动注册BlockingLoadBalancerClient

1
2
3
4
5
6
7
@Autowired
private LoadBalancerClientFactory loadBalancerClientFactory;

@Bean
public BlockingLoadBalancerClient blockingLoadBalancerClient(){
    return new BlockingLoadBalancerClient(loadBalancerClientFactory);
}

自定义负载均衡规则

1
2
3
4
5
6
7
8
9
public <T> ServiceInstance choose(String serviceId, Request<T> request) {
    ReactiveLoadBalancer<ServiceInstance> loadBalancer = this.loadBalancerClientFactory.getInstance(serviceId);
    if (loadBalancer == null) {
        return null;
    } else {
        Response<ServiceInstance> loadBalancerResponse = (Response)Mono.from(loadBalancer.choose(request)).block();
        return loadBalancerResponse == null ? null : (ServiceInstance)loadBalancerResponse.getServer();
    }
}

这是BlockingLoadBalancerClient实现的choose方法，可以看出具体负载规则是由ReactiveLoadBalancer类控制的，所以只需要实现ReactiveLoadBalancer并且配置即可

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@ConditionalOnDiscoveryEnabled
@LoadBalancerClient(name = "LoadBalancerClientConfiguration", configuration = LoadBalancerClientConfiguration.class)
@LoadBalancerClients(defaultConfiguration = LoadBalancerClientConfiguration.class)
public class LoadBalancerClientConfiguration {

    @Bean
    public ReactorLoadBalancer<ServiceInstance> reactorServiceInstanceLoadBalancer(Environment environment, LoadBalancerClientFactory loadBalancerClientFactory) {
        String name = environment.getProperty("loadbalancer.client.name");
        return new NacosSameClusterWeightedRule(loadBalancerClientFactory.getLazyProvider(name, ServiceInstanceListSupplier.class), name);
    }

    public class NacosSameClusterWeightedRule implements ReactorServiceInstanceLoadBalancer {

        @Autowired
        private NacosDiscoveryProperties nacosDiscoveryProperties;

        final String serviceId;
        ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;

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

        @Override
        public Mono<Response<ServiceInstance>> choose(Request request) {
            try {
                String clusterName = nacosDiscoveryProperties.getClusterName();

                NamingService namingService = NacosFactory.createNamingService(nacosDiscoveryProperties.getNacosProperties());
                List<Instance> instances = namingService.getAllInstances(serviceId).stream().filter(instance -> instance.getClusterName().equals(clusterName)).collect(Collectors.toList());

                Instance instance = ExtendsBalancer.getHostByRandomWeight2(instances);

                return Mono.create(responseMonoSink -> responseMonoSink.success(new DefaultResponse(toNacosServiceInstance(instance))));
            } catch (NacosException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public class NacosVersionRule implements ReactorServiceInstanceLoadBalancer {

        @Autowired
        private NacosDiscoveryProperties nacosDiscoveryProperties;

        final String serviceId;
        ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;

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

        @Override
        public Mono<Response<ServiceInstance>> choose(Request request) {
            try {
                String targetVersion = nacosDiscoveryProperties.getMetadata().get("target-version");

                NamingService namingService = NacosFactory.createNamingService(nacosDiscoveryProperties.getNacosProperties());
                List<Instance> instances = namingService.getAllInstances(serviceId).stream().filter(instance -> instance.getMetadata().get("target-version").equals(targetVersion)).collect(Collectors.toList());
                Instance instance = ExtendsBalancer.getHostByRandomWeight2(instances);

                return Mono.create(responseMonoSink -> responseMonoSink.success(new DefaultResponse(toNacosServiceInstance(instance))));
            } catch (NacosException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public class NacosWeightedRule implements ReactorServiceInstanceLoadBalancer {

        @Autowired
        private NacosDiscoveryProperties nacosDiscoveryProperties;

        final String serviceId;
        ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;

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

        @Override
        public Mono<Response<ServiceInstance>> choose(Request request) {
            try{
                NamingService namingService = NacosFactory.createNamingService(nacosDiscoveryProperties.getNacosProperties());
                //根据基于权重的业务方法得到一个实例
                Instance instance = namingService.selectOneHealthyInstance(serviceId);
                return Mono.create(responseMonoSink -> responseMonoSink.success(new DefaultResponse(toNacosServiceInstance(instance))));
            }catch (Exception e){
                throw new RuntimeException(e);
            }
        }
    }

    public static NacosServiceInstance toNacosServiceInstance(Instance instance) {
        NacosServiceInstance nacosServiceInstance = new NacosServiceInstance();
        nacosServiceInstance.setHost(instance.getIp());
        nacosServiceInstance.setMetadata(instance.getMetadata());
        nacosServiceInstance.setServiceId(instance.getServiceName());
        nacosServiceInstance.setPort(instance.getPort());

        return nacosServiceInstance;
    }

    static class ExtendsBalancer extends Balancer {
        static Instance getHostByRandomWeight2(List<Instance> instanceList){
            return getHostByRandomWeight(instanceList);
        }
    }

}

上述代码和上面ribbon实现的负载均衡一致，只是实现方式略有不同。loadBalancer的负载均衡不仅需要注册到spring，还需要使用到@LoadBalancerClient和@LoadBalancerClients，并且需要两个同时配置才会生效

LoadBalancer获取服务

LoadBalancer获取服务过程中比较重要的几个类

  1. org.springframework.cloud.client.discovery.DiscoveryClient：DiscoveryClient是spring-cloud-common下面的包，不同注册中心下有不同的实现，spring-cloud-loadbalancer在负载过程中通过调用这个接口获取服务列表
  2. com.alibaba.cloud.nacos.discovery.NacosDiscoveryClient：NacosDiscoveryClient是nacos对于DiscoveryClient的实现，底层是通过NamingService获取服务
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public interface DiscoveryClient extends Ordered {
    int DEFAULT_ORDER = 0;

    String description();

    List<ServiceInstance> getInstances(String serviceId);

    List<String> getServices();

    default void probe() {
        this.getServices();
    }

    default int getOrder() {
        return 0;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public class NacosDiscoveryClient implements DiscoveryClient {
    private static final Logger log = LoggerFactory.getLogger(NacosDiscoveryClient.class);
    public static final String DESCRIPTION = "Spring Cloud Nacos Discovery Client";
    private NacosServiceDiscovery serviceDiscovery;

    public NacosDiscoveryClient(NacosServiceDiscovery nacosServiceDiscovery) {
        this.serviceDiscovery = nacosServiceDiscovery;
    }

    public String description() {
        return "Spring Cloud Nacos Discovery Client";
    }

    public List<ServiceInstance> getInstances(String serviceId) {
        try {
            return this.serviceDiscovery.getInstances(serviceId);
        } catch (Exception var3) {
            throw new RuntimeException("Can not get hosts from nacos server. serviceId: " + serviceId, var3);
        }
    }

    public List<String> getServices() {
        try {
            return this.serviceDiscovery.getServices();
        } catch (Exception var2) {
            log.error("get service name from nacos server fail,", var2);
            return Collections.emptyList();
        }
    }
}

上述获取服务的流程，是在ReactiveLoadBalancer负载过程中获取服务的默认方式，如果我们自定义了负载均衡规则，那么获取服务的操作就在我们自己实现的负载均衡规则代码中实现

本站总访问量：次本站总访问数：人
本站由 Peng Tong 使用 Stellar 1.29.1 主题创建。
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议，转载请注明出处。