通过 Ribbon 查询 Nacos 服务实例

本文转载自微信公众号「运维开发故事」，作者老郑。转载本文请联系运维开发故事公众号。

创新互联建站服务项目包括岳普湖网站建设、岳普湖网站制作、岳普湖网页制作以及岳普湖网络营销策划等。多年来，我们专注于互联网行业，利用自身积累的技术优势、行业经验、深度合作伙伴关系等，向广大中小型企业、政府机构等提供互联网行业的解决方案，岳普湖网站推广取得了明显的社会效益与经济效益。目前，我们服务的客户以成都为中心已经辐射到岳普湖省份的部分城市，未来相信会继续扩大服务区域并继续获得客户的支持与信任！

Nacos 服务列表管理

Nacos 提供了开放 API 可通过 /nacos/v1/ns/instance/list 获取服务列表。如果我们采用 spring-cloud 方式去获取服务，最终会通过 Nacos Client + loadbalancer 的方式进行客户端负载均衡。

Ribbon 源码解析

Ribbon 简介

Spring Cloud Ribbon 是 Netflix Ribbon 实现的一套客户端负载均衡工具 简单的说，Ribbon 是 Netflix 发布的开源项目，主要功能是提供客户端的复杂算法和服务调用。 Ribbon 客户端组件提供一系列完善的配置项如超时、重试等。简单的说，就是配置文件中列出 load Balancer (简称 LB)后面所有的机器，Ribbon 会自动的帮助你基于某种规则(如简单轮询，随机链接等)去链接这些机器。我们很容易使用 Ribbon 自定义的负载均衡算法。

Ribbon 使用

首先需要定义 RestTemplate 使用 Ribbon 策略;

 
 
 
 
      
  
  
  
  1. @Configuration 
    2.   
  
  
  
  2. public class RestTemplateConfig { 
    3.   
  
  
  
  3.     @LoadBalanced 
    4.   
  
  
  
  4.     @Bean 
    5.   
  
  
  
  5.     public RestTemplate restTemplate() { 
    6.   
  
  
  
  6.         return new RestTemplate(); 
    7.   
  
  
  
  7.     } 
    8.   
  
  
  
  8. } 
    9.

本地使用 RestTemplate 调用远程接口;

 
 
 
 
      
  
  
  
  1. @Autowired 
    2.   
  
  
  
  2. private RestTemplate restTemplate; 
    3.   
  
  
  
  3. @RequestMapping(value = "/echo/{id}", method = RequestMethod.GET) 
    4.   
  
  
  
  4. public String echo(@PathVariable Long id) { 
    5.   
  
  
  
  5.     return restTemplate.getForObject("http://member-service/member/get/" + id, String.class); 
    6.   
  
  
  
  6. } 
    7.

Ribbon 源码分析

