本文轉載自微信公眾號「運維開發故事」，作者老鄭。轉載本文請聯系運維開發故事公眾號。

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 策略;

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

本地使用 RestTemplate 調用遠程接口;

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

Ribbon 源碼分析

RestTemplate 繼承 InterceptingHttpAccessor 通過 interceptors 字段接受 HttpRequestInterceptor 請求攔截器。對于 Ribbion 初始化類是 RibbonAutoConfiguration 中的， 它在 spring-cloud-netflix-ribbon 中定義。但是它在初始化之前，又需要加載 RibbonAutoConfiguration 配置，它是在 spring-cloud-common 中。具體的代碼如下:

@Configuration(proxyBeanMethods = false) 
// 工程中一定存在 RestTemplate 類 
@ConditionalOnClass(RestTemplate.class) 
// 容器中一定存在 LoadBalancerClient 類 Bean 實例 
@ConditionalOnBean(LoadBalancerClient.class) 
@EnableConfigurationProperties(LoadBalancerRetryProperties.class) 
public class LoadBalancerAutoConfiguration { 
 
   // 獲取 Spring 容器中所有的 RestTemplate 實例 
   @LoadBalanced 
   @Autowired(required = false) 
   private List<RestTemplate> restTemplates = Collections.emptyList(); 
 
   // 獲取 Spring 容器中 LoadBalancerRequestTransformer 實例 
   @Autowired(required = false) 
   private List<LoadBalancerRequestTransformer> transformers = Collections.emptyList(); 
 
   // 在 Bean 初始化完成后會調用 afterSingletonsInstantiated 方法 
   // 這里是一個 lambda 表達式方式的實現， 主要是為 restTemplate 實例設置 RestTemplateCustomizer 
   @Bean 
   public SmartInitializingSingleton loadBalancedRestTemplateInitializerDeprecated( 
         final ObjectProvider<List<RestTemplateCustomizer>> restTemplateCustomizers) { 
      return () -> restTemplateCustomizers.ifAvailable(customizers -> { 
         for (RestTemplate restTemplate : LoadBalancerAutoConfiguration.this.restTemplates) { 
            for (RestTemplateCustomizer customizer : customizers) { 
               customizer.customize(restTemplate); 
            } 
         } 
      }); 
   } 
 
   // LoadBalancerRequestFactory 工廠類 
   // 主要是用來提供 LoadBalancerClient 實例和 LoadBalancerRequestTransformer 
   @Bean 
   @ConditionalOnMissingBean 
   public LoadBalancerRequestFactory loadBalancerRequestFactory( 
         LoadBalancerClient loadBalancerClient) { 
      return new LoadBalancerRequestFactory(loadBalancerClient, this.transformers); 
   } 
 
   // LoadBalancerInterceptor 攔截器  
   @Configuration(proxyBeanMethods = false) 
   @ConditionalOnMissingClass("org.springframework.retry.support.RetryTemplate") 
   static class LoadBalancerInterceptorConfig { 
 
      // 創建默認的攔截器 LoadBalancerInterceptor 的實例 
      @Bean 
      public LoadBalancerInterceptor ribbonInterceptor( 
            LoadBalancerClient loadBalancerClient, 
            LoadBalancerRequestFactory requestFactory) { 
         return new LoadBalancerInterceptor(loadBalancerClient, requestFactory); 
      } 
 
      // 如果沒有 RestTemplateCustomizer 實例才會創建 
      // 這里就就會為咱們所有的  restTemplate 實例添加 loadBalancerInterceptor 攔截器 
      @Bean 
      @ConditionalOnMissingBean 
      public RestTemplateCustomizer restTemplateCustomizer( 
            final LoadBalancerInterceptor loadBalancerInterceptor) { 
         return restTemplate -> { 
            List<ClientHttpRequestInterceptor> list = new ArrayList<>( 
                  restTemplate.getInterceptors()); 
            list.add(loadBalancerInterceptor); 
            restTemplate.setInterceptors(list); 
         }; 
      } 
 
   } 
  // ... 
 
} 

針對下面的代碼我們可以總結一下：

如果需要使用負載均衡，工程下面必須要有 RestTemplate 類， 然后Spring 容器中要有 LoadBalancerClient 的實例。

LoadBalancerClient 在 spring-cloud-netflix-ribbon 中只有一個實現類: RibbonLoadBalancerClient

利用 Spring 的 SmartInitializingSingleton 拓展點，在 restTemplateCustomizer() 中為所有的 RestTemplate 添加 LoadBalancerInterceptor 攔截器

其實 LoadBalancer 的本質就是通過攔截器。利用 RestTemplate 的拓展點來實現請求服務的負載均衡。

LoadBalancerInterceptor

LoadBalancerInterceptor 攔截器會將請求交給 LoadBalancerClient 去處理，首先會選擇一個 ILoadBalancer 的實現來處理獲取和選擇服務，然后通過 serviceName 和負載均衡算法去選擇 Server 對象。最后執行請求。

public class LoadBalancerInterceptor implements ClientHttpRequestInterceptor { 
 
   // 負載均衡 
   private LoadBalancerClient loadBalancer; 
 
   // 構建請求  
   private LoadBalancerRequestFactory requestFactory; 
 
   // ... 
 
   @Override 
   public ClientHttpResponse intercept(final HttpRequest request, final byte[] body, 
         final ClientHttpRequestExecution execution) throws IOException { 
      final URI originalUri = request.getURI(); 
      String serviceName = originalUri.getHost();  
      return this.loadBalancer.execute(serviceName, 
            this.requestFactory.createRequest(request, body, execution)); 
   } 
 
} 

RibbonLoadBalancerClient

我們通過跟蹤 this.loadBalancer.execute 代碼發現。最終所有的請求都交由 RibbonLoadBalancerClient 去處理。它實現了。LoadBalancerClient 接口， 代碼如下：

public interface ServiceInstanceChooser { 
 
  // 通過 serviceId 選擇具體的服務實例 
  ServiceInstance choose(String serviceId); 
 
} 
 
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; 
  // 將服務實例信息替換還具體的 IP 信息  
  URI reconstructURI(ServiceInstance instance, URI original); 
 
} 

我們先來分析 RibbonLoadBalancerClient 的 choose 方法

@Override 
public ServiceInstance choose(String serviceId) { 
   return choose(serviceId, null); 
} 
 
// 通過服務名選擇具體的服務實例 
public ServiceInstance choose(String serviceId, Object hint) { 
   Server server = getServer(getLoadBalancer(serviceId), hint); 
   if (server == null) { 
      return null; 
   } 
   return new RibbonServer(serviceId, server, isSecure(server, serviceId), 
         serverIntrospector(serviceId).getMetadata(server)); 
} 
 
// 通過服務名選擇一個負載均衡器, 默認是 `ZoneAwareLoadBalancer` 
protected ILoadBalancer getLoadBalancer(String serviceId) { 
   return this.clientFactory.getLoadBalancer(serviceId); 
} 
 
// 獲取服務 
protected Server getServer(ILoadBalancer loadBalancer) { 
   return getServer(loadBalancer, null); 
} 
protected Server getServer(ILoadBalancer loadBalancer, Object hint) { 
   if (loadBalancer == null) { 
      return null; 
   } 
   // Use 'default' on a null hint, or just pass it on? 
   return loadBalancer.chooseServer(hint != null ? hint : "default"); 
} 

LoadBalancerInterceptor 執行的時候是直接委托執行的 loadBalancer.execute() 這個方法：

// LoadBalancerRequest 是通過 LoadBalancerRequestFactory.createRequest(request, body, execution) 創建 
// 它實現 LoadBalancerRequest 接口是用的一個匿名內部類，泛型類型是ClientHttpResponse 
// 因為最終執行的顯然還是執行器：ClientHttpRequestExecution.execute() 
@Override 
public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException { 
  return execute(serviceId, request, null); 
} 
 
public <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint) throws IOException { 
  // 拿到負載均衡器，然后拿到一個serverInstance實例 
  ILoadBalancer loadBalancer = getLoadBalancer(serviceId); 
  Server server = getServer(loadBalancer, hint); 
  if (server == null) { // 若沒找到就直接拋出異常。這里使用的是IllegalStateException這個異常 
    throw new IllegalStateException("No instances available for " + serviceId); 
  } 
 
  // 把Server適配為RibbonServer  isSecure：客戶端是否安全 
  // serverIntrospector內省  參考配置文件：ServerIntrospectorProperties 
  RibbonServer ribbonServer = new RibbonServer(serviceId, server, 
      isSecure(server, serviceId), serverIntrospector(serviceId).getMetadata(server)); 
 
  //調用本類的重載接口方法 
  return execute(serviceId, ribbonServer, request); 
} 
 
