詳解SpringCloud LoadBalancer 新一代負(fù)載均衡器

更新時(shí)間：2023年01月16日 09:54:03 作者：暮色妖嬈丶

這篇文章主要為大家介紹了SpringCloud LoadBalancer新一代負(fù)載均衡器詳解使用詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前言

工作中使用 OpenFeign 進(jìn)行跨服務(wù)調(diào)用，最近發(fā)現(xiàn)線上經(jīng)常會(huì)遇到請(qǐng)求失敗。

java.net.ConnectException: Connection refused: connect

通過(guò)排查我們發(fā)現(xiàn)不是接口超時(shí)，而是有時(shí)候會(huì)請(qǐng)求到已經(jīng)下線的服務(wù)導(dǎo)致報(bào)錯(cuò)。這多發(fā)生在服務(wù)提供者系統(tǒng)部署的時(shí)候，因?yàn)橄到y(tǒng)部署的時(shí)候會(huì)調(diào)用 Spring 容器的 shutdown() 方法， Eureka Server 那里能夠及時(shí)的剔除下線服務(wù)，但是我們上一篇文章中已經(jīng)知道 readOnlyCacheMap 和 readWriteCacheMap 同步間隔是 30S，Client 端拉取實(shí)例信息的間隔也是 30S，這就導(dǎo)致 Eureka Client 端存儲(chǔ)的實(shí)例信息數(shù)據(jù)在一個(gè)臨界時(shí)間范圍內(nèi)都是臟數(shù)據(jù)。

調(diào)整 Eureka 參數(shù)

既然由于 Eureka 本身的設(shè)計(jì)導(dǎo)致會(huì)存在服務(wù)實(shí)例信息延遲更新，那么我們嘗試去修改幾個(gè)參數(shù)來(lái)降低延遲

Client 端設(shè)置服務(wù)拉取間隔3S， eureka.client.registry-fetch-interval-seconds = 3
Server 端設(shè)置讀寫(xiě)緩存同步間隔 3S，eureka.server.response-cache-update-interval-ms=3000

這樣設(shè)置之后經(jīng)過(guò)一段時(shí)間的觀察發(fā)現(xiàn)情況有所改善，但還是存在這個(gè)問(wèn)題，而且并沒(méi)有改善多少。

LoadBalancer 如何獲取實(shí)例信息

在 Eureka 和 OpenFeign 的文章中都有提到，OpenFeign 進(jìn)行遠(yuǎn)程調(diào)用的時(shí)候會(huì)通過(guò)負(fù)載均衡器選取一個(gè)實(shí)例發(fā)起 Http 請(qǐng)求。我們 SpringCloud 版本是 2020，已經(jīng)移除了 ribbon，使用的是 LoadBalancer。

通過(guò) debug OpenFeign 調(diào)用的源碼發(fā)現(xiàn)它是從 DiscoveryClientServiceInstanceListSupplier的構(gòu)造方法獲取實(shí)例信息集合 List<ServiceInstance> 的，內(nèi)部調(diào)用到 CachingServiceInstanceListSupplier 構(gòu)造方法，重點(diǎn)看 CacheFlux.lookup()

public CachingServiceInstanceListSupplier(ServiceInstanceListSupplier delegate, CacheManager cacheManager) {
   super(delegate);
   this.serviceInstances = CacheFlux.lookup(key -> {
      // TODO: configurable cache name
      Cache cache = cacheManager.getCache(SERVICE_INSTANCE_CACHE_NAME);
      if (cache == null) {
         if (log.isErrorEnabled()) {
            log.error("Unable to find cache: " + SERVICE_INSTANCE_CACHE_NAME);
         }
         return Mono.empty();
      }
      List<ServiceInstance> list = cache.get(key, List.class);
      if (list == null || list.isEmpty()) {
         return Mono.empty();
      }
      return Flux.just(list).materialize().collectList();
   }, delegate.getServiceId()).onCacheMissResume(delegate.get().take(1))
         .andWriteWith((key, signals) -> Flux.fromIterable(signals).dematerialize().doOnNext(instances -> {
            Cache cache = cacheManager.getCache(SERVICE_INSTANCE_CACHE_NAME);
            if (cache == null) {
               if (log.isErrorEnabled()) {
                  log.error("Unable to find cache for writing: " + SERVICE_INSTANCE_CACHE_NAME);
               }
            }
            else {
               cache.put(key, instances);
            }
         }).then());
}