RestTemplate 继承 InterceptingHttpAccessor 通过 interceptors 字段接受 HttpRequestInterceptor 请求拦截器。对于 Ribbion 初始化类是 RibbonAutoConfiguration 中的， 它在 spring-cloud-netflix-ribbon 中定义。但是它在初始化之前，又需要加载 RibbonAutoConfiguration 配置，它是在 spring-cloud-common 中。具体的代码如下:

 
 
 
 
      
  
  
  
  1. @Configuration(proxyBeanMethods = false) 
    2.   
  
  
  
  2. // 工程中一定存在 RestTemplate 类 
    3.   
  
  
  
  3. @ConditionalOnClass(RestTemplate.class) 
    4.   
  
  
  
  4. // 容器中一定存在 LoadBalancerClient 类 Bean 实例 
    5.   
  
  
  
  5. @ConditionalOnBean(LoadBalancerClient.class) 
    6.   
  
  
  
  6. @EnableConfigurationProperties(LoadBalancerRetryProperties.class) 
    7.   
  
  
  
  7. public class LoadBalancerAutoConfiguration { 
    8.   
  
  
  
  8.  
    9.   
  
  
  
  9.    // 获取 Spring 容器中所有的 RestTemplate 实例 
    10.   
  
  
  
  10.    @LoadBalanced 
    11.   
  
  
  
  11.    @Autowired(required = false) 
    12.   
  
  
  
  12.    private List restTemplates = Collections.emptyList(); 
    13.   
  
  
  
  13.  
    14.   
  
  
  
  14.    // 获取 Spring 容器中 LoadBalancerRequestTransformer 实例 
    15.   
  
  
  
  15.    @Autowired(required = false) 
    16.   
  
  
  
  16.    private List transformers = Collections.emptyList(); 
    17.   
  
  
  
  17.  
    18.   
  
  
  
  18.    // 在 Bean 初始化完成后会调用 afterSingletonsInstantiated 方法 
    19.   
  
  
  
  19.    // 这里是一个 lambda 表达式方式的实现， 主要是为 restTemplate 实例设置 RestTemplateCustomizer 
    20.   
  
  
  
  20.    @Bean 
    21.   
  
  
  
  21.    public SmartInitializingSingleton loadBalancedRestTemplateInitializerDeprecated( 
    22.   
  
  
  
  22.          final ObjectProvider> restTemplateCustomizers) { 
    23.   
  
  
  
  23.       return () -> restTemplateCustomizers.ifAvailable(customizers -> { 
    24.   
  
  
  
  24.          for (RestTemplate restTemplate : LoadBalancerAutoConfiguration.this.restTemplates) { 
    25.   
  
  
  
  25.             for (RestTemplateCustomizer customizer : customizers) { 
    26.   
  
  
  
  26.                customizer.customize(restTemplate); 
    27.   
  
  
  
  27.             } 
    28.   
  
  
  
  28.          } 
    29.   
  
  
  
  29.       }); 
    30.   
  
  
  
  30.    } 
    31.   
  
  
  
  31.  
    32.   
  
  
  
  32.    // LoadBalancerRequestFactory 工厂类 
    33.   
  
  
  
  33.    // 主要是用来提供 LoadBalancerClient 实例和 LoadBalancerRequestTransformer 
    34.   
  
  
  
  34.    @Bean 
    35.   
  
  
  
  35.    @ConditionalOnMissingBean 
    36.   
  
  
  
  36.    public LoadBalancerRequestFactory loadBalancerRequestFactory( 
    37.   
  
  
  
  37.          LoadBalancerClient loadBalancerClient) { 
    38.   
  
  
  
  38.       return new LoadBalancerRequestFactory(loadBalancerClient, this.transformers); 
    39.   
  
  
  
  39.    } 
    40.   
  
  
  
  40.  
    41.   
  
  
  
  41.    // LoadBalancerInterceptor 拦截器  
    42.   
  
  
  
  42.    @Configuration(proxyBeanMethods = false) 
    43.   
  
  
  
  43.    @ConditionalOnMissingClass("org.springframework.retry.support.RetryTemplate") 
    44.   
  
  
  
  44.    static class LoadBalancerInterceptorConfig { 
    45.   
  
  
  
  45.  
    46.   
  
  
  
  46.       // 创建默认的拦截器 LoadBalancerInterceptor 的实例 
    47.   
  
  
  
  47.       @Bean 
    48.   
  
  
  
  48.       public LoadBalancerInterceptor ribbonInterceptor( 
    49.   
  
  
  
  49.             LoadBalancerClient loadBalancerClient, 
    50.   
  
  
  
  50.             LoadBalancerRequestFactory requestFactory) { 
    51.   
  
  
  
  51.          return new LoadBalancerInterceptor(loadBalancerClient, requestFactory); 
    52.   
  
  
  
  52.       } 
    53.   
  
  
  
  53.  
    54.   
  
  
  
  54.       // 如果没有 RestTemplateCustomizer 实例才会创建 
    55.   
  
  
  
  55.       // 这里就就会为咱们所有的  restTemplate 实例添加 loadBalancerInterceptor 拦截器 
    56.   
  
  
  
  56.       @Bean 
    57.   
  
  
  
  57.       @ConditionalOnMissingBean 
    58.   
  
  
  
  58.       public RestTemplateCustomizer restTemplateCustomizer( 
    59.   
  
  
  
  59.             final LoadBalancerInterceptor loadBalancerInterceptor) { 
    60.   
  
  
  
  60.          return restTemplate -> { 
    61.   
  
  
  
  61.             List list = new ArrayList<>( 
    62.   
  
  
  
  62.                   restTemplate.getInterceptors()); 
    63.   
  
  
  
  63.             list.add(loadBalancerInterceptor); 
    64.   
  
  
  
  64.             restTemplate.setInterceptors(list); 
    65.   
  
  
  
  65.          }; 
    66.   
  
  
  
  66.       } 
    67.   
  
  
  
  67.  
    68.   
  
  
  
  68.    } 
    69.   
  
  
  
  69.   // ... 
    70.   
  
  
  
  70.  
    71.   
  
  
  
  71. } 
    72.

针对下面的代码我们可以总结一下：

如果需要使用负载均衡，工程下面必须要有 RestTemplate 类， 然后Spring 容器中要有 LoadBalancerClient 的实例。

LoadBalancerClient 在 spring-cloud-netflix-ribbon 中只有一个实现类: RibbonLoadBalancerClient

利用 Spring 的 SmartInitializingSingleton 拓展点，在 restTemplateCustomizer() 中为所有的 RestTemplate 添加 LoadBalancerInterceptor 拦截器

其实 LoadBalancer 的本质就是通过拦截器。利用 RestTemplate 的拓展点来实现请求服务的负载均衡。

LoadBalancerInterceptor

