当前位置 : 主页 > 编程语言 > 其它开发 >

Feign 进行rpc 调用时使用ribbon负载均衡源码解析

来源:互联网 收集:自由互联 发布时间:2022-06-10
转载请注明出处: Feign客户端接口的动态代理生成是基于JDK的动态代理来实现的,那么在所有的方法调用的时候最终都会走InvocationHandler接口的实现,默认就是ReflectiveFeign.FeignInvocation

转载请注明出处:  

  Feign客户端接口的动态代理生成是基于JDK的动态代理来实现的,那么在所有的方法调用的时候最终都会走InvocationHandler接口的实现,默认就是ReflectiveFeign.FeignInvocationHandler,那我们接下来就来看看,FeignInvocationHandler是如何实现rpc调用的。

  FeignInvocationHandler对于invoke方法的实现。

public class ReflectiveFeign extends Feign {

static class FeignInvocationHandler implements InvocationHandler {
       
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            if (!"equals".equals(method.getName())) {
                if ("hashCode".equals(method.getName())) {
                    return this.hashCode();
                } else {
                    return "toString".equals(method.getName()) ? this.toString() : ((MethodHandler)this.dispatch.get(method)).invoke(args);
                }
            } else {
                try {
                    Object otherHandler = args.length > 0 && args[0] != null ? Proxy.getInvocationHandler(args[0]) : null;
                    return this.equals(otherHandler);
                } catch (IllegalArgumentException var5) {
                    return false;
                }
            }
        }
 }

  实现rpc 调用的方法是 this.dispatch.get(method)).invoke(args)

  从dispatch获取要调用的方法对应的MethodHandler,然后调用MethodHandler的invoke方法。那MethodHandler是什么时候生成的呢?MethodHandler是在构建动态代理的时候生成的, 那MethodHandler作用是什么呢?你可以理解为最终rpc的调用都是基于这个MethodHandler来实现的,每个方法都有对应MethodHandler来实现rpc调用,接下来我们就来看一下MethodHandler的invoke方法的实现。

  MethodHandler是个接口,有两个实现类,一个是DefaultMethodHandler,这个是处理接口中的默认方法的,另一个是SynchronousMethodHandler,这个是实现rpc调用的方法。接下来我们就看看SynchronousMethodHandler关于invoke方法的实现。

public Object invoke(Object[] argv) throws Throwable {
         // 构建了一个RequestTemplate,RequestTemplate可以看成是组装http请求所需各种参数的封装,比如什么情头,body之类的都放在这里面。
        RequestTemplate template = this.buildTemplateFromArgs.create(argv);
         //Options主要是封装了发送请求是连接超时时间和读超时时间的配置
        Options options = this.findOptions(argv);
         // 重试组件,实现重试
        Retryer retryer = this.retryer.clone();

        while(true) {
            try {
                // 进行restTemplate 请求
                return this.executeAndDecode(template, options);
            } catch (RetryableException var9) {
                RetryableException e = var9;

                try {
                    retryer.continueOrPropagate(e);
                } catch (RetryableException var8) {
                    Throwable cause = var8.getCause();
                    if (this.propagationPolicy == ExceptionPropagationPolicy.UNWRAP && cause != null) {
                        throw cause;
                    }

                    throw var8;
                }

                if (this.logLevel != Level.NONE) {
                    this.logger.logRetry(this.metadata.configKey(), this.logLevel);
                }
            }
        }
    

  查看 executeAndDecode 方法实现

Object executeAndDecode(RequestTemplate template, Options options) throws Throwable {
         // 该方法会遍历所有的拦截器RequestInterceptor,通过这个可以扩展实现一些自定义功能,如请求头等;
        Request request = this.targetRequest(template);

        long start = System.nanoTime();

        Response response;
        try {
            // 进行restTemplate 请求,并封装响应信息
            // client 为 LoadBalancerFeignClient,通过这个可以实现与ribbon 的结合
            response = this.client.execute(request, options);
            response = response.toBuilder().request(request).requestTemplate(template).build();
        } catch (IOException var12) {
            if (this.logLevel != Level.NONE) {
                this.logger.logIOException(this.metadata.configKey(), this.logLevel, var12, this.elapsedTime(start));
            } 

            throw FeignException.errorExecuting(request, var12);
        }

        long elapsedTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
        if (this.decoder != null) {
            return this.decoder.decode(response, this.metadata.returnType());
        } else {
            CompletableFuture<Object> resultFuture = new CompletableFuture();
            this.asyncResponseHandler.handleResponse(resultFuture, this.metadata.configKey(), response, this.metadata.returnType(), elapsedTime);
        }
    }

  Client是发送http请求的关键类 ; 当Feign客户端在构建动态代理的时候,填充很多组件到Feign.Builder中,其中有个组件就是Client的实现 ;这个组件的实现是要依赖负载均衡的,也就是这个组件是Feign用来整合Ribbon的入口。

 

Feign是如何通过ribbon实现负载均衡的

  查看 FeignRibbonClientAutoConfiguration 配置类:

@ConditionalOnClass({ILoadBalancer.class, Feign.class})
@ConditionalOnProperty(
    value = {"spring.cloud.loadbalancer.ribbon.enabled"},
    matchIfMissing = true
)
@Configuration(
    proxyBeanMethods = false
)
@AutoConfigureBefore({FeignAutoConfiguration.class})
@EnableConfigurationProperties({FeignHttpClientProperties.class})
@Import({HttpClientFeignLoadBalancedConfiguration.class, OkHttpFeignLoadBalancedConfiguration.class, DefaultFeignLoadBalancedConfiguration.class})
public class FeignRibbonClientAutoConfiguration {
    public FeignRibbonClientAutoConfiguration() {
    }
}

  

@Impot注解导入了三个配置类。

  • HttpClientFeignLoadBalancedConfiguration:基于HttpClient实现http调用的。

  • OkHttpFeignLoadBalancedConfiguration:基于OkHttp实现http调用的。

  • DefaultFeignLoadBalancedConfiguration:默认的,也就是Feign原生的发送http的实现。

  看一下DefaultFeignLoadBalancedConfiguration配置类,因为默认就是这,HttpClientFeignLoadBalancedConfiguration和OkHttpFeignLoadBalancedConfiguration都需要有引入HttpClient和OkHttp依赖才会有用

@Configuration(proxyBeanMethods = false)
class DefaultFeignLoadBalancedConfiguration {
 
  @Bean
  @ConditionalOnMissingBean
  public Client feignClient(CachingSpringLoadBalancerFactory cachingFactory, SpringClientFactory clientFactory) {
      return new LoadBalancerFeignClient(new Client.Default(null, null), cachingFactory, clientFactory);
  }
 
}

  这个配置类很简单,声明了LoadBalancerFeignClient到spring容器,传入了三个参数,一个Client的实现,一个CachingSpringLoadBalancerFactory和一个SpringClientFactory。LoadBalancerFeignClient这个类实现了Client接口,也就数说我们在构建Feign.Builder填充的就是这个对象,也就是上面说feign的执行流程最后用来执行请求的Client的实现。

  Client.Default:就是Feign自己实现的Client,里面封装了真正发送http发送请求的功能,LoadBalancerFeignClient虽然也实现了Client接口,但是这个实现其实是为了整合Ribbon用的,并没有发送http的功能,所以需要有个可以发送http功能的实现。

  这里就说完了Feign整合ribbon的配置类FeignRibbonClientAutoConfiguration,我们也找到了构造Feign.Builder的实现LoadBalancerFeignClient,接下来就来剖析LoadBalancerFeignClient的实现。  

  

public class LoadBalancerFeignClient implements Client {
 
  static final Request.Options DEFAULT_OPTIONS = new Request.Options();
 
  private final Client delegate;
 
  private CachingSpringLoadBalancerFactory lbClientFactory;
 
  private SpringClientFactory clientFactory;
 
  public LoadBalancerFeignClient(Client delegate,CachingSpringLoadBalancerFactory lbClientFactory,SpringClientFactory clientFactory) {
      this.delegate = delegate;
      this.lbClientFactory = lbClientFactory;
      this.clientFactory = clientFactory;
  }
 
  static URI cleanUrl(String originalUrl, String host) {
    String newUrl = originalUrl;
    if (originalUrl.startsWith("https://")) {
        newUrl = originalUrl.substring(0, 8) + originalUrl.substring(8 + host.length());
    } else if (originalUrl.startsWith("http")) {
        newUrl = originalUrl.substring(0, 7) + originalUrl.substring(7 + host.length());
    }
    StringBuffer buffer = new StringBuffer(newUrl);
    if ((newUrl.startsWith("https://") && newUrl.length() == 8)|| (newUrl.startsWith("http://") && newUrl.length() == 7)) {
        buffer.append("/");
    }
      return URI.create(buffer.toString());
  }
 
  @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);
    }
  }
 
  IClientConfig getClientConfig(Request.Options options, String clientName) {
    IClientConfig requestConfig;
    if (options == DEFAULT_OPTIONS) {
      requestConfig = this.clientFactory.getClientConfig(clientName);
    }
    else {
      requestConfig = new FeignOptionsClientConfig(options);
    }
    return requestConfig;
  }
 
  protected IOException findIOException(Throwable t) {
    if (t == null) {
      return null;
    }
    if (t instanceof IOException) {
      return (IOException) t;
    }
    return findIOException(t.getCause());
  }
 
  public Client getDelegate() {
    return this.delegate;
  }
 
  private FeignLoadBalancer lbClient(String clientName) {
    return this.lbClientFactory.create(clientName);
  }
 
  static class FeignOptionsClientConfig extends DefaultClientConfigImpl {
 
    FeignOptionsClientConfig(Request.Options options) {
      setProperty(CommonClientConfigKey.ConnectTimeout,
          options.connectTimeoutMillis());
      setProperty(CommonClientConfigKey.ReadTimeout, options.readTimeoutMillis());
    }
 
    @Override
    public void loadProperties(String clientName) {
 
    }
 
    @Override
    public void loadDefaultValues() {
 
    }
 
  }
 
}

  在动态代理调用的那里我们得出一个结论,那就是最后会调用Client接口的execute方法的实现,所以我们就看一下execute方法的实现,这里就是一堆操作,从请求的URL中拿到了clientName,也就是服务名。

  为什么可以拿到服务名?

  其实很简单,OpenFeign构建动态代理的时候,传入了一个HardCodedTarget,当时说在构建HardCodedTarget的时候传入了一个url,那个url当时说了其实就是http://服务名,所以到这里,虽然有具体的请求接口的路径,但是还是类似 http://服务名/api/sayHello这种,所以可以通过路径拿到你锁请求的服务名。

  拿到服务名之后,再拿到了一个配置类IClientConfig,最后调用lbClient,我们看一下lbClient的方法实现。

private FeignLoadBalancer lbClient(String clientName) {
    // 调用CachingSpringLoadBalancerFactory的create方法,从缓存中获取一个FeignLoadBalancer,获取不到就创建一个
    return this.lbClientFactory.create(clientName);
}



3.查看 FeignLoadBalancer 源码

  核心源码:

public class FeignLoadBalancer extends
    AbstractLoadBalancerAwareClient<FeignLoadBalancer.RibbonRequest, FeignLoadBalancer.RibbonResponse> {
 
  private final RibbonProperties ribbon;
 
  protected int connectTimeout;
 
  protected int readTimeout;
 
  protected IClientConfig clientConfig;
 
  protected ServerIntrospector serverIntrospector;
 
  public FeignLoadBalancer(ILoadBalancer lb, IClientConfig clientConfig,
      ServerIntrospector serverIntrospector) {
    super(lb, clientConfig);
    this.setRetryHandler(RetryHandler.DEFAULT);
    this.clientConfig = clientConfig;
    this.ribbon = RibbonProperties.from(clientConfig);
    RibbonProperties ribbon = this.ribbon;
    this.connectTimeout = ribbon.getConnectTimeout();
    this.readTimeout = ribbon.getReadTimeout();
    this.serverIntrospector = serverIntrospector;
  }
 
  @Override
  public RibbonResponse execute(RibbonRequest request, IClientConfig configOverride)
      throws IOException {
    Request.Options options;
    if (configOverride != null) {
      RibbonProperties override = RibbonProperties.from(configOverride);
      options = new Request.Options(override.connectTimeout(this.connectTimeout),
          override.readTimeout(this.readTimeout));
    }
    else {
      options = new Request.Options(this.connectTimeout, this.readTimeout);
    }
    Response response = request.client().execute(request.toRequest(), options);
    return new RibbonResponse(request.getUri(), response);
  }
 }

  FeignLoadBalancer继承自AbstractLoadBalancerAwareClient,AbstractLoadBalancerAwareClient类主要作用是通过ILoadBalancer组件获取一个Server,然后基于这个Server重构了URI,也就是将你的请求路径http://服务名/api/sayHello转换成类似http://192.168.1.101:8088/api/sayHello这种路径,也就是将原服务名替换成服务所在的某一台机器ip和端口,替换之后就交由子类实现的exceut方法来发送http请求。

  所以我们知道调用executeWithLoadBalancer之后,就会重构请求路径,将服务名替换成某个具体的服务器所在的ip和端口,之后交给子类execute来处理,对于这里来说,也就是FeignLoadBalancer的execute方法,因为FeignLoadBalancer继承AbstractLoadBalancerAwareClient。

  直接定位到execute方法最核心的一行代码

Response response = request.client().execute(request.toRequest(), options);

  request.client()就会拿到构建LoadBalancerFeignClient传入的那个Client的实现,这个Client的实现是具体发送请求的实现,默认的就是Client.Default类(不是默认就有可能是基于HttpClient或者是OkHttp的实现)。所以这行代码就是基于这个Client就成功的发送了Http请求,拿到响应,然后将这个Response 封装成一个RibbonResponse返回,最后就返回给MethodHandler,然后解析响应,封装成方法的返回值返回给调用者。

  其实到这里就完全知道Feign是如何整合Ribbon的,LoadBalancerFeignClient其实是OpenFeign适配Ribbon的入口,FeignLoadBalancer才是真正实现选择负载均衡,发送http请求的组件,因为他继承了AbstractLoadBalancerAwareClient。

 

参考文章:

  https://blog.csdn.net/u010342147/article/details/123669493

  https://blog.csdn.net/weixin_45630885/article/details/124391934

  https://blog.csdn.net/weixin_45630885/article/details/124533413

 

 

上一篇:@EnableFeignClients注解源码解析
下一篇:没有了
网友评论