// 它的參數是 ServiceInstance --> 已經確定了唯一的Server實例 
@Override 
public <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException { 
 
  // 拿到 Server，RibbonServer 是 execute 時的唯一實現 
  Server server = null; 
  if (serviceInstance instanceof RibbonServer) { 
    server = ((RibbonServer) serviceInstance).getServer(); 
  } 
  if (server == null) { 
    throw new IllegalStateException("No instances available for " + serviceId); 
  } 
 
  // 執行的上下文是和serviceId綁定的 
  RibbonLoadBalancerContext context = this.clientFactory.getLoadBalancerContext(serviceId); 
  ...  
  // 真正的向server發送請求，得到返回值 
  // 因為有攔截器，所以這里肯定說執行的是InterceptingRequestExecution.execute()方法 
  // so會調用ServiceRequestWrapper.getURI()，從而就會調用reconstructURI()方法 
    T returnVal = request.apply(serviceInstance); 
    return returnVal; 
  ... // 異常處理 
} 

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();接口來查詢服務列表。

public class NacosServerList extends AbstractServerList<NacosServer> { 
 
  private NacosDiscoveryProperties discoveryProperties; 
 
  @Override 
  public List<NacosServer> getUpdatedListOfServers() { 
    return getServers(); 
  } 
     
    private List<NacosServer> getServers() { 
    try { 
      String group = discoveryProperties.getGroup(); 
      // discoveryProperties.namingServiceInstance()  
            // 最終通過反射獲取 com.alibaba.nacos.client.naming.NacosNamingService 實例 
            List<Instance> instances = discoveryProperties.namingServiceInstance() 
          .selectInstances(serviceId, group, true); 
      return instancesToServerList(instances); 
    } 
    catch (Exception e) { 
      throw new IllegalStateException( 
          "Can not get service instances from nacos, serviceId=" + serviceId, 
          e); 
    } 
  } 
} 

然后調用 selectInstances 方法

@Override 
public List<Instance> selectInstances(String serviceName, String groupName, List<String> clusters, boolean healthy, 
                                      boolean subscribe) throws NacosException { 
 
    ServiceInfo serviceInfo; 
    // subscribe 默認傳的是 true 
    if (subscribe) { 
        serviceInfo = hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName), 
                                                 StringUtils.join(clusters, ",")); 
    } else { 
        serviceInfo = hostReactor 
            .getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName), 
                                              StringUtils.join(clusters, ",")); 
    } 
    return selectInstances(serviceInfo, healthy); 
} 

其實核心的邏輯在 hostReactor.getServiceInfo 在查詢服務信息里面會把當前的 serviceName、 clusters 轉換為 key， 然后通過 getServiceInfo0 方法查詢服務信息這里主要是查詢的是本地的數據。

如果 null == serviceObj 會在 updateServiceNow 里面去調用 /instance/list接口查詢服務信息

public ServiceInfo getServiceInfo(final String serviceName, final String clusters) { 
         
        NAMING_LOGGER.debug("failover-mode: " + failoverReactor.isFailoverSwitch()); 
        String key = ServiceInfo.getKey(serviceName, clusters); 
        if (failoverReactor.isFailoverSwitch()) { 
            return failoverReactor.getService(key); 
        } 
         
        ServiceInfo serviceObj = getServiceInfo0(serviceName, clusters); 
         
        if (null == serviceObj) { 
            serviceObj = new ServiceInfo(serviceName, clusters); 
             
            serviceInfoMap.put(serviceObj.getKey(), serviceObj); 
             
            updatingMap.put(serviceName, new Object()); 
            updateServiceNow(serviceName, clusters); 
            updatingMap.remove(serviceName); 
             
        } else if (updatingMap.containsKey(serviceName)) { 
            // UPDATE_HOLD_INTERVAL 為常量默認金輝進去 
            if (UPDATE_HOLD_INTERVAL > 0) { 
                // hold a moment waiting for update finish 
                synchronized (serviceObj) { 
                    try { 
                        // 最大等待時間 5s, 在更新 serviceObj 之后, 就會執行 notifyAll() 
                        // 方法入口 updateService(String serviceName, String clusters) 
                        // 最大延遲 2s DEFAULT_DELAY = 1 
                        serviceObj.wait(UPDATE_HOLD_INTERVAL); 
                    } catch (InterruptedException e) { 
                        NAMING_LOGGER 
                                .error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, e); 
                    } 
                } 
            } 
        } 
         
      // 通過 Schedule 更新 服務信息  
        scheduleUpdateIfAbsent(serviceName, clusters); 
         
      // 獲取最新的值 
        return serviceInfoMap.get(serviceObj.getKey()); 
    } 