這里先去查緩存，緩存有就直接返回，緩存沒(méi)有就去 CompositeDiscoveryClient.getInstances() 查詢。查詢完畢之后會(huì)回調(diào)到 CacheFlux.lookup(param,param2) 第二個(gè)參數(shù)的代碼塊，將結(jié)果放進(jìn)緩存。

@Override
public List<ServiceInstance> getInstances(String serviceId) {
   if (this.discoveryClients != null) {
      for (DiscoveryClient discoveryClient : this.discoveryClients) {
         List<ServiceInstance> instances = discoveryClient.getInstances(serviceId);
         if (instances != null && !instances.isEmpty()) {
            return instances;
         }
      }
   }
   return Collections.emptyList();
}

重點(diǎn)看這個(gè)方法，由于我們使用的是 Eureka 作為注冊(cè)中心。所以這里會(huì)調(diào)用 EurekaDiscoveryClient 的getInstances()，最終我們發(fā)現(xiàn)底層其實(shí)就是從 DiscoveryClient.localRegionApps 獲取的服務(wù)實(shí)例信息。

現(xiàn)在我們清楚了，OpenFeign 調(diào)用時(shí)，負(fù)載均衡策略還不是從 DiscoveryClient.localRegionApps 直接拿的實(shí)例信息，是自己緩存了一份。這樣一來(lái)，不僅要計(jì)算 Eureka 本身的延遲，還要算上緩存時(shí)間。

SpringCloud 中有很多內(nèi)存緩存的實(shí)現(xiàn)，這里我們選擇的是 Caffine

<dependency>
    <groupId>com.github.ben-manes.caffeine</groupId>
    <artifactId>caffeine</artifactId>
    <version>3.0.5</version>
</dependency>

引入依賴即可自動(dòng)配置，從 LoadBalancerCacheProperties 中我們能夠發(fā)現(xiàn)默認(rèn)的緩存時(shí)間是 35S，所以要解決我們的問(wèn)題還需要降低緩存時(shí)間，也可以直接不使用內(nèi)存緩存，每次都從 EurekaClient 拉取過(guò)來(lái)的實(shí)例信息讀取即可。

通過(guò)上面的分析我們可以發(fā)現(xiàn)使用 OpenFeign 內(nèi)部調(diào)用是無(wú)法根治這個(gè)問(wèn)題的，因?yàn)?Eureka 的延遲是無(wú)法根治的，只能說(shuō)在維持機(jī)器性能等各方面的前提下盡可能的縮短數(shù)據(jù)同步定時(shí)任務(wù)的時(shí)間間隔。所以我們可以換個(gè)角度，讓調(diào)用失敗的請(qǐng)求進(jìn)行重試。

LoadBalancer 的兩種負(fù)載均衡策略

通過(guò)源碼調(diào)試，發(fā)現(xiàn)它有兩種負(fù)載均衡策略 RoundRobinLoadBalancer、RandomLoadBalancer，輪詢和隨機(jī)，默認(rèn)的策略是輪詢

LoadBalancerClientConfiguration 類

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

這兩種策略都比較簡(jiǎn)單，沒(méi)什么好說(shuō)的。

輪詢策略存在的問(wèn)題

我們可以觀察下輪詢策略的實(shí)現(xiàn)，它有一個(gè)原子類型的成員變量，用來(lái)記錄下一次請(qǐng)求要落到哪一個(gè)實(shí)例

final AtomicInteger position;

核心邏輯

private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> instances) {
   if (instances.isEmpty()) {
      if (log.isWarnEnabled()) {
         log.warn("No servers available for service: " + serviceId);
      }
      return new EmptyResponse();
   }
   // TODO: enforce order?
   int pos = Math.abs(this.position.incrementAndGet());
   ServiceInstance instance = instances.get(pos % instances.size());
   return new DefaultResponse(instance);
}

