SpringCloud-原始碼分析 Ribbon
本文作者:陳剛,叩丁狼高階講師。原創文章,轉載請註明出處。
本文章會通過斷點跟蹤的方式來解讀 Ribbon 原始碼 ,希望同學們把文章看完之後不僅能夠了解 Ribbon的實現原理,還能掌握原始碼解讀的方式和技巧(重要)。
回顧
回顧一下我們的 Ribbon部分內容我們當時使用TestTemplate + LoadBalanced 做了這樣的一個案例
叩丁狼教育.png
當時我們在配置類中做了如下Bean的定義去開啟了RestTemplate的負載均衡功能
//通過@LoadBalanced註解表明這個restRemplate開啟負載均衡的功能。 //RestTemplate是spring內建的http請求封裝 @Bean @LoadBalanced RestTemplate restTemplate() { return new RestTemplate(); }
然後在Consumer中去請求Producer服務(當然會同時開啟多個Producer服務)
//訪問 PRODUCER 服務
String result =
restTemplate.getForObject("http://PRODUCER/provide?name="+name,String.class);
而我們需要達到的效果是該請求多次呼叫會從不同的Producer服務獲取到結果(根據負載均衡規則),然而我們發的請求始終只會有一個呀,那麼它要如何才能實現服務之間的切換呼叫呢?那麼猜想一下,Ribbon是不是需要需要先攔截到我們的請求,然後根據我們定義的負載均衡演算法,然後從服務清單中去選擇合適的服務例項,然後完成呼叫呢???(思考一下)
那麼接下來我們就來對這樣的一個請求進行原始碼追蹤分析。
@LoadBalanced
我們先來研究一下@LoadBalanced是一個什麼東西,檢視他的原始碼如下
/** * Annotation to mark a RestTemplate bean to be configured to use a LoadBalancerClient * @author Spencer Gibb */ @Target({ ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD }) @Retention(RetentionPolicy.RUNTIME) @Documented @Inherited @Qualifier public @interface LoadBalanced { }
註釋 “Annotation to mark a RestTemplate bean to be configured to use a LoadBalancerClient” 告訴我們:@LoadBalanced標籤是用來給RestTemplate標記,以使用LoadBalancerClient(負載均衡的客戶端)來配置它。
我們繼續追蹤 LoadBalancerClient的原始碼
/**
* Represents a client side load balancer
* @author Spencer Gibb
*/
public interface LoadBalancerClient extends ServiceInstanceChooser {
/**
* execute request using a ServiceInstance from the LoadBalancer for the specified
* service
* @param serviceId the service id to look up the LoadBalancer
* @param request allows implementations to execute pre and post actions such as
* incrementing metrics
* @return the result of the LoadBalancerRequest callback on the selected
* ServiceInstance
*/
<T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException;
/**
* execute request using a ServiceInstance from the LoadBalancer for the specified
* service
* @param serviceId the service id to look up the LoadBalancer
* @param serviceInstance the service to execute the request to
* @param request allows implementations to execute pre and post actions such as
* incrementing metrics
* @return the result of the LoadBalancerRequest callback on the selected
* ServiceInstance
*/
<T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException;
/**
* Create a proper URI with a real host and port for systems to utilize.
* Some systems use a URI with the logical serivce name as the host,
* such as http://myservice/path/to/service. This will replace the
* service name with the host:port from the ServiceInstance.
* @param instance
* @param original a URI with the host as a logical service name
* @return a reconstructed URI
*/
URI reconstructURI(ServiceInstance instance, URI original);
}
註釋“Represents a client side load balancer”表示該介面它是一個客戶端負載均衡器 ,提供了幾個方法,翻譯方法上的註釋得知他們的作用大致如下
execute : 使用LoadBalancer中的ServiceInstance為指定的服務執行請求,說白了就是通過它來實現服務的請求呼叫。
reconstructURI:使用真實主機和埠建立適當的URI以供系統使用,獲取要呼叫的服務的主機和埠
並且該介面它繼承ServiceInstanceChooser介面
/**
* Implemented by classes which use a load balancer to choose a server to
* send a request to.
*
* @author Ryan Baxter
*/
public interface ServiceInstanceChooser {
/**
* Choose a ServiceInstance from the LoadBalancer for the specified service
* @param serviceId the service id to look up the LoadBalancer
* @return a ServiceInstance that matches the serviceId
*/
ServiceInstance choose(String serviceId);
}
翻譯介面上的註釋“ Implemented by classes which use a load balancer to choose a server to send a request to.”大致意思為: 使用負載均衡器選擇一個服務,然後去發起請求,而 choose方法的大致作用為:從LoadBalancer負載均衡器中為指定的服務(serviceId)選擇一個服務例項(ServiceInstance) ,其實到這裡我們大致已經明白了LoadBalancerClient的目的,就是通過choose方法選擇要呼叫的服務例項,通過reconstructURI獲取服務和主機和埠,然後通過execute執行服務的呼叫,而 RibbonLoadBalancerClient是對 LoadBalancerClient 的實現 ,他們的層級關係如下(idea中按ctrl+alt+u檢視):
image.png
那麼LoadBalancer到底是怎麼讓RestTtemplate 實現負載均衡的呢?要揭露這個答案我們必須得跟蹤RestTemplate的的請求,斷點一下RestTemplate服務呼叫的程式碼,然後去瀏覽器請求該方法觸發斷點,看看底層是如何實現呼叫的
@RequestMapping("/consumer")
public String consumer(@RequestParam("name") String name){
//訪問 PRODUCER 服務
String result = restTemplate.getForObject("http://PRODUCER/provide?name="+name,String.class);
return result;
}
跟蹤 getForObject 方法進入
public <T> T getForObject(String url, Class<T> responseType, Object... uriVariables) throws RestClientException {
RequestCallback requestCallback = acceptHeaderRequestCallback(responseType);
HttpMessageConverterExtractor<T> responseExtractor =
new HttpMessageConverterExtractor<>(responseType, getMessageConverters(), logger);
return execute(url, HttpMethod.GET, requestCallback, responseExtractor, uriVariables);
}
這裡繼續呼叫了RestTemplate.execute 方法,並且把呼叫的服務地址傳入進去,然後使用HttpMethod.GET方式進行呼叫,繼續跟蹤下去
protected <T> T doExecute(URI url, @Nullable HttpMethod method, @Nullable RequestCallback requestCallback,
@Nullable ResponseExtractor<T> responseExtractor) throws RestClientException {
Assert.notNull(url, "'url' must not be null");
Assert.notNull(method, "'method' must not be null");
ClientHttpResponse response = null;
try {
ClientHttpRequest request = createRequest(url, method);
if (requestCallback != null) {
requestCallback.doWithRequest(request);
}
response = request.execute();
handleResponse(url, method, response);
if (responseExtractor != null) {
return responseExtractor.extractData(response);
}
else {
return null;
}
}
...省略程式碼...
這裡把服務地址包裝成 URI物件 ,然後呼叫 ClientHttpRequest request = createRequest(url, method); 去建立http客戶端請求物件,然後呼叫 response = request.execute();執行請求,到這裡我們依然沒有找到他是如何實現負載均衡的。我們斷點跟蹤一下 createRequest方法最終會呼叫HttpAccessor#createRequest方法,HttpAccessor是個什麼東西呢,我們在RestTemplate類中按下 ctrl+alt+u檢視他們的層級關係如下
叩丁狼教育.png
請注意看,RestTemplate繼承了InterceptingHttpAccessor(看名字像http請求的訪問攔截的意思,後續再說) 而 InterceptingHttpAccessor又繼承了HttpAccessor,而createRequest方法如下
protected ClientHttpRequest createRequest(URI url, HttpMethod method) throws IOException {
ClientHttpRequest request = getRequestFactory().createRequest(url, method);
if (logger.isDebugEnabled()) {
logger.debug("Created " + method.name() + " request for \"" + url + "\"");
}
return request;
}
這裡呼叫了 getRequestFactory()方法獲取到一個請求工廠,然後去建立 ClientHttpRequest ,而getRequestFactory()方法是在InterceptingHttpAccessor中進行復寫實現的,跟蹤interceptingHttpAccessor#getRequestFactory方法如下:
public ClientHttpRequestFactory getRequestFactory() {
List<ClientHttpRequestInterceptor> interceptors = getInterceptors();
if (!CollectionUtils.isEmpty(interceptors)) {
ClientHttpRequestFactory factory = this.interceptingRequestFactory;
if (factory == null) {
factory = new InterceptingClientHttpRequestFactory(super.getRequestFactory(), interceptors);
this.interceptingRequestFactory = factory;
}
return factory;
}
else {
return super.getRequestFactory();
}
}
/**
* Return the request interceptors that this accessor uses.
* <p>The returned {@link List} is active and may get appended to.
*/
public List<ClientHttpRequestInterceptor> getInterceptors() {
return this.interceptors;
}
可以看到這裡呼叫getInterceptors()方法在獲取 ClientHttpRequestInterceptor http客戶端請求攔截器 ,然後使用攔截器new 了一個 InterceptingClientHttpRequestFactory 工廠出來,他這裡用到了請求攔截器,是想幹嘛呢???(思考一下)
我們繼續往下執行結束完getRequestFactory() 方法我們跟蹤一下 getRequestFactory().createRequest(url, method); 的createRequest 方法
@Override
protected ClientHttpRequest createRequest(
URI uri, HttpMethod httpMethod, ClientHttpRequestFactory requestFactory) {
return new InterceptingClientHttpRequest(requestFactory, this.interceptors, uri, httpMethod);
}
沒有任何懸念,斷點跟蹤下去通過呼叫了InterceptingClientHttpRequestFactory#createRequest方法來建立 l了一個InterceptingClientHttpRequest物件,而InterceptingClientHttpRequest肯定是ClientHttpRequest的實現類
/**
* Wrapper for a {@link ClientHttpRequest} that has support for {@link ClientHttpRequestInterceptor}s.
*
* @author Arjen Poutsma
* @since 3.1
*/
class InterceptingClientHttpRequest extends AbstractBufferingClientHttpRequest {
private final ClientHttpRequestFactory requestFactory;
private final List<ClientHttpRequestInterceptor> interceptors;
private HttpMethod method;
private URI uri;
protected InterceptingClientHttpRequest(ClientHttpRequestFactory requestFactory,
List<ClientHttpRequestInterceptor> interceptors, URI uri, HttpMethod method) {
this.requestFactory = requestFactory;
this.interceptors = interceptors;
this.method = method;
this.uri = uri;
}
註釋:“Wrapper for a {@link ClientHttpRequest} that has support for {@link ClientHttpRequestInterceptor}s.” 告訴我們
InterceptingClientHttpRequest他是對ClientHttpRequest做了包裝,並且ClientHttpRequestInterceptor提供支援,對HttpMethod (請求方式)和 URI(服務地址)亦有描述 ,我們看一下他的繼承體系
ok,看到這裡我們知道 RestTemplate底層是通過 ClientHttpRequestFactory工廠建立了 InterceptingClientHttpRequest 擁有可攔截功能的http客戶端請求物件,那麼他是怎麼呼叫的呢??又是如何實現攔截的呢???
我們回到 org.springframework.web.client.RestTemplate#doExecute 方法中的 response = request.execute(); 程式碼繼續跟蹤,然後你就會發現,當執行request.execute方法的時候會跳轉到 org.springframework.cloud.client.loadbalancer.LoadBalancerInterceptor#intercept 攔截器方法中,看名字就能猜到這個攔截器跟負載均衡(LoadBalancer)有關,他是ClientHttpRequestInterceptor的實現,
/**
* @author Spencer Gibb
* @author Dave Syer
* @author Ryan Baxter
* @author William Tran
*/
public class LoadBalancerInterceptor implements ClientHttpRequestInterceptor {
private LoadBalancerClient loadBalancer;
private LoadBalancerRequestFactory requestFactory;
public LoadBalancerInterceptor(LoadBalancerClient loadBalancer, LoadBalancerRequestFactory requestFactory) {
this.loadBalancer = loadBalancer;
this.requestFactory = requestFactory;
}
public LoadBalancerInterceptor(LoadBalancerClient loadBalancer) {
// for backwards compatibility
this(loadBalancer, new LoadBalancerRequestFactory(loadBalancer));
}
@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, requestFactory.createRequest(request, body, execution));
}
}
繼續跟蹤一下 ClientHttpRequestInterceptor 介面
/**
* Intercepts client-side HTTP requests. Implementations of this interface can be
* {@linkplain org.springframework.web.client.RestTemplate#setInterceptors registered}
* with the {@link org.springframework.web.client.RestTemplate RestTemplate},
* as to modify the outgoing {@link ClientHttpRequest} and/or the incoming
* {@link ClientHttpResponse}.
*
* <p>The main entry point for interceptors is
* {@link #intercept(HttpRequest, byte[], ClientHttpRequestExecution)}.
*
* @author Arjen Poutsma
* @since 3.1
*/
@FunctionalInterface
public interface ClientHttpRequestInterceptor {
/**
* Intercept the given request, and return a response. The given
* {@link ClientHttpRequestExecution} allows the interceptor to pass on the
* request and response to the next entity in the chain.
* <p>A typical implementation of this method would follow the following pattern:
* <ol>
* <li>Examine the {@linkplain HttpRequest request} and body</li>
* <li>Optionally {@linkplain org.springframework.http.client.support.HttpRequestWrapper
* wrap} the request to filter HTTP attributes.</li>
* <li>Optionally modify the body of the request.</li>
* <li><strong>Either</strong>
* <ul>
* <li>execute the request using
* {@link ClientHttpRequestExecution#execute(org.springframework.http.HttpRequest, byte[])},</li>
* <strong>or</strong>
* <li>do not execute the request to block the execution altogether.</li>
* </ul>
* <li>Optionally wrap the response to filter HTTP attributes.</li>
* </ol>
* @param request the request, containing method, URI, and headers
* @param body the body of the request
* @param execution the request execution
* @return the response
* @throws IOException in case of I/O errors
*/
ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution)
throws IOException;
翻譯註釋“Intercepts client-side HTTP requests. Implementations of this interface can be” 我們知道 ClientHttpRequestInterceptor就是對http請求的攔截到這裡,我們的請求已經被攔截到,貌似離真相越來越近了,我們仔細分析一下 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, requestFactory.createRequest(request, body, execution));
}
這裡通過 request(之前建立的InterceptingClientHttpRequest) 物件獲取到URI呼叫的服務地址 ,然後獲取到 呼叫的服務名字 serviceName ,然後呼叫org.springframework.cloud.client.loadbalancer.LoadBalancerClient.execute去執行 ,LoadBalancerClient是什麼,不就是最開始我們分析的負載均衡器客戶端嘛???他使用的是RibbonLoadBalancerClient實現類
繼續跟蹤下去 RibbonLoadBalancerClient.execute
@Override
public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException {
ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
Server server = getServer(loadBalancer);
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);
}
ILoadBalancer loadBalancer = getLoadBalancer(serviceId);通過服務id獲取負載均衡器 ,跟蹤一下 ILoadBalancer原始碼
/**
* Interface that defines the operations for a software loadbalancer. A typical
* loadbalancer minimally need a set of servers to loadbalance for, a method to
* mark a particular server to be out of rotation and a call that will choose a
* server from the existing list of server.
*
* @author stonse
*
*/
public interface ILoadBalancer {
/**
* Initial list of servers.
* This API also serves to add additional ones at a later time
* The same logical server (host:port) could essentially be added multiple times
* (helpful in cases where you want to give more "weightage" perhaps ..)
*
* @param newServers new servers to add
*/
public void addServers(List<Server> newServers);
/**
* Choose a server from load balancer.
*
* @param key An object that the load balancer may use to determine which server to return. null if
* the load balancer does not use this parameter.
* @return server chosen
*/
public Server chooseServer(Object key);
/**
* To be called by the clients of the load balancer to notify that a Server is down
* else, the LB will think its still Alive until the next Ping cycle - potentially
* (assuming that the LB Impl does a ping)
*
* @param server Server to mark as down
*/
public void markServerDown(Server server);
/**
* @deprecated 2016-01-20 This method is deprecated in favor of the
* cleaner {@link #getReachableServers} (equivalent to availableOnly=true)
* and {@link #getAllServers} API (equivalent to availableOnly=false).
*
* Get the current list of servers.
*
* @param availableOnly if true, only live and available servers should be returned
*/
@Deprecated
public List<Server> getServerList(boolean availableOnly);
/**
* @return Only the servers that are up and reachable.
*/
public List<Server> getReachableServers();
/**
* @return All known servers, both reachable and unreachable.
*/
public List<Server> getAllServers();
}
翻譯註釋“Interface that defines the operations for a software loadbalancer.”它定義軟負載均衡器操作的介面規範
addServers:初始化服務列表
chooseServer:選擇一個服務
markServerDown:標記服務下線,從服務列表中移除
getReachableServers:獲取可訪問的服務
getAllServers:獲取所有的服務
而預設情況下這裡預設使用的是 ZoneAwareLoadBalancer,斷點跟蹤或者檢視RibbonClientConfiguration配置類的ribbonLoadBalancer方法得知
@Bean
@ConditionalOnMissingBean
public ILoadBalancer ribbonLoadBalancer(IClientConfig config,
ServerList<Server> serverList, ServerListFilter<Server> serverListFilter,
IRule rule, IPing ping, ServerListUpdater serverListUpdater) {
if (this.propertiesFactory.isSet(ILoadBalancer.class, name)) {
return this.propertiesFactory.get(ILoadBalancer.class, config, name);
}
return new ZoneAwareLoadBalancer<>(config, rule, ping, serverList,
serverListFilter, serverListUpdater);
}
那麼我們繼續跟蹤 RibbonLoadBalancerClient.execute中的 getServer(loadBalancer);程式碼這裡是在根據負載均衡器選擇將要執行呼叫的服務,跟蹤下去
com.netflix.loadbalancer.ZoneAwareLoadBalancer#chooseServer
@Override
public Server chooseServer(Object key) {
if (!ENABLED.get() || getLoadBalancerStats().getAvailableZones().size() <= 1) {
logger.debug("Zone aware logic disabled or there is only one zone");
return super.chooseServer(key);
}
Server server = null;
try {
LoadBalancerStats lbStats = getLoadBalancerStats();
Map<String, ZoneSnapshot> zoneSnapshot = ZoneAvoidanceRule.createSnapshot(lbStats);
logger.debug("Zone snapshots: {}", zoneSnapshot);
if (triggeringLoad == null) {
triggeringLoad = DynamicPropertyFactory.getInstance().getDoubleProperty(
"ZoneAwareNIWSDiscov
這裡的key接受了一個預設值"default",因為我們的可用的區域只有一個,所以這裡呼叫了super.chooseServer即com.netflix.loadbalancer.BaseLoadBalancer#chooseServer
public Server chooseServer(Object key) {
if (counter == null) {
counter = createCounter();
}
counter.increment();
if (rule == null) {
return null;
} else {
try {
return rule.choose(key);
} catch (Exception e) {
logger.warn("LoadBalancer [{}]: Error choosing server for key {}", name, key, e);
return null;
}
}
}
繼續跟蹤 rule.choose(key); 這裡呼叫了 com.netflix.loadbalancer.PredicateBasedRule#choose 的服務選擇方法,
/**
* A rule which delegates the server filtering logic to an instance of {@link AbstractServerPredicate}.
* After filtering, a server is returned from filtered list in a round robin fashion.
*
*
* @author awang
*
*/
public abstract class PredicateBasedRule extends ClientConfigEnabledRoundRobinRule {
/**
* Method that provides an instance of {@link AbstractServerPredicate} to be used by this class.
*
*/
public abstract AbstractServerPredicate getPredicate();
/**
* Get a server by calling {@link AbstractServerPredicate#chooseRandomlyAfterFiltering(java.util.List, Object)}.
* The performance for this method is O(n) where n is number of servers to be filtered.
*/
@Override
public Server choose(Object key) {
ILoadBalancer lb = getLoadBalancer();
Optional<Server> server = getPredicate().chooseRoundRobinAfterFiltering(lb.getAllServers(), key);
if (server.isPresent()) {
return server.get();
} else {
return null;
}
}
}
PredicateBasedRule本身是一個抽象策略,繼承自ClientConfigEnabledRoundRobinRule ,而ClientConfigEnabledRoundRobinRule是一個實現了輪詢策略的客戶端配置
/**
RoundRobinRule :輪詢
* This class essentially contains the RoundRobinRule class defined in the
* loadbalancer package
*
* @author stonse
*
*/
public class ClientConfigEnabledRoundRobinRule extends AbstractLoadBalancerRule {
//輪詢策略
RoundRobinRule roundRobinRule = new RoundRobinRule();
繼續跟蹤chooseRoundRobinAfterFiltering方法
/**
* Choose a server in a round robin fashion after the predicate filters a given list of servers and load balancer key.
*/
public Optional<Server> chooseRoundRobinAfterFiltering(List<Server> servers, Object loadBalancerKey) {
List<Server> eligible = getEligibleServers(servers, loadBalancerKey);
if (eligible.size() == 0) {
return Optional.absent();
}
return Optional.of(eligible.get(incrementAndGetModulo(eligible.size())));
}
翻譯方法註釋“ Choose a server in a round robin fashion after the predicate filters a given list of servers and load balancer key.”告訴我們這裡是把過濾後的服務進行輪詢選擇 ,再看程式碼 List<Server> eligible獲取到合格的伺服器 ,incrementAndGetModulo就是在以輪詢的方式獲取到服務的索引。
繼續往下回到 RibbonLoadBalancerClient#execute方法
@Override
public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException {
ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
Server server = getServer(loadBalancer);
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);
}
這裡拿到要呼叫的服務的例項的後,把服務名,服務例項等資訊包裝到 RibbonServer物件中
public static class RibbonServer implements ServiceInstance {
private final String serviceId;
private final Server server;
private final boolean secure;
private Map<String, String> metadata;
public RibbonServer(String serviceId, Server server) {
this(serviceId, server, false, Collections.<String, String> emptyMap());
}
public RibbonServer(String serviceId, Server server, boolean secure,
Map<String, String> metadata) {
this.serviceId = serviceId;
this.server = server;
this.secure = secure;
this.metadata = metadata;
}
.....
public interface ServiceInstance {
/**
* @return the service id as registered.
*/
String getServiceId();
/**
* @return the hostname of the registered ServiceInstance
*/
String getHost();
/**
* @return the port of the registered ServiceInstance
*/
int getPort();
/**
* @return if the port of the registered ServiceInstance is https or not
*/
boolean isSecure();
/**
* @return the service uri address
*/
URI getUri();
/**
* @return the key value pair metadata associated with the service instance
*/
Map<String, String> getMetadata();
/**
* @return the scheme of the instance
*/
default String getScheme() {
return null;
}
}
RibbonServer 實現了 ServiceInstance介面, ServiceInstance本身就是對服務例項的規範,有獲取服務id,主機,埠等方法。
然後繼續呼叫 excute方法執行請求
@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
......
看程式碼 T returnVal = request.apply(serviceInstance);
這裡最終會呼叫InterceptingClientHttpRequest.InterceptingRequestExecution的execute方法,傳入服務例項,執行請求,拿到返回結果。
private class InterceptingRequestExecution implements ClientHttpRequestExecution {
private final Iterator<ClientHttpRequestInterceptor> iterator;
public InterceptingRequestExecution() {
this.iterator = interceptors.iterator();
}
@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();
}
}
}
跟蹤 requestFactory.createRequest方法
@Override
public ClientHttpRequest createRequest(URI uri, HttpMethod httpMethod) throws IOException {
HttpURLConnection connection = openConnection(uri.toURL(), this.proxy);
prepareConnection(connection, httpMethod.name());
if (this.bufferRequestBody) {
return new SimpleBufferingClientHttpRequest(connection, this.outputStreaming);
}
else {
return new SimpleStreamingClientHttpRequest(connection, this.chunkSize, this.outputStreaming);
}
}
.....
protected HttpURLConnection openConnection(URL url, @Nullable Proxy proxy) throws IOException {
URLConnection urlConnection = (proxy != null ? url.openConnection(proxy) : url.openConnection());
if (!HttpURLConnection.class.isInstance(urlConnection)) {
throw new IllegalStateException("HttpURLConnection required for [" + url + "] but got: " + urlConnection);
}
return (HttpURLConnection) urlConnection;
}
到這裡我們看到他是通過 URLConnection 來呼叫遠端服務的。。。
好吧總結一下,這個遠端服務呼叫的背後到底做了哪些事情呢??
[email protected]開啟了RibbonLoadBalancerClient負載均衡支援
2.RestTemplate對服務的地址(Uri),主機(host),埠(port)等做了一些描述,然後建立了 InterceptingClientHttpRequest http請求的客戶端物件,用來執行請求用,
3.當呼叫RestTemplate發起請求時會被 LoadBalancerInterceptor請求攔截器給攔截到
4.攔截器中使用了 RibbonLoadBalancerClient執行請求,然後根據服務id獲取了負載均衡器,預設 ZoneAwareLoadBalancer
5.然後負載均衡器進行服務的選擇,預設使用了輪詢策略
6.拿到服務例項後呼叫 InterceptingClientHttpRequest 完成服務呼叫請求,獲取返回結果。
想獲取更多技術乾貨,請前往叩丁狼官網:http://www.wolfcode.cn/all_article.html