中文字幕av专区_日韩电影在线播放_精品国产精品久久一区免费式_av在线免费观看网站

溫馨提示×

溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊×
其他方式登錄
點擊 登錄注冊 即表示同意《億速云用戶服務條款》

Spring Cloud Feign組件實例解析

發布時間:2020-10-03 08:27:47 來源:腳本之家 閱讀:205 作者:技術與人生 欄目:編程語言

這篇文章主要介紹了Spring Cloud Feign組件實例解析,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

采用Spring Cloud微服務框架后,經常會涉及到服務間調用,服務間調用采用了Feign組件。

由于之前有使用dubbo經驗。dubbo的負載均衡策略(輪訓、最小連接數、隨機輪訓、加權輪訓),dubbo失敗策略(快速失敗、失敗重試等等),

所以Feign負載均衡策略的是什么? 失敗后是否會重試,重試策略又是什么?帶這個疑問,查了一些資料,最后還是看了下代碼。畢竟代碼就是一切

Spring boot集成Feign的大概流程:

1、利用FeignAutoConfiguration自動配置。并根據EnableFeignClients 自動注冊產生Feign的代理類。

2、注冊方式利用FeignClientFactoryBean,熟悉Spring知道FactoryBean 產生bean的工廠,有個重要方法getObject產生FeignClient容器bean

3、同時代理類中使用hystrix做資源隔離,Feign代理類中 構造 RequestTemplate ,RequestTemlate要做的向負載均衡選中的server發送http請求,并進行編碼和解碼一系列操作。

下面只是粗略的看了下整體流程,先有整體再有細節吧,下面利用IDEA看下細節:

一、Feign失敗重試

SynchronousMethodHandler的方法中的處理邏輯:

@Override
 public Object invoke(Object[] argv) throws Throwable {
  RequestTemplate template = buildTemplateFromArgs.create(argv);
  Retryer retryer = this.retryer.clone();
  while (true) {
   try {
    return executeAndDecode(template);
   } catch (RetryableException e) {
    retryer.continueOrPropagate(e);
    if (logLevel != Logger.Level.NONE) {
     logger.logRetry(metadata.configKey(), logLevel);
    }
    continue;
   }
  }
 }
  • 上面的邏輯很簡單。構造 template 并去進行服務間的http調用,然后對返回結果進行解碼
  • 當拋出 RetryableException 后,異常邏輯是否重試? 重試多少次? 帶這個問題,看了retryer.continueOrPropagate(e);

具體邏輯如下:

public void continueOrPropagate(RetryableException e) {
   if (attempt++ >= maxAttempts) {
    throw e;
   }
 
   long interval;
   if (e.retryAfter() != null) {
    interval = e.retryAfter().getTime() - currentTimeMillis();
    if (interval > maxPeriod) {
     interval = maxPeriod;
    }
    if (interval < 0) {
     return;
    }
   } else {
    interval = nextMaxInterval();
   }
   try {
    Thread.sleep(interval);
   } catch (InterruptedException ignored) {
    Thread.currentThread().interrupt();
   }
   sleptForMillis += interval;
  }
  • 當重試次數大于默認次數5時候,直接拋出異常,不在重試
  • 否則每隔一段時間 默認值最大1ms 后重試一次。

這就Feign這塊的重試這塊的粗略邏輯,由于之前工作中一直使用dubbo。同樣是否需要將生產環境中重試操作關閉?

思考:之前dubbo生產環境的重試操作都會關閉。原因有幾個:

  • 一般第一次失敗,重試也會失敗,極端情況下不斷的重試,會占用大量dubbo連接池,造成連接池被打滿,影響核心功能
  • 也是比較重要的一點原因,重試帶來的業務邏輯的影響,即如果接口不是冪等的,重試會帶來業務邏輯的錯誤,引發問題

二、Feign負載均衡策略

那么負載均衡的策略又是什么呢?由上圖中可知 executeAndDecode(template)

Object executeAndDecode(RequestTemplate template) throws Throwable {
  Request request = targetRequest(template);

  if (logLevel != Logger.Level.NONE) {
   logger.logRequest(metadata.configKey(), logLevel, request);
  }

  Response response;
  long start = System.nanoTime();
  try {
   response = client.execute(request, options);
   // ensure the request is set. TODO: remove in Feign 10
   response.toBuilder().request(request).build();
  } catch (IOException e) {
   if (logLevel != Logger.Level.NONE) {
    logger.logIOException(metadata.configKey(), logLevel, e, elapsedTime(start));
   }
   throw errorExecuting(request, e);
  }
  long elapsedTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);

  boolean shouldClose = true;
  try {
   if (logLevel != Logger.Level.NONE) {
    response =
      logger.logAndRebufferResponse(metadata.configKey(), logLevel, response, elapsedTime);
    // ensure the request is set. TODO: remove in Feign 10
    response.toBuilder().request(request).build();
   }
   if (Response.class == metadata.returnType()) {
    if (response.body() == null) {
     return response;
    }
    if (response.body().length() == null ||
        response.body().length() > MAX_RESPONSE_BUFFER_SIZE) {
     shouldClose = false;
     return response;
    }
    // Ensure the response body is disconnected
    byte[] bodyData = Util.toByteArray(response.body().asInputStream());
    return response.toBuilder().body(bodyData).build();
   }
   if (response.status() >= 200 && response.status() < 300) {
    if (void.class == metadata.returnType()) {
     return null;
    } else {
     return decode(response);
    }
   } else if (decode404 && response.status() == 404 && void.class != metadata.returnType()) {
    return decode(response);
   } else {
    throw errorDecoder.decode(metadata.configKey(), response);
   }
  } catch (IOException e) {
   if (logLevel != Logger.Level.NONE) {
    logger.logIOException(metadata.configKey(), logLevel, e, elapsedTime);
   }
   throw errorReading(request, response, e);
  } finally {
   if (shouldClose) {
    ensureClosed(response.body());
   }
  }
 }

概括的說主要做了兩件事:發送HTTP請求,解碼響應數據

想看的負載均衡應該在11行 response = client.execute(request, options); 而client的實現方式有兩種 Default、LoadBalancerFeignClient

猜的話應該是LoadBalancerFeignClient,帶這個問題去看源碼(其實個人更喜歡帶著問題看源碼,沒有目的一是看很難將復雜的源碼關聯起來,二是很容易迷失其中)

果然通過一番查找發現 Client 實例就是LoadBalancerFeignClient,而設置這個Client就是通過上面說的FeignClientFactoryBean的getObject方法中設置的,具體不說了

下面重點看LoadBalancerFeignClient execute(request, options)

@Override
  public Response execute(Request request, Request.Options options) throws IOException {
    try {
      URI asUri = URI.create(request.url());
      String clientName = asUri.getHost();
      URI uriWithoutHost = cleanUrl(request.url(), clientName);
      FeignLoadBalancer.RibbonRequest ribbonRequest = new FeignLoadBalancer.RibbonRequest(
          this.delegate, request, uriWithoutHost);

      IClientConfig requestConfig = getClientConfig(options, clientName);
      return lbClient(clientName).executeWithLoadBalancer(ribbonRequest,
          requestConfig).toResponse();
    }
    catch (ClientException e) {
      IOException io = findIOException(e);
      if (io != null) {
        throw io;
      }
      throw new RuntimeException(e);
    }
  }

通過幾行代碼比較重要的點RibbonRequest ,原來Feign負載均衡還是通過Ribbon實現的,那么Ribbo又是如何實現負載均衡的呢?