可以看到實(shí)現(xiàn)邏輯很簡(jiǎn)單，用 position 自增，然后實(shí)例數(shù)量進(jìn)行求余，達(dá)到輪詢的效果。乍一看好像沒(méi)問(wèn)題，但是它存在這樣一種情況?，F(xiàn)在我們有兩個(gè)實(shí)例 192.168.1.121、192.168.1.122，這時(shí)候兩個(gè)請(qǐng)求 A、B 過(guò)來(lái)，A 請(qǐng)求了 121 的，B 請(qǐng)求了 122 的，然后 A 請(qǐng)求失敗了觸發(fā)重試，由于輪詢機(jī)制 A 重試的實(shí)例又回到了 121 ，這樣就有問(wèn)題了，因?yàn)檫€是失敗，我們要讓重試的請(qǐng)求一定能重試到其他的服務(wù)實(shí)例。

使用 TraceId 實(shí)現(xiàn)自定義負(fù)載均衡策略

因?yàn)橹卦嚨臅r(shí)候是在 OpenFeign 內(nèi)部重新發(fā)起了一次 HTTP 請(qǐng)求，所以 traceId 并沒(méi)有變，我們可以先從 MDC 上下文獲取 traceId，再?gòu)木彺嬷蝎@取 traceId 對(duì)應(yīng)的值，如果沒(méi)有就隨機(jī)生成一個(gè)數(shù)字然后和 RoundRobinLoadBalancer 一樣自增求余，如果緩存中已經(jīng)有了就直接自增求余，這樣就一定能重試到不同的實(shí)例。

這里我們緩存組件還是使用 Caffeine

private final LoadingCache<String, AtomicInteger> positionCache = Caffeine.newBuilder().expireAfterWrite(1, TimeUnit.MINUTES)
      .build(k -> new AtomicInteger(ThreadLocalRandom.current().nextInt(0, 1000)));

private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> serviceInstances) {
   if (serviceInstances.isEmpty()) {
      log.warn("No servers available for service: " + serviceId);
      return new EmptyResponse();
   }
   String traceId = MDC.get("traceId");
   if (traceId == null) {
      traceId = UUID.randomUUID().toString();
   }
   AtomicInteger seed = positionCache.get(traceId);
   int s = seed.getAndIncrement();
   int pos = s % serviceInstances.size();
   return new DefaultResponse(serviceInstances.stream()
         .sorted(Comparator.comparing(ServiceInstance::getInstanceId))
         .collect(Collectors.toList()).get(pos));
}

這個(gè)方法是從哈希哥那里學(xué)到的，他的主頁(yè) juejin.cn/user/501033… 。

完了之后聲明我們自己的負(fù)載均衡器的 Bean

public class FeignLoadBalancerConfiguration {
    @Bean
    public ReactorLoadBalancer<ServiceInstance> reactorServiceInstanceLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSuppliers, Environment environment) {
        String name = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME);
        return new RoundRobinRetryDifferentInstanceLoadBalancer(serviceInstanceListSuppliers,name);
    }
}

之后在主啟動(dòng)類上使用 @LoadBalancerClient 指定我們自定義的負(fù)載均衡器

@LoadBalancerClient(name = "feign-test-product", configuration = FeignLoadBalancerConfiguration.class)

設(shè)置 LoadBalancer Zone

還記得之前 Eureka 我們?yōu)榱私鉀Q本機(jī)調(diào)用的時(shí)候會(huì)通過(guò)負(fù)載均衡調(diào)用到開(kāi)發(fā)環(huán)境的機(jī)器設(shè)置了 zone，SpringCloud LoadBalancer 也提供了這個(gè)配置，并且從源碼中我們可以發(fā)現(xiàn)，最終會(huì)以 LoadBalancer 設(shè)置的為準(zhǔn)，如果沒(méi)有為它設(shè)置，那么會(huì)使用 Eureka 中的 zone 配置，如果設(shè)置了就會(huì)覆蓋 Eureka 的 zone 設(shè)置

EurekaLoadBalancerClientConfiguration.postprocess()

@PostConstruct
public void postprocess() {
   if (!StringUtils.isEmpty(zoneConfig.getZone())) {
      return;
   }
   String zone = getZoneFromEureka();
   if (!StringUtils.isEmpty(zone)) {
      if (LOG.isDebugEnabled()) {
         LOG.debug("Setting the value of '" + LOADBALANCER_ZONE + "' to " + zone);
      }
      zoneConfig.setZone(zone);
   }
}

以上就是詳解SpringCloud LoadBalancer 新一代負(fù)載均衡器的詳細(xì)內(nèi)容，更多關(guān)于SpringCloud LoadBalancer負(fù)載均衡器的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: