SpringCloud整合Feign的调用源码流程解析

Feign的内部组件

在说feign的整个初始化过程之前，我们先来看看feign的内部组件有哪些

我们直接来到Feign这个类，里面有一个内部类Builder，不难猜到创建一个Feign对象应该就是使用构造者模式

public static class Builder {private final List<RequestInterceptor> requestInterceptors =new ArrayList<RequestInterceptor>();private Logger.Level logLevel = Logger.Level.NONE;private Contract contract = new Contract.Default();private Client client = new Client.Default(null, null);private Retryer retryer = new Retryer.Default();private Logger logger = new NoOpLogger();private Encoder encoder = new Encoder.Default();private Decoder decoder = new Decoder.Default();private QueryMapEncoder queryMapEncoder = new QueryMapEncoder.Default();private ErrorDecoder errorDecoder = new ErrorDecoder.Default();private Options options = new Options();private InvocationHandlerFactory invocationHandlerFactory =new InvocationHandlerFactory.Default();private boolean decode404;private boolean closeAfterDecode = true;private ExceptionPropagationPolicy propagationPolicy = NONE;// 省略上面各组件的构建方法public <T> T target(Class<T> apiType, String url) {return target(new HardCodedTarget<T>(apiType, url));}public <T> T target(Target<T> target) {return build().newInstance(target);}public Feign build() {SynchronousMethodHandler.Factory synchronousMethodHandlerFactory =new SynchronousMethodHandler.Factory(client, retryer, requestInterceptors, logger,logLevel, decode404, closeAfterDecode, propagationPolicy);ParseHandlersByName handlersByName =new ParseHandlersByName(contract, options, encoder, decoder, queryMapEncoder,errorDecoder, synchronousMethodHandlerFactory);return new ReflectiveFeign(handlersByName, invocationHandlerFactory, queryMapEncoder);}
}

从上面我们可以看到Feign里面的组件很多，其中比较重要的就是Encoder，Decoder编解码组件，Contract约定组件（用来解析feign的请求注解以及路径，与springcloud整合之后具体用的是SpringMVCContract，它能够识别解析SpringMVC的注解），Options请求连接超时时间，响应超时时间等等，最后有一个特别重要的组件就是Client组件，这个组件估计不难猜到就是用来发起客户端请求的组件，因为客户端请求框架又很多，feign原生默认用的是jdk自带的HttpURLConnection去进行客户端的网络请求，但是与springcloud整合之后，就能够选择HttpClient或者OkHttp进行整合了

Feign的调用原理

为什么feign能够通过一个没有实现类的接口直接去调用其方法？

答案很简单就是动态代理，我们下面来看下feign是怎么对接口生成代理对象的

 public <T> T target(Target<T> target) {return build().newInstance(target);}public Feign build() {SynchronousMethodHandler.Factory synchronousMethodHandlerFactory =new SynchronousMethodHandler.Factory(client, retryer, requestInterceptors, logger,logLevel, decode404, closeAfterDecode, propagationPolicy);ParseHandlersByName handlersByName =new ParseHandlersByName(contract, options, encoder, decoder, queryMapEncoder,errorDecoder, synchronousMethodHandlerFactory);return new ReflectiveFeign(handlersByName, invocationHandlerFactory, queryMapEncoder);}

我们来看Build构造类的build方法，这个方法返回了一个ReflectiveFeign对象，我们来看这个类

public <T> T newInstance(Target<T> target) {// 返回把接口类里面的接口方法的名字（类名#方法名）以及这个方法对应的MethodHandler（SynchronousMethodHandler）// 具体的发起请求的逻辑就在MethodHandler中Map<String, MethodHandler> nameToHandler = targetToHandlersByName.apply(target);// method对象->MethodHandler对象Map<Method, MethodHandler> methodToHandler = new LinkedHashMap<Method, MethodHandler>();List<DefaultMethodHandler> defaultMethodHandlers = new LinkedList<DefaultMethodHandler>();for (Method method : target.type().getMethods()) {if (method.getDeclaringClass() == Object.class) {continue;} else if (Util.isDefault(method)) {DefaultMethodHandler handler = new DefaultMethodHandler(method);defaultMethodHandlers.add(handler);methodToHandler.put(method, handler);} else {methodToHandler.put(method, nameToHandler.get(Feign.configKey(target.type(), method)));}}// jdk动态代理InvocationHandler handler = factory.create(target, methodToHandler);T proxy = (T) Proxy.newProxyInstance(target.type().getClassLoader(),new Class<?>[] {target.type()}, handler);for (DefaultMethodHandler defaultMethodHandler : defaultMethodHandlers) {defaultMethodHandler.bindTo(proxy);}return proxy;
}

可以看到这个类有一个newInstance方法，这个方法里面使用了jdk的动态代理并且返回代理对象，那么当我们调用接口的方法时候，代理对象又是如何代理方法的呢？我们直接看它的代理逻辑类InvocationHandler（FeignInvocationHandler）

static class FeignInvocationHandler implements InvocationHandler {private final Target target;private final Map<Method, MethodHandler> dispatch;FeignInvocationHandler(Target target, Map<Method, MethodHandler> dispatch) {this.target = checkNotNull(target, "target");this.dispatch = checkNotNull(dispatch, "dispatch for %s", target);}@Overridepublic Object invoke(Object proxy, Method method, Object[] args) throws Throwable {if ("equals".equals(method.getName())) {try {Object otherHandler =args.length > 0 && args[0] != null ? Proxy.getInvocationHandler(args[0]) : null;return equals(otherHandler);} catch (IllegalArgumentException e) {return false;}} else if ("hashCode".equals(method.getName())) {return hashCode();} else if ("toString".equals(method.getName())) {return toString();}// 直接根据请求的方法得到对应的MethodHandlerreturn dispatch.get(method).invoke(args);}@Overridepublic boolean equals(Object obj) {if (obj instanceof FeignInvocationHandler) {FeignInvocationHandler other = (FeignInvocationHandler) obj;return target.equals(other.target);}return false;}@Overridepublic int hashCode() {return target.hashCode();}@Overridepublic String toString() {return target.toString();}
}

上面的那个dispatch属性就是之前的methodToHandler，通过要调用的method得到MethodHandler并且执行它的invoke方法，invoke方法里面就是发起服务请求

feign.SynchronousMethodHandler#invoke

@Override
public Object invoke(Object[] argv) throws Throwable {// 根据方法参数构造出请求模板对象RequestTemplate template = buildTemplateFromArgs.create(argv);// 可以在方法参数上面指定请求配置对象OptionsOptions options = findOptions(argv);Retryer retryer = this.retryer.clone();while (true) {try {// 开始请求return executeAndDecode(template, options);} catch (RetryableException e) {try {// 抛异常，重试retryer.continueOrPropagate(e);} catch (RetryableException th) {Throwable cause = th.getCause();if (propagationPolicy == UNWRAP && cause != null) {throw cause;} else {throw th;}}if (logLevel != Logger.Level.NONE) {logger.logRetry(metadata.configKey(), logLevel);}continue;}}
}

feign.SynchronousMethodHandler#executeAndDecode

Object executeAndDecode(RequestTemplate template, Options options) throws Throwable {// 根据请求模板对象得到请求request对象Request request = targetRequest(template);if (logLevel != Logger.Level.NONE) {logger.logRequest(metadata.configKey(), logLevel, request);}Response response;long start = System.nanoTime();try {// 通过具体的client组件去进行网络请求response = client.execute(request, options);} 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);}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 disconnectedbyte[] 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 {Object result = decode(response);shouldClose = closeAfterDecode;return result;}} else if (decode404 && response.status() == 404 && void.class != metadata.returnType()) {Object result = decode(response);shouldClose = closeAfterDecode;return result;} 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());}}
}

可以看到response = client.execute(request, options)这句代码，这里就是通过具体的client组件去发起网络请求了，而这个具体的client是哪个呢？当我们引入了springcloud-feign之后，springcloud会帮我们自动地把ribbon给整合进来了，所以这里的client就是LoaderBalancerClient，这个client组件整合了ribbon的负载均衡功能

org.springframework.cloud.openfeign.ribbon.LoadBalancerFeignClient#execute

public Response execute(Request request, Request.Options options) throws IOException {try {// 得到uri对象URI asUri = URI.create(request.url());// 得到host，也就是我们的服务名String clientName = asUri.getHost();// 把url上面的服务名去掉URI uriWithoutHost = cleanUrl(request.url(), clientName);// 把client，request对象以及url封装成ribbon的一个ClientRequest请求对象// 因为下面调用的是ribbon的原生api，所以这里封装成ClientRequest请求对象是为了适配ribbon的apiFeignLoadBalancer.RibbonRequest ribbonRequest = new FeignLoadBalancer.RibbonRequest(this.delegate, request, uriWithoutHost);IClientConfig requestConfig = getClientConfig(options, clientName);// 获取FeignLoadBalancer对象，该对象继承ribbon原生的扩展抽象类AbstractLoadBalancerAwareClientreturn lbClient(clientName).executeWithLoadBalancer(ribbonRequest, requestConfig).toResponse();}catch (ClientException e) {IOException io = findIOException(e);if (io != null) {throw io;}throw new RuntimeException(e);}
}

com.netflix.client.AbstractLoadBalancerAwareClient#executeWithLoadBalancer(S, com.netflix.client.config.IClientConfig)

public T executeWithLoadBalancer(final S request, final IClientConfig requestConfig) throws ClientException {// 构造LoadBalancerCommand对象LoadBalancerCommand<T> command = buildLoadBalancerCommand(request, requestConfig);try {return command.submit(new ServerOperation<T>() {@Overridepublic Observable<T> call(Server server) {// 重写请求的urlURI finalUri = reconstructURIWithServer(server, request.getUri());S requestForServer = (S) request.replaceUri(finalUri);try {// 调用真正的client发起请求return Observable.just(AbstractLoadBalancerAwareClient.this.execute(requestForServer, requestConfig));} catch (Exception e) {return Observable.error(e);}}}).toBlocking().single();} catch (Exception e) {Throwable t = e.getCause();if (t instanceof ClientException) {throw (ClientException) t;} else {throw new ClientException(e);}}}

这里的关键代码在LoadBalancerCommand的submit方法上

com.netflix.loadbalancer.reactive.LoadBalancerCommand#submit

public Observable<T> submit(final ServerOperation<T> operation) {// ......Observable<T> o = // 选择目标服务实例(server == null ? selectServer() : Observable.just(server)).concatMap(new Func1<Server, Observable<T>>() {@Override// Called for each server being selectedpublic Observable<T> call(Server server) {context.setServer(server);final ServerStats stats = loadBalancerContext.getServerStats(server);// Called for each attempt and retryObservable<T> o = Observable.just(server).concatMap(new Func1<Server, Observable<T>>() {@Overridepublic Observable<T> call(final Server server) {// ......// 根据目标服务实例信息构造真实请求的urlreturn operation.call(server).doOnEach(new Observer<T>() {// ......// ......
}

这段代码很长，我们只看重点的那些，其中一开始的时候会调用selectServer方法去选择目标服务实例，那么是如何选择目标服务实例的呢？这里我们猜想应该就是使用ribbon的LoadBlancer组件去进行目标服务实例的筛选

private Observable<Server> selectServer() {return Observable.create(new OnSubscribe<Server>() {@Overridepublic void call(Subscriber<? super Server> next) {try {Server server = loadBalancerContext.getServerFromLoadBalancer(loadBalancerURI, loadBalancerKey);   next.onNext(server);next.onCompleted();} catch (Exception e) {next.onError(e);}}});
}

public Server getServerFromLoadBalancer(@Nullable URI original, @Nullable Object loadBalancerKey) throws ClientException {// ......ILoadBalancer lb = getLoadBalancer();if (host == null) {if (lb != null){Server svc = lb.chooseServer(loadBalancerKey);// ......
}

果然，最终就是调用了LoadBalancer组件的chooseServer方法，里面就是我们熟悉的ribbon选择目标服务实例的逻辑了

org.springframework.cloud.openfeign.ribbon.FeignLoadBalancer#reconstructURIWithServer

public URI reconstructURIWithServer(Server server, URI original) {URI uri = updateToSecureConnectionIfNeeded(original, this.clientConfig,this.serverIntrospector, server);return super.reconstructURIWithServer(server, uri);
}

当选择返回了具体的某一个目标服务实例的时候，就需要对这个实例的请求url进行重写了，updateToSecureConnectionIfNeeded方法中作用就是处理请求协议的，会把http更改为https，然后再执行父类的实现

public URI reconstructURIWithServer(Server server, URI original) {String host = server.getHost();int port = server.getPort();String scheme = server.getScheme();if (host.equals(original.getHost()) && port == original.getPort()&& scheme == original.getScheme()) {return original;}if (scheme == null) {scheme = original.getScheme();}if (scheme == null) {scheme = deriveSchemeAndPortFromPartialUri(original).first();}try {StringBuilder sb = new StringBuilder();sb.append(scheme).append("://");if (!Strings.isNullOrEmpty(original.getRawUserInfo())) {sb.append(original.getRawUserInfo()).append("@");}sb.append(host);if (port >= 0) {sb.append(":").append(port);}sb.append(original.getRawPath());if (!Strings.isNullOrEmpty(original.getRawQuery())) {sb.append("?").append(original.getRawQuery());}if (!Strings.isNullOrEmpty(original.getRawFragment())) {sb.append("#").append(original.getRawFragment());}URI newURI = new URI(sb.toString());return newURI;            } catch (URISyntaxException e) {throw new RuntimeException(e);}
}

在父类的实现中就是进行了一番字符串的拼接操作，把url的host替换为选择的服务实例的host

org.springframework.cloud.openfeign.ribbon.FeignLoadBalancer#execute

得到重写的请求url之后，就要真正地发起请求了

public RibbonResponse execute(RibbonRequest request, IClientConfig configOverride)throws IOException {Request.Options options;if (configOverride != null) {// 如果ribbon的配置不为空，就用ribbon的配置进行覆盖连接超时配置和响应超时配置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);}// 这里就是使用原始的client去发起请求，jdk自带，httpclient，okhttpResponse response = request.client().execute(request.toRequest(), options);return new RibbonResponse(request.getUri(), response);
}