LoadBalancerInterceptor 拦截器会将请求交给 LoadBalancerClient 去处理，首先会选择一个 ILoadBalancer 的实现来处理获取和选择服务，然后通过 serviceName 和负载均衡算法去选择 Server 对象。最后执行请求。

 
 
 
 
      
  
  
  
  1. public class LoadBalancerInterceptor implements ClientHttpRequestInterceptor { 
    2.   
  
  
  
  2.  
    3.   
  
  
  
  3.    // 负载均衡 
    4.   
  
  
  
  4.    private LoadBalancerClient loadBalancer; 
    5.   
  
  
  
  5.  
    6.   
  
  
  
  6.    // 构建请求  
    7.   
  
  
  
  7.    private LoadBalancerRequestFactory requestFactory; 
    8.   
  
  
  
  8.  
    9.   
  
  
  
  9.    // ... 
    10.   
  
  
  
  10.  
    11.   
  
  
  
  11.    @Override 
    12.   
  
  
  
  12.    public ClientHttpResponse intercept(final HttpRequest request, final byte[] body, 
    13.   
  
  
  
  13.          final ClientHttpRequestExecution execution) throws IOException { 
    14.   
  
  
  
  14.       final URI originalUri = request.getURI(); 
    15.   
  
  
  
  15.       String serviceName = originalUri.getHost();  
    16.   
  
  
  
  16.       return this.loadBalancer.execute(serviceName, 
    17.   
  
  
  
  17.             this.requestFactory.createRequest(request, body, execution)); 
    18.   
  
  
  
  18.    } 
    19.   
  
  
  
  19.  
    20.   
  
  
  
  20. } 
    21.

RibbonLoadBalancerClient

我们通过跟踪 this.loadBalancer.execute 代码发现。最终所有的请求都交由 RibbonLoadBalancerClient 去处理。它实现了。LoadBalancerClient 接口， 代码如下：

 
 
 
 
      
  
  
  
  1. public interface ServiceInstanceChooser { 
    2.   
  
  
  
  2.  
    3.   
  
  
  
  3.   // 通过 serviceId 选择具体的服务实例 
    4.   
  
  
  
  4.   ServiceInstance choose(String serviceId); 
    5.   
  
  
  
  5.  
    6.   
  
  
  
  6. } 
    7.   
  
  
  
  7.  
    8.   
  
  
  
  8. public interface LoadBalancerClient extends ServiceInstanceChooser { 
    9.   
  
  
  
  9.  
    10.   
  
  
  
  10.    T execute(String serviceId, LoadBalancerRequest request) throws IOException; 
    11.   
  
  
  
  11.    T execute(String serviceId, ServiceInstance serviceInstance, 
    12.   
  
  
  
  12.          LoadBalancerRequest request) throws IOException; 
    13.   
  
  
  
  13.   // 将服务实例信息替换还具体的 IP 信息  
    14.   
  
  
  
  14.   URI reconstructURI(ServiceInstance instance, URI original); 
    15.   
  
  
  
  15.  
    16.   
  
  
  
  16. } 
    17.

我们先来分析 RibbonLoadBalancerClient 的 choose 方法

 
 
 
 
      
  
  
  
  1. @Override 
    2.   
  
  
  
  2. public ServiceInstance choose(String serviceId) { 
    3.   
  
  
  
  3.    return choose(serviceId, null); 
    4.   
  
  
  
  4. } 
    5.   
  
  
  
  5.  
    6.   
  
  
  
  6. // 通过服务名选择具体的服务实例 
    7.   
  
  
  
  7. public ServiceInstance choose(String serviceId, Object hint) { 
    8.   
  
  
  
  8.    Server server = getServer(getLoadBalancer(serviceId), hint); 
    9.   
  
  
  
  9.    if (server == null) { 
    10.   
  
  
  
  10.       return null; 
    11.   
  
  
  
  11.    } 
    12.   
  
  
  
  12.    return new RibbonServer(serviceId, server, isSecure(server, serviceId), 
    13.   
  
  
  
  13.          serverIntrospector(serviceId).getMetadata(server)); 
    14.   
  
  
  
  14. } 
    15.   
  
  
  
  15.  
    16.   
  
  
  
  16. // 通过服务名选择一个负载均衡器, 默认是 `ZoneAwareLoadBalancer` 
    17.   
  
  
  
  17. protected ILoadBalancer getLoadBalancer(String serviceId) { 
    18.   
  
  
  
  18.    return this.clientFactory.getLoadBalancer(serviceId); 
    19.   
  
  
  
  19. } 
    20.   
  
  
  
  20.  
    21.   
  
  
  
  21. // 获取服务 
    22.   
  
  
  
  22. protected Server getServer(ILoadBalancer loadBalancer) { 
    23.   
  
  
  
  23.    return getServer(loadBalancer, null); 
    24.   
  
  
  
  24. } 
    25.   
  
  
  
  25. protected Server getServer(ILoadBalancer loadBalancer, Object hint) { 
    26.   
  
  
  
  26.    if (loadBalancer == null) { 
    27.   
  
  
  
  27.       return null; 
    28.   
  
  
  
  28.    } 
    29.   
  
  
  
  29.    // Use 'default' on a null hint, or just pass it on? 
    30.   
  
  
  
  30.    return loadBalancer.chooseServer(hint != null ? hint : "default"); 
    31.   
  
  
  
  31. } 
    32.

LoadBalancerInterceptor 执行的时候是直接委托执行的 loadBalancer.execute() 这个方法：

 
 
 
 
      
  
  
  
  1. // LoadBalancerRequest 是通过 LoadBalancerRequestFactory.createRequest(request, body, execution) 创建 
    2.   
  
  
  
  2. // 它实现 LoadBalancerRequest 接口是用的一个匿名内部类，泛型类型是ClientHttpResponse 
    3.   
  
  
  
  3. // 因为最终执行的显然还是执行器：ClientHttpRequestExecution.execute() 
    4.   
  
  
  
  4. @Override 
    5.   
  
  
  
  5. public  T execute(String serviceId, LoadBalancerRequest request) throws IOException { 
    6.   
  
  
  
  6.   return execute(serviceId, request, null); 
    7.   
  
  
  
  7. } 
    8.   
  
  
  
  8.  
    9.   
  
  
  
  9. public  T execute(String serviceId, LoadBalancerRequest request, Object hint) throws IOException { 
    10.   
  
  
  
  10.   // 拿到负载均衡器，然后拿到一个serverInstance实例 
    11.   
  
  
  
  11.   ILoadBalancer loadBalancer = getLoadBalancer(serviceId); 
    12.   
  
  
  
  12.   Server server = getServer(loadBalancer, hint); 
    13.   
  
  
  
  13.   if (server == null) { // 若没找到就直接抛出异常。这里使用的是IllegalStateException这个异常 
    14.   
  
  
  
  14.     throw new IllegalStateException("No instances available for " + serviceId); 
    15.   
  
  
  
  15.   } 
    16.   
  
  
  
  16.  
    17.   
  
  
  
  17.   // 把Server适配为RibbonServer  isSecure：客户端是否安全 
    18.   
  
  
  
  18.   // serverIntrospector内省  参考配置文件：ServerIntrospectorProperties 
    19.   
  
  
  
  19.   RibbonServer ribbonServer = new RibbonServer(serviceId, server, 
    20.   
  
  
  
  20.       isSecure(server, serviceId), serverIntrospector(serviceId).getMetadata(server)); 
    21.   
  
  
  
  21.  
    22.   
  
  
  
  22.   //调用本类的重载接口方法 
    23.   
  
  
  
  23.   return execute(serviceId, ribbonServer, request); 
    24.   
  
  
  
  24. } 
    25.   
  
  
  
  25.  
    26.   
  
  
  
  26. // 它的参数是 ServiceInstance --> 已经确定了唯一的Server实例 
    27.   
  
  
  
  27. @Override 
    28.   
  
  
  
  28. public  T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest request) throws IOException { 
    29.   
  
  
  
  29.  
    30.   
  
  
  
  30.   // 拿到 Server，RibbonServer 是 execute 时的唯一实现 
    31.   
  
  
  
  31.   Server server = null; 
    32.   
  
  
  
  32.   if (serviceInstance instanceof RibbonServer) { 
    33.   
  
  
  
  33.     server = ((RibbonServer) serviceInstance).getServer(); 
    34.   
  
  
  
  34.   } 
    35.   
  
  
  
  35.   if (server == null) { 
    36.   
  
  
  
  36.     throw new IllegalStateException("No instances available for " + serviceId); 
    37.   
  
  
  
  37.   } 
    38.   
  
  
  
  38.  
    39.   
  
  
  
  39.   // 执行的上下文是和serviceId绑定的 
    40.   
  
  
  
  40.   RibbonLoadBalancerContext context = this.clientFactory.getLoadBalancerContext(serviceId); 
    41.   
  
  
  
  41.   ...  
    42.   
  
  
  
  42.   // 真正的向server发送请求，得到返回值 
    43.   
  
  
  
  43.   // 因为有拦截器，所以这里肯定说执行的是InterceptingRequestExecution.execute()方法 
    44.   
  
  
  
  44.   // so会调用ServiceRequestWrapper.getURI()，从而就会调用reconstructURI()方法 
    45.   
  
  
  
  45.     T returnVal = request.apply(serviceInstance); 
    46.   
  
  
  
  46.     return returnVal; 
    47.   
  
  
  
  47.   ... // 异常处理 
    48.   
  
  
  
  48. } 
    49.

returnVal 是一个 ClientHttpResponse，最后交给 handleResponse()方法来处理异常情况(若存在的话)，若无异常就交给提取器提值：responseExtractor.extractData(response)，这样整个请求就算全部完成了。

ZoneAwareLoadBalancer

负载均衡器 ZoneAwareLoadBalancer 的类图结构如下图所示。它 DynamicServerListLoadBalancer 它的父类， 核心方法 重置和初始化：restOfInit(clientConfig) 更新服务列表：updateListOfServers(); 这个方需要调用到 ServerList.getUpdatedListOfServers() 这里就会调用到具体的注册中心实现，以 Nacos 为例他的实现就是 NacosServerList#getUpdatedListOfServers();

  • 更新所有服务列表：updateAllServerList();
  • 设置所有服务列表 setServersList() ZoneAwareLoadBalancer 它的核心方法：
  • 选择服务实例 chooseServer()
  • 选择负载均衡器 getLoadBalancer
  • 选择区域内的服务实例：zoneLoadBalancer.chooseServer

Ribbon 总结

针对 @LoadBalanced 下的 RestTemplate 的使用，我总结如下：

  • 传入的String类型的url必须是绝对路径(http://...)，否则抛出异常：java.lang.IllegalArgumentException: URI is not absolute
  • serviceId 不区分大小写(http://order-service/...效果同http://OERDER-SERVICE/...)
  • serviceId 后请不要跟 port 端口号

最后，需要特别指出的是：标注有@LoadBalanced 的 RestTemplate 只能填写 serviceId 而不能再写 IP地址/域名去发送请求了， 若你的项目中两种 case 都有需要，需要定义多个 RestTemplate 分别应对不同的使用场景

Nacos 服务查询

客户端查询

如果我们使用默认的 Nacos 客户端，那么走的就是 NacosServerList#getUpdatedListOfServers();接口来查询服务列表。

 
 
 
 
      
  
  
  
  1. public class NacosServerList extends AbstractServerList { 
    2.   
  
  
  
  2.  
    3.   
  
  
  
  3.   private NacosDiscoveryProperties discoveryProperties; 
    4.   
  
  
  
  4.  
    5.   
  
  
  
  5.   @Override 
    6.   
  
  
  
  6.   public List getUpdatedListOfServers() { 
    7.   
  
  
  
  7.     return getServers(); 
    8.   
  
  
  
  8.   } 
    9.   
  
  
  
  9.      
    10.   
  
  
  
  10.     private List getServers() { 
    11.   
  
  
  
  11.     try { 
    12.   
  
  
  
  12.       String group = discoveryProperties.getGroup(); 
    13.   
  
  
  
  13.       // discoveryProperties.namingServiceInstance()  
    14.   
  
  
  
  14.             // 最终通过反射获取 com.alibaba.nacos.client.naming.NacosNamingService 实例 
    15.   
  
  
  
  15.             List instances = discoveryProperties.namingServiceInstance() 
    16.   
  
  
  
  16.           .selectInstances(serviceId, group, true); 
    17.   
  
  
  
  17.       return instancesToServerList(instances); 
    18.   
  
  
  
  18.     } 
    19.   
  
  
  
  19.     catch (Exception e) { 
    20.   
  
  
  
  20.       throw new IllegalStateException( 
    21.   
  
  
  
  21.           "Can not get service instances from nacos, serviceId=" + serviceId, 
    22.   
  
  
  
  22.           e); 
    23.   
  
  
  
  23.     } 
    24.   
  
  
  
  24.   } 
    25.   
  
  
  
  25. } 
    26.

然后调用 selectInstances 方法

 
 
 
 
      
  
  
  
  1. @Override 
    2.   
  
  
  
  2. public List selectInstances(String serviceName, String groupName, List clusters, boolean healthy, 
    3.   
  
  
  
  3.                                       boolean subscribe) throws NacosException { 
    4.   
  
  
  
  4.  
    5.   
  
  
  
  5.     ServiceInfo serviceInfo; 
    6.   
  
  
  
  6.     // subscribe 默认传的是 true 
    7.   
  
  
  
  7.     if (subscribe) { 
    8.   
  
  
  
  8.         serviceInfo = hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName), 
    9.   
  
  
  
  9.                                                  StringUtils.join(clusters, ",")); 
    10.   
  
  
  
  10.     } else { 
    11.   
  
  
  
  11.         serviceInfo = hostReactor 
    12.   
  
  
  
  12.             .getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName), 
    13.   
  
  
  
  13.                                               StringUtils.join(clusters, ",")); 
    14.   
  
  
  
  14.     } 
    15.   
  
  
  
  15.     return selectInstances(serviceInfo, healthy); 
    16.   
  
  
  
  16. } 
    17.

其实核心的逻辑在 hostReactor.getServiceInfo 在查询服务信息里面会把当前的 serviceName、 clusters 转换为 key， 然后通过 getServiceInfo0 方法查询服务信息这里主要是查询的是本地的数据。

如果 null == serviceObj 会在 updateServiceNow 里面去调用 /instance/list接口查询服务信息

 
 
 
 
      
  
  
  
  1. public ServiceInfo getServiceInfo(final String serviceName, final String clusters) { 
    2.   
  
  
  
  2.          
    3.   
  
  
  
  3.         NAMING_LOGGER.debug("failover-mode: " + failoverReactor.isFailoverSwitch()); 
    4.   
  
  
  
  4.         String key = ServiceInfo.getKey(serviceName, clusters); 
    5.   
  
  
  
  5.         if (failoverReactor.isFailoverSwitch()) { 
    6.   
  
  
  
  6.             return failoverReactor.getService(key); 
    7.   
  
  
  
  7.         } 
    8.   
  
  
  
  8.          
    9.   
  
  
  
  9.         ServiceInfo serviceObj = getServiceInfo0(serviceName, clusters); 
    10.   
  
  
  
  10.          
    11.   
  
  
  
  11.         if (null == serviceObj) { 
    12.   
  
  
  
  12.             serviceObj = new ServiceInfo(serviceName, clusters); 
    13.   
  
  
  
  13.              
    14.   
  
  
  
  14.             serviceInfoMap.put(serviceObj.getKey(), serviceObj); 
    15.   
  
  
  
  15.              
    16.   
  
  
  
  16.             updatingMap.put(serviceName, new Object()); 
    17.   
  
  
  
  17.             updateServiceNow(serviceName, clusters); 
    18.   
  
  
  
  18.             updatingMap.remove(serviceName); 
    19.   
  
  
  
  19.              
    20.   
  
  
  
  20.         } else if (updatingMap.containsKey(serviceName)) { 
    21.   
  
  
  
  21.             // UPDATE_HOLD_INTERVAL 为常量默认金辉进去 
    22.   
  
  
  
  22.             if (UPDATE_HOLD_INTERVAL > 0) { 
    23.   
  
  
  
  23.                 // hold a moment waiting for update finish 
    24.   
  
  
  
  24.                 synchronized (serviceObj) { 
    25.   
  
  
  
  25.                     try { 
    26.   
  
  
  
  26.                         // 最大等待时间 5s, 在更新 serviceObj 之后, 就会执行 notifyAll() 
    27.   
  
  
  
  27.                         // 方法入口 updateService(String serviceName, String clusters) 
    28.   
  
  
  
  28.                         // 最大延迟 2s DEFAULT_DELAY = 1 
    29.   
  
  
  
  29.                         serviceObj.wait(UPDATE_HOLD_INTERVAL); 
    30.   
  
  
  
  30.                     } catch (InterruptedException e) { 
    31.   
  
  
  
  31.                         NAMING_LOGGER 
    32.   
  
  
  
  32.                                 .error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, e); 
    33.   
  
  
  
  33.                     } 
    34.   
  
  
  
  34.                 } 
    35.   
  
  
  
  35.             } 
    36.   
  
  
  
  36.         } 
    37.   
  
  
  
  37.          
    38.   
  
  
  
  38.       // 通过 Schedule 更新 服务信息  
    39.   
  
  
  
  39.         scheduleUpdateIfAbsent(serviceName, clusters); 
    40.   
  
  
  
  40.          
    41.   
  
  
  
  41.       // 获取最新的值 
    42.   
  
  
  
  42.         return serviceInfoMap.get(serviceObj.getKey()); 
    43.   
  
  
  
  43.     } 
    44.

代码看到这里我们不难理解，为什么第一次 Ribbon 调用的时候都会比较慢，因为它回去初始化服务列表，然后通过 Nacos Client 去 Nacos 查询服务实例信息。

服务端处理

服务端通过 /instance/list 接口来处理服务实例信息查询请求。首先服务实例信息都是被存储在 ConcurrentHashMap 中

 
 
 
 
      
  
  
  
  1. /** 
    2.   
  
  
  
  2.  * Map(namespace, Map(group::serviceName, Service)). 
    3.   
  
  
  
  3.  */ 
    4.   
  
  
  
  4. private final Map> serviceMap = new ConcurrentHashMap<>(); 
    5.