public Observable<T> submit(final ServerOperation<T> operation) {
    final ExecutionInfoContext context = new ExecutionInfoContext();
    
    if (listenerInvoker != null) {
      try {
        listenerInvoker.onExecutionStart();
      } catch (AbortExecutionException e) {
        return Observable.error(e);
      }
    }

    final int maxRetrysSame = retryHandler.getMaxRetriesOnSameServer();
    final int maxRetrysNext = retryHandler.getMaxRetriesOnNextServer();

    // Use the load balancer
    Observable<T> o = 
        (server == null ? selectServer() : Observable.just(server))
        .concatMap(new Func1<Server, Observable<T>>() {
          @Override
          // Called for each server being selected
          public Observable<T> call(Server server) {
            context.setServer(server);
            final ServerStats stats = loadBalancerContext.getServerStats(server);
            
            // Called for each attempt and retry
            Observable<T> o = Observable
                .just(server)
                .concatMap(new Func1<Server, Observable<T>>() {
                  @Override
                  public Observable<T> call(final Server server) {
                    context.incAttemptCount();
                    loadBalancerContext.noteOpenConnection(stats);
                    
                    if (listenerInvoker != null) {
                      try {
                        listenerInvoker.onStartWithServer(context.toExecutionInfo());
                      } catch (AbortExecutionException e) {
                        return Observable.error(e);
                      }
                    }
                    
                    final Stopwatch tracer = loadBalancerContext.getExecuteTracer().start();
                    
                    return operation.call(server).doOnEach(new Observer<T>() {
                      private T entity;
                      @Override
                      public void onCompleted() {
                        recordStats(tracer, stats, entity, null);
                        // TODO: What to do if onNext or onError are never called?
                      }

                      @Override
                      public void onError(Throwable e) {
                        recordStats(tracer, stats, null, e);
                        logger.debug("Got error {} when executed on server {}", e, server);
                        if (listenerInvoker != null) {
                          listenerInvoker.onExceptionWithServer(e, context.toExecutionInfo());
                        }
                      }

                      @Override
                      public void onNext(T entity) {
                        this.entity = entity;
                        if (listenerInvoker != null) {
                          listenerInvoker.onExecutionSuccess(entity, context.toExecutionInfo());
                        }
                      }              
                      
                      private void recordStats(Stopwatch tracer, ServerStats stats, Object entity, Throwable exception) {
                        tracer.stop();
                        loadBalancerContext.noteRequestCompletion(stats, entity, exception, tracer.getDuration(TimeUnit.MILLISECONDS), retryHandler);
                      }
                    });
                  }
                });
            
            if (maxRetrysSame > 0) 
              o = o.retry(retryPolicy(maxRetrysSame, true));
            return o;
          }
        });
      
    if (maxRetrysNext > 0 && server == null) 
      o = o.retry(retryPolicy(maxRetrysNext, false));
    
    return o.onErrorResumeNext(new Func1<Throwable, Observable<T>>() {
      @Override
      public Observable<T> call(Throwable e) {
        if (context.getAttemptCount() > 0) {
          if (maxRetrysNext > 0 && context.getServerAttemptCount() == (maxRetrysNext + 1)) {
            e = new ClientException(ClientException.ErrorType.NUMBEROF_RETRIES_NEXTSERVER_EXCEEDED,
                "Number of retries on next server exceeded max " + maxRetrysNext
                + " retries, while making a call for: " + context.getServer(), e);
          }
          else if (maxRetrysSame > 0 && context.getAttemptCount() == (maxRetrysSame + 1)) {
            e = new ClientException(ClientException.ErrorType.NUMBEROF_RETRIES_EXEEDED,
                "Number of retries exceeded max " + maxRetrysSame
                + " retries, while making a call for: " + context.getServer(), e);
          }
        }
        if (listenerInvoker != null) {
          listenerInvoker.onExecutionFailed(e, context.toFinalExecutionInfo());
        }
        return Observable.error(e);
      }
    });
  }

通過上面代碼分析,發現Ribbon和Hystrix一樣都是利用了rxjava看來有必要掌握下rxjava了又。這里面 比較重要的就是17行,

selectServer() 方法選擇指定的Server,負載均衡的策略主要是有ILoadBalancer接口不同實現方式:

  • BaseLoadBalancer采用的規則為RoundRobinRule 輪訓規則
  • DynamicServerListLoadBalancer繼承了BaseLoadBalancer,主要運行時改變Server列表
  • NoOpLoadBalancer 什么操作都不做
  • ZoneAwareLoadBalancer 功能主要是根據區域Zone分組的實例列表

以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支持億速云。

向AI問一下細節

免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。

AI

固始县| 泉州市| 双鸭山市| 远安县| 车险| 寻甸| 白沙| 恭城| 泾川县| 灵台县| 昆明市| 罗定市| 克山县| 玉林市| 临夏县| 中方县| 太保市| 榆树市| 遂昌县| 台东市| 新竹县| 南开区| 德清县| 湛江市| 安义县| 祁阳县| 安顺市| 故城县| 泰安市| 磴口县| 和田县| 蒙城县| 津市市| 湖口县| 泰宁县| 宁化县| 西林县| 百色市| 凤阳县| 浪卡子县| 宝鸡市|