代碼看到這里我們不難理解，為什么第一次 Ribbon 調用的時候都會比較慢，因為它回去初始化服務列表，然后通過 Nacos Client 去 Nacos 查詢服務實例信息。

服務端處理

服務端通過 /instance/list 接口來處理服務實例信息查詢請求。首先服務實例信息都是被存儲在 ConcurrentHashMap 中

/** 
 * Map(namespace, Map(group::serviceName, Service)). 
 */ 
private final Map<String, Map<String, Service>> serviceMap = new ConcurrentHashMap<>(); 

在我們查詢的過程中主要是通過 ServiceManager 來進行管理， 核心的入口方法在 InstanceController#doSrvIpxt 中

public ObjectNode doSrvIpxt(String namespaceId, String serviceName, String agent, String clusters, String clientIP, 
          int udpPort, String env, boolean isCheck, String app, String tid, boolean healthyOnly) throws Exception { 
       
      ClientInfo clientInfo = new ClientInfo(agent); 
      ObjectNode result = JacksonUtils.createEmptyJsonNode(); 
      Service service = serviceManager.getService(namespaceId, serviceName); 
      long cacheMillis = switchDomain.getDefaultCacheMillis(); 
       
      // now try to enable the push 
      try { 
          if (udpPort > 0 && pushService.canEnablePush(agent)) { 
               
              pushService 
                      .addClient(namespaceId, serviceName, clusters, agent, new InetSocketAddress(clientIP, udpPort), 
                              pushDataSource, tid, app); 
              cacheMillis = switchDomain.getPushCacheMillis(serviceName); 
          } 
      } catch (Exception e) { 
          Loggers.SRV_LOG 
                  .error("[NACOS-API] failed to added push client {}, {}:{}", clientInfo, clientIP, udpPort, e); 
          cacheMillis = switchDomain.getDefaultCacheMillis(); 
      } 
       
      if (service == null) { 
          if (Loggers.SRV_LOG.isDebugEnabled()) { 
              Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName); 
          } 
          result.put("name", serviceName); 
          result.put("clusters", clusters); 
          result.put("cacheMillis", cacheMillis); 
          result.replace("hosts", JacksonUtils.createEmptyArrayNode()); 
          return result; 
      } 
       
      checkIfDisabled(service); 
       
      List<Instance> srvedIPs; 
       
      // 查詢所有的服務 
      // 內部會更新服務列表 
      // allInstances.addAll(persistentInstances); 
      // allInstances.addAll(ephemeralInstances); 
      srvedIPs = service.srvIPs(Arrays.asList(StringUtils.split(clusters, ","))); 
       
      // filter ips using selector: 
      if (service.getSelector() != null && StringUtils.isNotBlank(clientIP)) { 
          srvedIPs = service.getSelector().select(clientIP, srvedIPs); 
      } 
       
      if (CollectionUtils.isEmpty(srvedIPs)) { 
           
          if (Loggers.SRV_LOG.isDebugEnabled()) { 
              Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName); 
          } 
           
          if (clientInfo.type == ClientInfo.ClientType.JAVA 
                  && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) { 
              result.put("dom", serviceName); 
          } else { 
              result.put("dom", NamingUtils.getServiceName(serviceName)); 
          } 
           
          result.put("name", serviceName); 
          result.put("cacheMillis", cacheMillis); 
          result.put("lastRefTime", System.currentTimeMillis()); 
          result.put("checksum", service.getChecksum()); 
          result.put("useSpecifiedURL", false); 
          result.put("clusters", clusters); 
          result.put("env", env); 
          result.set("hosts", JacksonUtils.createEmptyArrayNode()); 
          result.set("metadata", JacksonUtils.transferToJsonNode(service.getMetadata())); 
          return result; 
      } 
       
      Map<Boolean, List<Instance>> ipMap = new HashMap<>(2); 
      ipMap.put(Boolean.TRUE, new ArrayList<>()); 
      ipMap.put(Boolean.FALSE, new ArrayList<>()); 
       
      for (Instance ip : srvedIPs) { 
          ipMap.get(ip.isHealthy()).add(ip); 
      } 
       
      if (isCheck) { 
          result.put("reachProtectThreshold", false); 
      } 
       
      double threshold = service.getProtectThreshold(); 
       
      if ((float) ipMap.get(Boolean.TRUE).size() / srvedIPs.size() <= threshold) { 
           
          Loggers.SRV_LOG.warn("protect threshold reached, return all ips, service: {}", serviceName); 
          if (isCheck) { 
              result.put("reachProtectThreshold", true); 
          } 
           
          ipMap.get(Boolean.TRUE).addAll(ipMap.get(Boolean.FALSE)); 
          ipMap.get(Boolean.FALSE).clear(); 
      } 
       
      if (isCheck) { 
          result.put("protectThreshold", service.getProtectThreshold()); 
          result.put("reachLocalSiteCallThreshold", false); 
           
          return JacksonUtils.createEmptyJsonNode(); 
      } 
       
      ArrayNode hosts = JacksonUtils.createEmptyArrayNode(); 
       
      for (Map.Entry<Boolean, List<Instance>> entry : ipMap.entrySet()) { 
          List<Instance> ips = entry.getValue(); 
           
          if (healthyOnly && !entry.getKey()) { 
              continue; 
          } 
           
          for (Instance instance : ips) { 
               
              // remove disabled instance: 
              if (!instance.isEnabled()) { 
                  continue; 
              } 
               
              ObjectNode ipObj = JacksonUtils.createEmptyJsonNode(); 
               
              ipObj.put("ip", instance.getIp()); 
              ipObj.put("port", instance.getPort()); 
              // deprecated since nacos 1.0.0: 
              ipObj.put("valid", entry.getKey()); 
              ipObj.put("healthy", entry.getKey()); 
              ipObj.put("marked", instance.isMarked()); 
              ipObj.put("instanceId", instance.getInstanceId()); 
              ipObj.set("metadata", JacksonUtils.transferToJsonNode(instance.getMetadata())); 
              ipObj.put("enabled", instance.isEnabled()); 
              ipObj.put("weight", instance.getWeight()); 
              ipObj.put("clusterName", instance.getClusterName()); 
              if (clientInfo.type == ClientInfo.ClientType.JAVA 
                      && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) { 
                  ipObj.put("serviceName", instance.getServiceName()); 
              } else { 
                  ipObj.put("serviceName", NamingUtils.getServiceName(instance.getServiceName())); 
              } 
               
              ipObj.put("ephemeral", instance.isEphemeral()); 
              hosts.add(ipObj); 
               
          } 
      } 
       
      result.replace("hosts", hosts); 
      if (clientInfo.type == ClientInfo.ClientType.JAVA 
              && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) { 
          result.put("dom", serviceName); 
      } else { 
          result.put("dom", NamingUtils.getServiceName(serviceName)); 
      } 
      result.put("name", serviceName); 
      result.put("cacheMillis", cacheMillis); 
      result.put("lastRefTime", System.currentTimeMillis()); 
      result.put("checksum", service.getChecksum()); 
      result.put("useSpecifiedURL", false); 
      result.put("clusters", clusters); 
      result.put("env", env); 
      result.replace("metadata", JacksonUtils.transferToJsonNode(service.getMetadata())); 
      return result; 
  } 

在上面的核心邏輯主要是：

• 調用 service.srvIPs 方法查詢所有的服務實例信息
• Cluster#allIPs會將所有的服務注冊信息寫到服務注冊列表。

參考鏈接

https://nacos.io

https://zhuanlan.zhihu.com

https://blog.csdn.net/f641385712/article/details/100788040