在我们查询的过程中主要是通过 ServiceManager 来进行管理， 核心的入口方法在 InstanceController#doSrvIpxt 中

 
 
 
 
      
  
  
  
  1. public ObjectNode doSrvIpxt(String namespaceId, String serviceName, String agent, String clusters, String clientIP, 
    2.   
  
  
  
  2.           int udpPort, String env, boolean isCheck, String app, String tid, boolean healthyOnly) throws Exception { 
    3.   
  
  
  
  3.        
    4.   
  
  
  
  4.       ClientInfo clientInfo = new ClientInfo(agent); 
    5.   
  
  
  
  5.       ObjectNode result = JacksonUtils.createEmptyJsonNode(); 
    6.   
  
  
  
  6.       Service service = serviceManager.getService(namespaceId, serviceName); 
    7.   
  
  
  
  7.       long cacheMillis = switchDomain.getDefaultCacheMillis(); 
    8.   
  
  
  
  8.        
    9.   
  
  
  
  9.       // now try to enable the push 
    10.   
  
  
  
  10.       try { 
    11.   
  
  
  
  11.           if (udpPort > 0 && pushService.canEnablePush(agent)) { 
    12.   
  
  
  
  12.                
    13.   
  
  
  
  13.               pushService 
    14.   
  
  
  
  14.                       .addClient(namespaceId, serviceName, clusters, agent, new InetSocketAddress(clientIP, udpPort), 
    15.   
  
  
  
  15.                               pushDataSource, tid, app); 
    16.   
  
  
  
  16.               cacheMillis = switchDomain.getPushCacheMillis(serviceName); 
    17.   
  
  
  
  17.           } 
    18.   
  
  
  
  18.       } catch (Exception e) { 
    19.   
  
  
  
  19.           Loggers.SRV_LOG 
    20.   
  
  
  
  20.                   .error("[NACOS-API] failed to added push client {}, {}:{}", clientInfo, clientIP, udpPort, e); 
    21.   
  
  
  
  21.           cacheMillis = switchDomain.getDefaultCacheMillis(); 
    22.   
  
  
  
  22.       } 
    23.   
  
  
  
  23.        
    24.   
  
  
  
  24.       if (service == null) { 
    25.   
  
  
  
  25.           if (Loggers.SRV_LOG.isDebugEnabled()) { 
    26.   
  
  
  
  26.               Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName); 
    27.   
  
  
  
  27.           } 
    28.   
  
  
  
  28.           result.put("name", serviceName); 
    29.   
  
  
  
  29.           result.put("clusters", clusters); 
    30.   
  
  
  
  30.           result.put("cacheMillis", cacheMillis); 
    31.   
  
  
  
  31.           result.replace("hosts", JacksonUtils.createEmptyArrayNode()); 
    32.   
  
  
  
  32.           return result; 
    33.   
  
  
  
  33.       } 
    34.   
  
  
  
  34.        
    35.   
  
  
  
  35.       checkIfDisabled(service); 
    36.   
  
  
  
  36.        
    37.   
  
  
  
  37.       List srvedIPs; 
    38.   
  
  
  
  38.        
    39.   
  
  
  
  39.       // 查询所有的服务 
    40.   
  
  
  
  40.       // 内部会更新服务列表 
    41.   
  
  
  
  41.       // allInstances.addAll(persistentInstances); 
    42.   
  
  
  
  42.       // allInstances.addAll(ephemeralInstances); 
    43.   
  
  
  
  43.       srvedIPs = service.srvIPs(Arrays.asList(StringUtils.split(clusters, ","))); 
    44.   
  
  
  
  44.        
    45.   
  
  
  
  45.       // filter ips using selector: 
    46.   
  
  
  
  46.       if (service.getSelector() != null && StringUtils.isNotBlank(clientIP)) { 
    47.   
  
  
  
  47.           srvedIPs = service.getSelector().select(clientIP, srvedIPs); 
    48.   
  
  
  
  48.       } 
    49.   
  
  
  
  49.        
    50.   
  
  
  
  50.       if (CollectionUtils.isEmpty(srvedIPs)) { 
    51.   
  
  
  
  51.            
    52.   
  
  
  
  52.           if (Loggers.SRV_LOG.isDebugEnabled()) { 
    53.   
  
  
  
  53.               Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName); 
    54.   
  
  
  
  54.           } 
    55.   
  
  
  
  55.            
    56.   
  
  
  
  56.           if (clientInfo.type == ClientInfo.ClientType.JAVA 
    57.   
  
  
  
  57.                   && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) { 
    58.   
  
  
  
  58.               result.put("dom", serviceName); 
    59.   
  
  
  
  59.           } else { 
    60.   
  
  
  
  60.               result.put("dom", NamingUtils.getServiceName(serviceName)); 
    61.   
  
  
  
  61.           } 
    62.   
  
  
  
  62.            
    63.   
  
  
  
  63.           result.put("name", serviceName); 
    64.   
  
  
  
  64.           result.put("cacheMillis", cacheMillis); 
    65.   
  
  
  
  65.           result.put("lastRefTime", System.currentTimeMillis()); 
    66.   
  
  
  
  66.           result.put("checksum", service.getChecksum()); 
    67.   
  
  
  
  67.           result.put("useSpecifiedURL", false); 
    68.   
  
  
  
  68.           result.put("clusters", clusters); 
    69.   
  
  
  
  69.           result.put("env", env); 
    70.   
  
  
  
  70.           result.set("hosts", JacksonUtils.createEmptyArrayNode()); 
    71.   
  
  
  
  71.           result.set("metadata", JacksonUtils.transferToJsonNode(service.getMetadata())); 
    72.   
  
  
  
  72.           return result; 
    73.   
  
  
  
  73.       } 
    74.   
  
  
  
  74.        
    75.   
  
  
  
  75.       Map> ipMap = new HashMap<>(2); 
    76.   
  
  
  
  76.       ipMap.put(Boolean.TRUE, new ArrayList<>()); 
    77.   
  
  
  
  77.       ipMap.put(Boolean.FALSE, new ArrayList<>()); 
    78.   
  
  
  
  78.        
    79.   
  
  
  
  79.       for (Instance ip : srvedIPs) { 
    80.   
  
  
  
  80.           ipMap.get(ip.isHealthy()).add(ip); 
    81.   
  
  
  
  81.       } 
    82.   
  
  
  
  82.        
    83.   
  
  
  
  83.       if (isCheck) { 
    84.   
  
  
  
  84.           result.put("reachProtectThreshold", false); 
    85.   
  
  
  
  85.       } 
    86.   
  
  
  
  86.        
    87.   
  
  
  
  87.       double threshold = service.getProtectThreshold(); 
    88.   
  
  
  
  88.        
    89.   
  
  
  
  89.       if ((float) ipMap.get(Boolean.TRUE).size() / srvedIPs.size() <= threshold) { 
    90.   
  
  
  
  90.            
    91.   
  
  
  
  91.           Loggers.SRV_LOG.warn("protect threshold reached, return all ips, service: {}", serviceName); 
    92.   
  
  
  
  92.           if (isCheck) { 
    93.   
  
  
  
  93.               result.put("reachProtectThreshold", true); 
    94.   
  
  
  
  94.           } 
    95.   
  
  
  
  95.            
    96.   
  
  
  
  96.           ipMap.get(Boolean.TRUE).addAll(ipMap.get(Boolean.FALSE)); 
    97.   
  
  
  
  97.           ipMap.get(Boolean.FALSE).clear(); 
    98.   
  
  
  
  98.       } 
    99.   
  
  
  
  99.        
    100.   
  
  
  
  100.       if (isCheck) { 
    101.   
  
  
  
  101.           result.put("protectThreshold", service.getProtectThreshold()); 
    102.   
  
  
  
  102.           result.put("reachLocalSiteCallThreshold", false); 
    103.   
  
  
  
  103.            
    104.   
  
  
  
  104.           return JacksonUtils.createEmptyJsonNode(); 
    105.   
  
  
  
  105.       } 
    106.   
  
  
  
  106.        
    107.   
  
  
  
  107.       ArrayNode hosts = JacksonUtils.createEmptyArrayNode(); 
    108.   
  
  
  
  108.        
    109.   
  
  
  
  109.       for (Map.Entry> entry : ipMap.entrySet()) { 
    110.   
  
  
  
  110.           List ips = entry.getValue(); 
    111.   
  
  
  
  111.            
    112.   
  
  
  
  112.           if (healthyOnly && !entry.getKey()) { 
    113.   
  
  
  
  113.               continue; 
    114.   
  
  
  
  114.           } 
    115.   
  
  
  
  115.            
    116.   
  
  
  
  116.           for (Instance instance : ips) { 
    117.   
  
  
  
  117.                
    118.   
  
  
  
  118.               // remove disabled instance: 
    119.   
  
  
  
  119.               if (!instance.isEnabled()) { 
    120.   
  
  
  
  120.                   continue; 
    121.   
  
  
  
  121.               } 
    122.   
  
  
  
  122.                
    123.   
  
  
  
  123.               ObjectNode ipObj = JacksonUtils.createEmptyJsonNode(); 
    124.   
  
  
  
  124.                
    125.   
  
  
  
  125.               ipObj.put("ip", instance.getIp()); 
    126.   
  
  
  
  126.               ipObj.put("port", instance.getPort()); 
    127.   
  
  
  
  127.               // deprecated since nacos 1.0.0: 
    128.   
  
  
  
  128.               ipObj.put("valid", entry.getKey()); 
    129.   
  
  
  
  129.               ipObj.put("healthy", entry.getKey()); 
    130.   
  
  
  
  130.               ipObj.put("marked", instance.isMarked()); 
    131.   
  
  
  
  131.               ipObj.put("instanceId", instance.getInstanceId()); 
    132.   
  
  
  
  132.               ipObj.set("metadata", JacksonUtils.transferToJsonNode(instance.getMetadata())); 
    133.   
  
  
  
  133.               ipObj.put("enabled", instance.isEnabled()); 
    134.   
  
  
  
  134.               ipObj.put("weight", instance.getWeight()); 
    135.   
  
  
  
  135.               ipObj.put("clusterName", instance.getClusterName()); 
    136.   
  
  
  
  136.               if (clientInfo.type == ClientInfo.ClientType.JAVA 
    137.   
  
  
  
  137.                       && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) { 
    138.   
  
  
  
  138.                   ipObj.put("serviceName", instance.getServiceName()); 
    139.   
  
  
  
  139.               } else { 
    140.   
  
  
  
  140.                   ipObj.put("serviceName", NamingUtils.getServiceName(instance.getServiceName())); 
    141.   
  
  
  
  141.               } 
    142.   
  
  
  
  142.                
    143.   
  
  
  
  143.               ipObj.put("ephemeral", instance.isEphemeral()); 
    144.   
  
  
  
  144.               hosts.add(ipObj); 
    145.   
  
  
  
  145.                
    146.   
  
  
  
  146.           } 
    147.   
  
  
  
  147.       } 
    148.   
  
  
  
  148.        
    149.   
  
  
  
  149.       result.replace("hosts", hosts); 
    150.   
  
  
  
  150.       if (clientInfo.type == ClientInfo.ClientType.JAVA 
    151.   
  
  
  
  151.               && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) { 
    152.   
  
  
  
  152.           result.put("dom", serviceName); 
    153.   
  
  
  
  153.       } else { 
    154.   
  
  
  
  154.          &n            
    
                新闻标题：通过 Ribbon 查询 Nacos 服务实例
                
    
                文章分享：http://www.shufengxianlan.com/qtweb/news26/98376.html
            
    
            
    
                网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等
                                
    
            
    
                广告
                
            
    
            
    
                声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源：
                创新互联

猜你还喜欢下面的内容

用户体验知识

同城分类信息

成都恒翔机电维修　　 　深圳惠宜防蚊纱窗　　 　成都网站建设　　 　成都网站建设　　 　成都封阳台　　 　企业网站建设公司　　 　力比特机械　　 　四川成都网站设计　　 　域名申请　　 　网站运维托管　　 　重庆网站制作　　 　成都网站推广　　 　企业网站建设　　 　响应式网站建设　　 　定制网站建设　　 　成都集团网站建设　　 　成都网站设计　　 　四川国泰民安消防　　 　温江网站建设　　 　SSL证书