简单研究下RestTemplate、 ribbon、 OpenFeign 关系 三者之间的关系。
1. RestTemplate
RestTemplate 使用的是: spring-web 包下面的http 模块的http包中的API。 也就是Spring 自己封装的一套的httpclient API， 下面还是走java 的HttpurlConnection 建立连接然后传输数据。从名字也可以看出其是作为一个模板类可以在项目中使用。类似的还有RedisTemplate 等工具类。
可以单独使用，单独使用的时候相当于直接连接到url， 不走微服务里面的服务发现以及负载均衡等机制。
1. 测试直接url
1. 注入bean
@Bean public RestTemplate getRestTemplate() { return new RestTemplate(); }
2. 添加测试url
@GetMapping("/test")
public String test() {
return restTemplate.getForObject("http://news.baidu.com/guonei", String.class);
}
3. 查看调用链： 断点打在java.net.Socket#connect(java.net.SocketAddress, int) 。 通过HttpURLConnection 建立连接会调用到这里进行连接, 调用链如下：
2. 增加负载均衡能力
1. 注入bean
@Bean @LoadBalanced public RestTemplate getRestTemplate() { return new RestTemplate(); }
2. 测试：
private static final String PAYMENT_URL = "http://CLOUD-PAYMENT-SERVICE"; @GetMapping("/pay/getServerPort") public JSONResultUtil<String> getServerPort() { return restTemplate.getForObject(PAYMENT_URL + "/pay/getServerPort", JSONResultUtil.class); }
3. 查看其调用链：
上面增加@LoadBalanced 注解之后，给RestTemplate 增加了org.springframework.cloud.client.loadbalancer.LoadBalancerInterceptor 拦截器， 会走ribbon 进行服务负载均衡的机制。
可以看到最终经过ribbon 的Inteceptor 之后还是会交给Spring的web.http.client 相关的API进行处理。
列出几个重要的方法：
1》 org.springframework.http.client.InterceptingClientHttpRequest.InterceptingRequestExecution#execute 有拦截器的情况进行拦截器拦截
@Override public ClientHttpResponse execute(HttpRequest request, byte[] body) throws IOException { if (this.iterator.hasNext()) { ClientHttpRequestInterceptor nextInterceptor = this.iterator.next(); return nextInterceptor.intercept(request, body, this); } else { HttpMethod method = request.getMethod(); Assert.state(method != null, "No standard HTTP method"); ClientHttpRequest delegate = requestFactory.createRequest(request.getURI(), method); request.getHeaders().forEach((key, value) -> delegate.getHeaders().addAll(key, value)); if (body.length > 0) { if (delegate instanceof StreamingHttpOutputMessage) { StreamingHttpOutputMessage streamingOutputMessage = (StreamingHttpOutputMessage) delegate; streamingOutputMessage.setBody(outputStream -> StreamUtils.copy(body, outputStream)); } else { StreamUtils.copy(body, delegate.getBody()); } } return delegate.execute(); } }
2》 调用到 org.springframework.cloud.client.loadbalancer.LoadBalancerInterceptor#intercept：
@Override public ClientHttpResponse intercept(final HttpRequest request, final byte[] body, final ClientHttpRequestExecution execution) throws IOException { final URI originalUri = request.getURI(); String serviceName = originalUri.getHost(); Assert.state(serviceName != null, "Request URI does not contain a valid hostname: " + originalUri); return this.loadBalancer.execute(serviceName, this.requestFactory.createRequest(request, body, execution)); }
3》 调用到org.springframework.cloud.netflix.ribbon.RibbonLoadBalancerClient#execute(java.lang.String, org.springframework.cloud.client.loadbalancer.LoadBalancerRequest<T>, java.lang.Object)：
public <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint) throws IOException { ILoadBalancer loadBalancer = getLoadBalancer(serviceId); Server server = getServer(loadBalancer, hint); if (server == null) { throw new IllegalStateException("No instances available for " + serviceId); } RibbonServer ribbonServer = new RibbonServer(serviceId, server, isSecure(server, serviceId), serverIntrospector(serviceId).getMetadata(server)); return execute(serviceId, ribbonServer, request); } @Override public <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException { Server server = null; if (serviceInstance instanceof RibbonServer) { server = ((RibbonServer) serviceInstance).getServer(); } if (server == null) { throw new IllegalStateException("No instances available for " + serviceId); } RibbonLoadBalancerContext context = this.clientFactory .getLoadBalancerContext(serviceId); RibbonStatsRecorder statsRecorder = new RibbonStatsRecorder(context, server); try { T returnVal = request.apply(serviceInstance); statsRecorder.recordStats(returnVal); return returnVal; } // catch IOException and rethrow so RestTemplate behaves correctly catch (IOException ex) { statsRecorder.recordStats(ex); throw ex; } catch (Exception ex) { statsRecorder.recordStats(ex); ReflectionUtils.rethrowRuntimeException(ex); } return null; }
4》. 继续调用到： org.springframework.cloud.client.loadbalancer.LoadBalancerRequestFactory#createRequest
public LoadBalancerRequest<ClientHttpResponse> createRequest( final HttpRequest request, final byte[] body, final ClientHttpRequestExecution execution) { return instance -> { HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance, this.loadBalancer); if (this.transformers != null) { for (LoadBalancerRequestTransformer transformer : this.transformers) { serviceRequest = transformer.transformRequest(serviceRequest, instance); } } return execution.execute(serviceRequest, body); }; }
5》 调用到org.springframework.http.client.InterceptingClientHttpRequest.InterceptingRequestExecution#execute 开始继续上面流程，直到拦截器走完会走原来调用 spring.web.http.client 的相关API发送请求。
2. Ribbon
Ribbon是Netflix Ribbon实现的一套客户端负载均衡的工具。主要提供客户端的软件负载均衡和服务调用。核心是提供负载均衡能力， 也就是有从注册中心获取服务信息、然后根据服务ServiceId 进行负载均衡的能力。
Ribbon客户端提供一系列完善的配置项如连接超时，重试等。简单的说，就是在配置文件中列出LoadBalancer(简称LB)后面所有的机器，Ribbon会基于某种规则(轮询、随机连接)等去连接这些机器。也可以实现自定义的负载均衡算法。
Ribbon 一个重要的类是com.netflix.loadbalancer.BaseLoadBalancer。 这个类里面维护了相关线上的Server 对象， 进行负载均衡的时候从这里获取到服务用自己的负载均衡算法进行选择服务后继续后面的操作。
所以Ribbon 相当于是在原来RestTemplate 的功能上加了个Inteceptor， 使其具有负载均衡以及根据服务名称进行调用的能力。最终根据选择的服务，替换掉服务名称之后交给原来的spring.web.http.client 封装的相关API进行调用。
1. Ribbon 默认也是从注册中心 获取的服务，并且通过自己的定时任务以及心跳检测来判断服务的状态。
1》 在创建com.netflix.loadbalancer.ZoneAwareLoadBalancer#ZoneAwareLoadBalancer(com.netflix.client.config.IClientConfig, com.netflix.loadbalancer.IRule, com.netflix.loadbalancer.IPing, com.netflix.loadbalancer.ServerList<T>, com.netflix.loadbalancer.ServerListFilter<T>, com.netflix.loadbalancer.ServerListUpdater) 对象的时候会从Eureka 获取。其调用链如下：
最终到： com.netflix.niws.loadbalancer.DiscoveryEnabledNIWSServerList#obtainServersViaDiscovery 获取服务
private List<DiscoveryEnabledServer> obtainServersViaDiscovery() { List<DiscoveryEnabledServer> serverList = new ArrayList<DiscoveryEnabledServer>(); if (eurekaClientProvider == null || eurekaClientProvider.get() == null) { logger.warn("EurekaClient has not been initialized yet, returning an empty list"); return new ArrayList<DiscoveryEnabledServer>(); } EurekaClient eurekaClient = eurekaClientProvider.get(); if (vipAddresses!=null){ for (String vipAddress : vipAddresses.split(",")) { // if targetRegion is null, it will be interpreted as the same region of client List<InstanceInfo> listOfInstanceInfo = eurekaClient.getInstancesByVipAddress(vipAddress, isSecure, targetRegion); for (InstanceInfo ii : listOfInstanceInfo) { if (ii.getStatus().equals(InstanceStatus.UP)) { if(shouldUseOverridePort){ if(logger.isDebugEnabled()){ logger.debug("Overriding port on client name: " + clientName + " to " + overridePort); } // copy is necessary since the InstanceInfo builder just uses the original reference, // and we don't want to corrupt the global eureka copy of the object which may be // used by other clients in our system InstanceInfo copy = new InstanceInfo(ii); if(isSecure){ ii = new InstanceInfo.Builder(copy).setSecurePort(overridePort).build(); }else{ ii = new InstanceInfo.Builder(copy).setPort(overridePort).build(); } } DiscoveryEnabledServer des = createServer(ii, isSecure, shouldUseIpAddr); serverList.add(des); } } if (serverList.size()>0 && prioritizeVipAddressBasedServers){ break; // if the current vipAddress has servers, we dont use subsequent vipAddress based servers } } } return serverList; }
2》 同时，在一个定时任务中更新服务列表与检测服务状态：
com.netflix.loadbalancer.PollingServerListUpdater#start 开启定时任务更新内部Server
@Override public synchronized void start(final UpdateAction updateAction) { if (isActive.compareAndSet(false, true)) { final Runnable wrapperRunnable = new Runnable() { @Override public void run() { if (!isActive.get()) { if (scheduledFuture != null) { scheduledFuture.cancel(true); } return; } try { updateAction.doUpdate(); lastUpdated = System.currentTimeMillis(); } catch (Exception e) { logger.warn("Failed one update cycle", e); } } }; scheduledFuture = getRefreshExecutor().scheduleWithFixedDelay( wrapperRunnable, initialDelayMs, refreshIntervalMs, TimeUnit.MILLISECONDS ); } else { logger.info("Already active, no-op"); } }
UpdateAction 如下： com.netflix.loadbalancer.DynamicServerListLoadBalancer#DynamicServerListLoadBalancer(com.netflix.client.config.IClientConfig)
public DynamicServerListLoadBalancer(IClientConfig clientConfig) { this.isSecure = false; this.useTunnel = false; this.serverListUpdateInProgress = new AtomicBoolean(false); class NamelessClass_1 implements UpdateAction { NamelessClass_1() { } public void doUpdate() { DynamicServerListLoadBalancer.this.updateListOfServers(); } } this.updateAction = new NamelessClass_1(); this.initWithNiwsConfig(clientConfig); }
调用到： com.netflix.loadbalancer.DynamicServerListLoadBalancer#updateListOfServers
public void updateListOfServers() { List<T> servers = new ArrayList(); if (this.serverListImpl != null) { servers = this.serverListImpl.getUpdatedListOfServers(); LOGGER.debug("List of Servers for {} obtained from Discovery client: {}", this.getIdentifier(), servers); if (this.filter != null) { servers = this.filter.getFilteredListOfServers((List)servers); LOGGER.debug("Filtered List of Servers for {} obtained from Discovery client: {}", this.getIdentifier(), servers); } } this.updateAllServerList((List)servers); }
这里的 serverListImpl 也是个策略模式，适配不同的注册中心：
调用到：org.springframework.cloud.netflix.ribbon.eureka.DomainExtractingServerList#getUpdatedListOfServers
@Override public List<DiscoveryEnabledServer> getUpdatedListOfServers() { List<DiscoveryEnabledServer> servers = setZones( this.list.getUpdatedListOfServers()); return servers; }
调用到： com.netflix.niws.loadbalancer.DiscoveryEnabledNIWSServerList#getUpdatedListOfServers
public List<DiscoveryEnabledServer> getUpdatedListOfServers(){ return obtainServersViaDiscovery(); } private List<DiscoveryEnabledServer> obtainServersViaDiscovery() { List<DiscoveryEnabledServer> serverList = new ArrayList<DiscoveryEnabledServer>(); if (eurekaClientProvider == null || eurekaClientProvider.get() == null) { logger.warn("EurekaClient has not been initialized yet, returning an empty list"); return new ArrayList<DiscoveryEnabledServer>(); } EurekaClient eurekaClient = eurekaClientProvider.get(); if (vipAddresses!=null){ for (String vipAddress : vipAddresses.split(",")) { // if targetRegion is null, it will be interpreted as the same region of client List<InstanceInfo> listOfInstanceInfo = eurekaClient.getInstancesByVipAddress(vipAddress, isSecure, targetRegion); for (InstanceInfo ii : listOfInstanceInfo) { if (ii.getStatus().equals(InstanceStatus.UP)) { if(shouldUseOverridePort){ if(logger.isDebugEnabled()){ logger.debug("Overriding port on client name: " + clientName + " to " + overridePort); } // copy is necessary since the InstanceInfo builder just uses the original reference, // and we don't want to corrupt the global eureka copy of the object which may be // used by other clients in our system InstanceInfo copy = new InstanceInfo(ii); if(isSecure){ ii = new InstanceInfo.Builder(copy).setSecurePort(overridePort).build(); }else{ ii = new InstanceInfo.Builder(copy).setPort(overridePort).build(); } } DiscoveryEnabledServer des = createServer(ii, isSecure, shouldUseIpAddr); serverList.add(des); } } if (serverList.size()>0 && prioritizeVipAddressBasedServers){ break; // if the current vipAddress has servers, we dont use subsequent vipAddress based servers } } } return serverList; }
3. OpenFeign
OpenFeign 我理解相当于一个RPC 远程调用的组件， 只需要声明接口，然后通过生成JDK代理对象即可用接口发起http 的形式发起调用。
原理参考：https://www.cnblogs.com/qlqwjy/p/14568086.html
OpenFeign 可以结合负载均衡使用，也可以单独作为远程调用工具进行调用。单独使用的时候时可以使用自己的Client(简单的封装了下HttpURLConnection)， 也可以使用一些第三方连接池作为HttpURLConnection 调用的组件。 如果使用服务名称进行调用，可以使用ribbon提供的负载均衡能力进行负载均衡以及超时控制等操作。
1. 不使用负载均衡
1. 接口
package cn.qz.cloud.service; import org.springframework.cloud.openfeign.FeignClient; import org.springframework.web.bind.annotation.GetMapping; @FeignClient(url = "http://news.baidu.com/", name = "TestFeignService") public interface TestFeignService { @GetMapping("guonei") String guoji(); }
2. 测试：
@Autowired private TestFeignService testFeignService; @GetMapping("guoji") public String test2() { return testFeignService.guoji(); }
3. 断点查看： 断点打到java.net.HttpURLConnection#HttpURLConnection， 查看调用链
可以看到最后也是走的HttpURLConnection， 只不过在前面进行了一系列的代理操作。 重要方法如下：
1》 feign.ReflectiveFeign.FeignInvocationHandler#invoke feign 生成的代理接口的调用入口
static class FeignInvocationHandler implements InvocationHandler { private final Target target; private final Map<Method, MethodHandler> dispatch; FeignInvocationHandler(Target target, Map<Method, MethodHandler> dispatch) { this.target = (Target)Util.checkNotNull(target, "target", new Object[0]); this.dispatch = (Map)Util.checkNotNull(dispatch, "dispatch for %s", new Object[]{target}); } 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; } } } public boolean equals(Object obj) { if (obj instanceof ReflectiveFeign.FeignInvocationHandler) { ReflectiveFeign.FeignInvocationHandler other = (ReflectiveFeign.FeignInvocationHandler)obj; return this.target.equals(other.target); } else { return false; } } public int hashCode() { return this.target.hashCode(); } public String toString() { return this.target.toString(); } }
2》 继续调用到：feign.SynchronousMethodHandler#invoke
// // Source code recreated from a .class file by IntelliJ IDEA // (powered by FernFlower decompiler) // package feign; import feign.InvocationHandlerFactory.MethodHandler; import feign.Logger.Level; import feign.Request.Options; import feign.codec.DecodeException; import feign.codec.Decoder; import feign.codec.ErrorDecoder; import java.io.IOException; import java.util.Iterator; import java.util.List; import java.util.concurrent.TimeUnit; import java.util.stream.Stream; final class SynchronousMethodHandler implements MethodHandler { private static final long MAX_RESPONSE_BUFFER_SIZE = 8192L; private final MethodMetadata metadata; private final Target<?> target; private final Client client; private final Retryer retryer; private final List<RequestInterceptor> requestInterceptors; private final Logger logger; private final Level logLevel; private final feign.RequestTemplate.Factory buildTemplateFromArgs; private final Options options; private final Decoder decoder; private final ErrorDecoder errorDecoder; private final boolean decode404; private final boolean closeAfterDecode; private final ExceptionPropagationPolicy propagationPolicy; private SynchronousMethodHandler(Target<?> target, Client client, Retryer retryer, List<RequestInterceptor> requestInterceptors, Logger logger, Level logLevel, MethodMetadata metadata, feign.RequestTemplate.Factory buildTemplateFromArgs, Options options, Decoder decoder, ErrorDecoder errorDecoder, boolean decode404, boolean closeAfterDecode, ExceptionPropagationPolicy propagationPolicy) { this.target = (Target)Util.checkNotNull(target, "target", new Object[0]); this.client = (Client)Util.checkNotNull(client, "client for %s", new Object[]{target}); this.retryer = (Retryer)Util.checkNotNull(retryer, "retryer for %s", new Object[]{target}); this.requestInterceptors = (List)Util.checkNotNull(requestInterceptors, "requestInterceptors for %s", new Object[]{target}); this.logger = (Logger)Util.checkNotNull(logger, "logger for %s", new Object[]{target}); this.logLevel = (Level)Util.checkNotNull(logLevel, "logLevel for %s", new Object[]{target}); this.metadata = (MethodMetadata)Util.checkNotNull(metadata, "metadata for %s", new Object[]{target}); this.buildTemplateFromArgs = (feign.RequestTemplate.Factory)Util.checkNotNull(buildTemplateFromArgs, "metadata for %s", new Object[]{target}); this.options = (Options)Util.checkNotNull(options, "options for %s", new Object[]{target}); this.errorDecoder = (ErrorDecoder)Util.checkNotNull(errorDecoder, "errorDecoder for %s", new Object[]{target}); this.decoder = (Decoder)Util.checkNotNull(decoder, "decoder for %s", new Object[]{target}); this.decode404 = decode404; this.closeAfterDecode = closeAfterDecode; this.propagationPolicy = propagationPolicy; } public Object invoke(Object[] argv) throws Throwable { RequestTemplate template = this.buildTemplateFromArgs.create(argv); Options options = this.findOptions(argv); Retryer retryer = this.retryer.clone(); while(true) { try { 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); } } } } Object executeAndDecode(RequestTemplate template, Options options) throws Throwable { Request request = this.targetRequest(template); if (this.logLevel != Level.NONE) { this.logger.logRequest(this.metadata.configKey(), this.logLevel, request); } long start = System.nanoTime(); Response response; try { response = this.client.execute(request, options); } catch (IOException var16) { if (this.logLevel != Level.NONE) { this.logger.logIOException(this.metadata.configKey(), this.logLevel, var16, this.elapsedTime(start)); } throw FeignException.errorExecuting(request, var16); } long elapsedTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); boolean shouldClose = true; try { if (this.logLevel != Level.NONE) { response = this.logger.logAndRebufferResponse(this.metadata.configKey(), this.logLevel, response, elapsedTime); } if (Response.class == this.metadata.returnType()) { Response var19; if (response.body() == null) { var19 = response; return var19; } else if (response.body().length() != null && (long)response.body().length() <= 8192L) { byte[] bodyData = Util.toByteArray(response.body().asInputStream()); Response var21 = response.toBuilder().body(bodyData).build(); return var21; } else { shouldClose = false; var19 = response; return var19; } } else { Object result; Object var11; if (response.status() >= 200 && response.status() < 300) { if (Void.TYPE != this.metadata.returnType()) { result = this.decode(response); shouldClose = this.closeAfterDecode; var11 = result; return var11; } else { result = null; return result; } } else if (this.decode404 && response.status() == 404 && Void.TYPE != this.metadata.returnType()) { result = this.decode(response); shouldClose = this.closeAfterDecode; var11 = result; return var11; } else { throw this.errorDecoder.decode(this.metadata.configKey(), response); } } } catch (IOException var17) { if (this.logLevel != Level.NONE) { this.logger.logIOException(this.metadata.configKey(), this.logLevel, var17, elapsedTime); } throw FeignException.errorReading(request, response, var17); } finally { if (shouldClose) { Util.ensureClosed(response.body()); } } } long elapsedTime(long start) { return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); } Request targetRequest(RequestTemplate template) { Iterator var2 = this.requestInterceptors.iterator(); while(var2.hasNext()) { RequestInterceptor interceptor = (RequestInterceptor)var2.next(); interceptor.apply(template); } return this.target.apply(template); } Object decode(Response response) throws Throwable { try { return this.decoder.decode(response, this.metadata.returnType()); } catch (FeignException var3) { throw var3; } catch (RuntimeException var4) { throw new DecodeException(response.status(), var4.getMessage(), response.request(), var4); } } Options findOptions(Object[] argv) { return argv != null && argv.length != 0 ? (Options)Stream.of(argv).filter((o) -> { return o instanceof Options; }).findFirst().orElse(this.options) : this.options; } static class Factory { private final Client client; private final Retryer retryer; private final List<RequestInterceptor> requestInterceptors; private final Logger logger; private final Level logLevel; private final boolean decode404; private final boolean closeAfterDecode; private final ExceptionPropagationPolicy propagationPolicy; Factory(Client client, Retryer retryer, List<RequestInterceptor> requestInterceptors, Logger logger, Level logLevel, boolean decode404, boolean closeAfterDecode, ExceptionPropagationPolicy propagationPolicy) { this.client = (Client)Util.checkNotNull(client, "client", new Object[0]); this.retryer = (Retryer)Util.checkNotNull(retryer, "retryer", new Object[0]); this.requestInterceptors = (List)Util.checkNotNull(requestInterceptors, "requestInterceptors", new Object[0]); this.logger = (Logger)Util.checkNotNull(logger, "logger", new Object[0]); this.logLevel = (Level)Util.checkNotNull(logLevel, "logLevel", new Object[0]); this.decode404 = decode404; this.closeAfterDecode = closeAfterDecode; this.propagationPolicy = propagationPolicy; } public MethodHandler create(Target<?> target, MethodMetadata md, feign.RequestTemplate.Factory buildTemplateFromArgs, Options options, Decoder decoder, ErrorDecoder errorDecoder) { return new SynchronousMethodHandler(target, this.client, this.retryer, this.requestInterceptors, this.logger, this.logLevel, md, buildTemplateFromArgs, options, decoder, errorDecoder, this.decode404, this.closeAfterDecode, this.propagationPolicy); } } }
3》 feign.Client.Default#execute
class Default implements Client { private final SSLSocketFactory sslContextFactory; private final HostnameVerifier hostnameVerifier; /** * Null parameters imply platform defaults. */ public Default(SSLSocketFactory sslContextFactory, HostnameVerifier hostnameVerifier) { this.sslContextFactory = sslContextFactory; this.hostnameVerifier = hostnameVerifier; } @Override public Response execute(Request request, Options options) throws IOException { HttpURLConnection connection = convertAndSend(request, options); return convertResponse(connection, request); } Response convertResponse(HttpURLConnection connection, Request request) throws IOException { int status = connection.getResponseCode(); String reason = connection.getResponseMessage(); if (status < 0) { throw new IOException(format("Invalid status(%s) executing %s %s", status, connection.getRequestMethod(), connection.getURL())); } Map<String, Collection<String>> headers = new LinkedHashMap<>(); for (Map.Entry<String, List<String>> field : connection.getHeaderFields().entrySet()) { // response message if (field.getKey() != null) { headers.put(field.getKey(), field.getValue()); } } Integer length = connection.getContentLength(); if (length == -1) { length = null; } InputStream stream; if (status >= 400) { stream = connection.getErrorStream(); } else { stream = connection.getInputStream(); } return Response.builder() .status(status) .reason(reason) .headers(headers) .request(request) .body(stream, length) .build(); } public HttpURLConnection getConnection(final URL url) throws IOException { return (HttpURLConnection) url.openConnection(); } HttpURLConnection convertAndSend(Request request, Options options) throws IOException { final URL url = new URL(request.url()); final HttpURLConnection connection = this.getConnection(url); if (connection instanceof HttpsURLConnection) { HttpsURLConnection sslCon = (HttpsURLConnection) connection; if (sslContextFactory != null) { sslCon.setSSLSocketFactory(sslContextFactory); } if (hostnameVerifier != null) { sslCon.setHostnameVerifier(hostnameVerifier); } } connection.setConnectTimeout(options.connectTimeoutMillis()); connection.setReadTimeout(options.readTimeoutMillis()); connection.setAllowUserInteraction(false); connection.setInstanceFollowRedirects(options.isFollowRedirects()); connection.setRequestMethod(request.httpMethod().name()); Collection<String> contentEncodingValues = request.headers().get(CONTENT_ENCODING); boolean gzipEncodedRequest = contentEncodingValues != null && contentEncodingValues.contains(ENCODING_GZIP); boolean deflateEncodedRequest = contentEncodingValues != null && contentEncodingValues.contains(ENCODING_DEFLATE); boolean hasAcceptHeader = false; Integer contentLength = null; for (String field : request.headers().keySet()) { if (field.equalsIgnoreCase("Accept")) { hasAcceptHeader = true; } for (String value : request.headers().get(field)) { if (field.equals(CONTENT_LENGTH)) { if (!gzipEncodedRequest && !deflateEncodedRequest) { contentLength = Integer.valueOf(value); connection.addRequestProperty(field, value); } } else { connection.addRequestProperty(field, value); } } } // Some servers choke on the default accept string. if (!hasAcceptHeader) { connection.addRequestProperty("Accept", "*/*"); } if (request.requestBody().asBytes() != null) { if (contentLength != null) { connection.setFixedLengthStreamingMode(contentLength); } else { connection.setChunkedStreamingMode(8196); } connection.setDoOutput(true); OutputStream out = connection.getOutputStream(); if (gzipEncodedRequest) { out = new GZIPOutputStream(out); } else if (deflateEncodedRequest) { out = new DeflaterOutputStream(out); } try { out.write(request.requestBody().asBytes()); } finally { try { out.close(); } catch (IOException suppressed) { // NOPMD } } } return connection; } }
可以看到是自己包装了一个简单的Client， 然后发起HttpURLConnection 进行调用。
2. 使用服务名-增加负载均衡
1. 接口
package cn.qz.cloud.service; import cn.qz.cloud.utils.JSONResultUtil; import org.springframework.cloud.openfeign.FeignClient; import org.springframework.web.bind.annotation.*; import java.util.List; import java.util.Map; @FeignClient(value = "CLOUD-PAYMENT-SERVICE") public interface PaymentFeignService { @GetMapping("/pay/getServerPort") JSONResultUtil<String> getServerPort(); }
2. 测试
@GetMapping("/pay/getServerPort")
JSONResultUtil<String> getServerPort() {
return paymentFeignService.getServerPort();
}
3. 查看调用链：
<init>:322, HttpURLConnection (java.net) <init>:863, HttpURLConnection (sun.net.www.protocol.http) openConnection:62, Handler (sun.net.www.protocol.http) openConnection:57, Handler (sun.net.www.protocol.http) openConnection:1026, URL (java.net) getConnection:118, Client$Default (feign) convertAndSend:123, Client$Default (feign) execute:77, Client$Default (feign) execute:100, TracingFeignClient (org.springframework.cloud.sleuth.instrument.web.client.feign) execute:60, LazyTracingFeignClient (org.springframework.cloud.sleuth.instrument.web.client.feign) execute:93, FeignLoadBalancer (org.springframework.cloud.openfeign.ribbon) execute:56, FeignLoadBalancer (org.springframework.cloud.openfeign.ribbon) call:104, AbstractLoadBalancerAwareClient$1 (com.netflix.client) call:303, LoadBalancerCommand$3$1 (com.netflix.loadbalancer.reactive) call:287, LoadBalancerCommand$3$1 (com.netflix.loadbalancer.reactive) call:231, ScalarSynchronousObservable$3 (rx.internal.util) call:228, ScalarSynchronousObservable$3 (rx.internal.util) unsafeSubscribe:10327, Observable (rx) drain:286, OnSubscribeConcatMap$ConcatMapSubscriber (rx.internal.operators) onNext:144, OnSubscribeConcatMap$ConcatMapSubscriber (rx.internal.operators) call:185, LoadBalancerCommand$1 (com.netflix.loadbalancer.reactive) call:180, LoadBalancerCommand$1 (com.netflix.loadbalancer.reactive) unsafeSubscribe:10327, Observable (rx) call:94, OnSubscribeConcatMap (rx.internal.operators) call:42, OnSubscribeConcatMap (rx.internal.operators) unsafeSubscribe:10327, Observable (rx) call:127, OperatorRetryWithPredicate$SourceSubscriber$1 (rx.internal.operators) enqueue:73, TrampolineScheduler$InnerCurrentThreadScheduler (rx.internal.schedulers) schedule:52, TrampolineScheduler$InnerCurrentThreadScheduler (rx.internal.schedulers) onNext:79, OperatorRetryWithPredicate$SourceSubscriber (rx.internal.operators) onNext:45, OperatorRetryWithPredicate$SourceSubscriber (rx.internal.operators) request:276, ScalarSynchronousObservable$WeakSingleProducer (rx.internal.util) setProducer:209, Subscriber (rx) call:138, ScalarSynchronousObservable$JustOnSubscribe (rx.internal.util) call:129, ScalarSynchronousObservable$JustOnSubscribe (rx.internal.util) call:48, OnSubscribeLift (rx.internal.operators) call:30, OnSubscribeLift (rx.internal.operators) call:48, OnSubscribeLift (rx.internal.operators) call:30, OnSubscribeLift (rx.internal.operators) call:48, OnSubscribeLift (rx.internal.operators) call:30, OnSubscribeLift (rx.internal.operators) subscribe:10423, Observable (rx) subscribe:10390, Observable (rx) blockForSingle:443, BlockingObservable (rx.observables) single:340, BlockingObservable (rx.observables) executeWithLoadBalancer:112, AbstractLoadBalancerAwareClient (com.netflix.client) execute:83, LoadBalancerFeignClient (org.springframework.cloud.openfeign.ribbon) execute:71, TraceLoadBalancerFeignClient (org.springframework.cloud.sleuth.instrument.web.client.feign) executeAndDecode:110, SynchronousMethodHandler (feign) invoke:80, SynchronousMethodHandler (feign) invoke:103, ReflectiveFeign$FeignInvocationHandler (feign) getServerPort:-1, $Proxy133 (com.sun.proxy) getServerPort:38, OrderFeignController (cn.qz.cloud.controller) ...
从调用链看到中间经过ribbon 相关， 进行负载均衡后再交给Feign的Client， 然后使用Java 原生的ttpURLConnection 进行调用。
3. OpenFeign 整合HttpClient 使用连接池
1. pom 增加如下配置：
<!-- 使用Apache HttpClient替换Feign原生httpURLConnection -->
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
</dependency>
<dependency>
<groupId>io.github.openfeign</groupId>
<artifactId>feign-httpclient</artifactId>
<version>10.4.0</version>
</dependency>
2. yml 增加
feign: httpclient: enabled: true
3.依赖关系查看
4. 调用链查看： (第一次会建立连接，后面会复用连接， 也就是池子化)
核心是： feign.httpclient.ApacheHttpClient 替换了原来的Feign 默认的Default 实现类。 源码如下：
package feign.httpclient; import org.apache.http.Header; import org.apache.http.HttpEntity; import org.apache.http.HttpResponse; import org.apache.http.NameValuePair; import org.apache.http.StatusLine; import org.apache.http.client.HttpClient; import org.apache.http.client.config.RequestConfig; import org.apache.http.client.methods.Configurable; import org.apache.http.client.methods.HttpUriRequest; import org.apache.http.client.methods.RequestBuilder; import org.apache.http.client.utils.URIBuilder; import org.apache.http.client.utils.URLEncodedUtils; import org.apache.http.entity.ByteArrayEntity; import org.apache.http.entity.ContentType; import org.apache.http.entity.StringEntity; import org.apache.http.impl.client.HttpClientBuilder; import org.apache.http.util.EntityUtils; import java.io.IOException; import java.io.InputStream; import java.io.InputStreamReader; import java.io.Reader; import java.nio.charset.Charset; import java.io.UnsupportedEncodingException; import java.net.MalformedURLException; import java.net.URI; import java.net.URISyntaxException; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; import feign.Client; import feign.Request; import feign.Response; import feign.Util; import static feign.Util.UTF_8; /** * This module directs Feign's http requests to Apache's * <a href="https://hc.apache.org/httpcomponents-client-ga/">HttpClient</a>. Ex. * * <pre> * GitHub github = Feign.builder().client(new ApacheHttpClient()).target(GitHub.class, * "https://api.github.com"); */ /* * Based on Square, Inc's Retrofit ApacheClient implementation */ public final class ApacheHttpClient implements Client { private static final String ACCEPT_HEADER_NAME = "Accept"; private final HttpClient client; public ApacheHttpClient() { this(HttpClientBuilder.create().build()); } public ApacheHttpClient(HttpClient client) { this.client = client; } @Override public Response execute(Request request, Request.Options options) throws IOException { HttpUriRequest httpUriRequest; try { httpUriRequest = toHttpUriRequest(request, options); } catch (URISyntaxException e) { throw new IOException("URL '" + request.url() + "' couldn't be parsed into a URI", e); } HttpResponse httpResponse = client.execute(httpUriRequest); return toFeignResponse(httpResponse, request); } HttpUriRequest toHttpUriRequest(Request request, Request.Options options) throws URISyntaxException { RequestBuilder requestBuilder = RequestBuilder.create(request.httpMethod().name()); // per request timeouts RequestConfig requestConfig = (client instanceof Configurable ? RequestConfig.copy(((Configurable) client).getConfig()) : RequestConfig.custom()) .setConnectTimeout(options.connectTimeoutMillis()) .setSocketTimeout(options.readTimeoutMillis()) .build(); requestBuilder.setConfig(requestConfig); URI uri = new URIBuilder(request.url()).build(); requestBuilder.setUri(uri.getScheme() + "://" + uri.getAuthority() + uri.getRawPath()); // request query params List<NameValuePair> queryParams = URLEncodedUtils.parse(uri, requestBuilder.getCharset().name()); for (NameValuePair queryParam : queryParams) { requestBuilder.addParameter(queryParam); } // request headers boolean hasAcceptHeader = false; for (Map.Entry<String, Collection<String>> headerEntry : request.headers().entrySet()) { String headerName = headerEntry.getKey(); if (headerName.equalsIgnoreCase(ACCEPT_HEADER_NAME)) { hasAcceptHeader = true; } if (headerName.equalsIgnoreCase(Util.CONTENT_LENGTH)) { // The 'Content-Length' header is always set by the Apache client and it // doesn't like us to set it as well. continue; } for (String headerValue : headerEntry.getValue()) { requestBuilder.addHeader(headerName, headerValue); } } // some servers choke on the default accept string, so we'll set it to anything if (!hasAcceptHeader) { requestBuilder.addHeader(ACCEPT_HEADER_NAME, "*/*"); } // request body if (request.requestBody().asBytes() != null) { HttpEntity entity = null; if (request.charset() != null) { ContentType contentType = getContentType(request); String content = new String(request.requestBody().asBytes(), request.charset()); entity = new StringEntity(content, contentType); } else { entity = new ByteArrayEntity(request.requestBody().asBytes()); } requestBuilder.setEntity(entity); } else { requestBuilder.setEntity(new ByteArrayEntity(new byte[0])); } return requestBuilder.build(); } private ContentType getContentType(Request request) { ContentType contentType = null; for (Map.Entry<String, Collection<String>> entry : request.headers().entrySet()) if (entry.getKey().equalsIgnoreCase("Content-Type")) { Collection<String> values = entry.getValue(); if (values != null && !values.isEmpty()) { contentType = ContentType.parse(values.iterator().next()); if (contentType.getCharset() == null) { contentType = contentType.withCharset(request.charset()); } break; } } return contentType; } Response toFeignResponse(HttpResponse httpResponse, Request request) throws IOException { StatusLine statusLine = httpResponse.getStatusLine(); int statusCode = statusLine.getStatusCode(); String reason = statusLine.getReasonPhrase(); Map<String, Collection<String>> headers = new HashMap<String, Collection<String>>(); for (Header header : httpResponse.getAllHeaders()) { String name = header.getName(); String value = header.getValue(); Collection<String> headerValues = headers.get(name); if (headerValues == null) { headerValues = new ArrayList<String>(); headers.put(name, headerValues); } headerValues.add(value); } return Response.builder() .status(statusCode) .reason(reason) .headers(headers) .request(request) .body(toFeignBody(httpResponse)) .build(); } Response.Body toFeignBody(HttpResponse httpResponse) { final HttpEntity entity = httpResponse.getEntity(); if (entity == null) { return null; } return new Response.Body() { @Override public Integer length() { return entity.getContentLength() >= 0 && entity.getContentLength() <= Integer.MAX_VALUE ? (int) entity.getContentLength() : null; } @Override public boolean isRepeatable() { return entity.isRepeatable(); } @Override public InputStream asInputStream() throws IOException { return entity.getContent(); } @Override public Reader asReader() throws IOException { return new InputStreamReader(asInputStream(), UTF_8); } @Override public Reader asReader(Charset charset) throws IOException { Util.checkNotNull(charset, "charset should not be null"); return new InputStreamReader(asInputStream(), charset); } @Override public void close() throws IOException { EntityUtils.consume(entity); } }; } }
5. 查看替换源码： org.springframework.cloud.openfeign.ribbon.HttpClientFeignLoadBalancedConfiguration
package org.springframework.cloud.openfeign.ribbon; import feign.Client; import feign.httpclient.ApacheHttpClient; import org.apache.http.client.HttpClient; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.cloud.netflix.ribbon.SpringClientFactory; import org.springframework.cloud.openfeign.clientconfig.HttpClientFeignConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; /** * @author Spencer Gibb * @author Olga Maciaszek-Sharma */ @Configuration(proxyBeanMethods = false) @ConditionalOnClass(ApacheHttpClient.class) @ConditionalOnProperty(value = "feign.httpclient.enabled", matchIfMissing = true) @Import(HttpClientFeignConfiguration.class) class HttpClientFeignLoadBalancedConfiguration { @Bean @ConditionalOnMissingBean(Client.class) public Client feignClient(CachingSpringLoadBalancerFactory cachingFactory, SpringClientFactory clientFactory, HttpClient httpClient) { ApacheHttpClient delegate = new ApacheHttpClient(httpClient); return new LoadBalancerFeignClient(delegate, cachingFactory, clientFactory); } }
这里使用ApacheHttpClient， 然后达到可以复用连接的效果。