2. 程式人生 > >Ribbon之Loadbalance的原始碼分析

Ribbon之Loadbalance的原始碼分析

阿新 • • 發佈：2018-11-07

Load Balance負載均衡是用於解決一臺機器(一個程序)無法解決所有請求而產生的一種演算法。

像nginx可以使用負載均衡分配流量，ribbon為客戶端提供負載均衡，dubbo服務呼叫裡的負載均衡等等，很多地方都使用到了負載均衡。

使用負載均衡帶來的好處很明顯：

  1. 當叢集裡的1臺或者多臺伺服器down的時候，剩餘的沒有down的伺服器可以保證服務的繼續使用
  2. 使用了更多的機器保證了機器的良性使用，不會由於某一高峰時刻導致系統cpu急劇上升

負載均衡有好幾種實現策略，常見的有：

  1. 隨機 (Random)
  2. 輪詢 (RoundRobin)
  3. 一致性雜湊 (ConsistentHash)
  4. 雜湊 (Hash)
  5. 加權（Weighted）

我們以ribbon的實現為基礎，看看其中的一些演算法是如何實現的。

ribbon是一個為客戶端提供負載均衡功能的服務，它內部提供了一個叫做ILoadBalance的介面代表負載均衡器的操作，比如有新增伺服器操作、選擇伺服器操作、獲取所有的伺服器列表、獲取可用的伺服器列表等等。

還提供了一個叫做IRule的介面代表負載均衡策略：

複製程式碼

package com.netflix.loadbalancer;

public interface IRule{
    /*
     * choose one alive server from lb.allServers or
     * lb.upServers according to key
     * 
     * @return choosen Server object. NULL is returned if none
     *  server is available 
     */

    public Server choose(Object key);
    
    public void setLoadBalancer(ILoadBalancer lb);
    
    public ILoadBalancer getLoadBalancer();    
}

複製程式碼

IRule介面的實現類有以下幾種：

image

 

其中RandomRule表示隨機策略、RoundRobin表示輪詢策略、WeightedResponseTimeRule表示加權策略、BestAvailableRule表示請求數最少策略等等。

隨機策略很簡單，就是從伺服器中隨機選擇一個伺服器，RandomRule的實現程式碼如下：

複製程式碼

public Server choose(ILoadBalancer lb, Object key) {
    if (lb == null) {
        return null;
    }
    Server server = null;
 
    while (server == null) {
        if (Thread.interrupted()) {
            return null;
        }
        List<Server> upList = lb.getReachableServers();
        List<Server> allList = lb.getAllServers();
        int serverCount = allList.size();
        if (serverCount == 0) {
            return null;
        }
        int index = rand.nextInt(serverCount); // 使用jdk內部的Random類隨機獲取索引值index
        server = upList.get(index); // 得到伺服器例項
 
        if (server == null) {
            Thread.yield();
            continue;
        }
 
        if (server.isAlive()) {
            return (server);
        }
 
        server = null;
        Thread.yield();
    }
    return server;
}

複製程式碼

RoundRobin輪詢策略表示每次都取下一個伺服器，比如一共有5臺伺服器，第1次取第1臺，第2次取第2臺，第3次取第3臺，以此類推：

複製程式碼

public Server choose(ILoadBalancer lb, Object key) {
    if (lb == null) {
        log.warn("no load balancer");
        return null;
    }
 
    Server server = null;
    int count = 0;
    while (server == null && count++ < 10) { // retry 10 次
        List<Server> reachableServers = lb.getReachableServers();
        List<Server> allServers = lb.getAllServers();
        int upCount = reachableServers.size();
        int serverCount = allServers.size();
 
        if ((upCount == 0) || (serverCount == 0)) {
            log.warn("No up servers available from load balancer: " + lb);
            return null;
        }
 
        int nextServerIndex = incrementAndGetModulo(serverCount); // incrementAndGetModulo方法內部使用nextServerCyclicCounter這個AtomicInteger屬性原子遞增對serverCount取模得到索引值
        server = allServers.get(nextServerIndex); // 得到伺服器例項
 
        if (server == null) {
            Thread.yield();
            continue;
        }
 
        if (server.isAlive() && (server.isReadyToServe())) {
            return (server);
        }
 
        server = null;
    }
 
    if (count >= 10) {
        log.warn("No available alive servers after 10 tries from load balancer: "
                + lb);
    }
    return server;
}

複製程式碼

複製程式碼

    private int incrementAndGetModulo(int modulo) {
        for (;;) {
            int current = nextServerCyclicCounter.get();
            int next = (current + 1) % modulo;
            if (nextServerCyclicCounter.compareAndSet(current, next))
                return next;
        }
    }

複製程式碼

BestAvailableRule策略用來選取最少併發量請求的伺服器：

複製程式碼

public Server choose(Object key) {
    if (loadBalancerStats == null) {
        return super.choose(key);
    }
    List<Server> serverList = getLoadBalancer().getAllServers(); // 獲取所有的伺服器列表
    int minimalConcurrentConnections = Integer.MAX_VALUE;
    long currentTime = System.currentTimeMillis();
    Server chosen = null;
    for (Server server: serverList) { // 遍歷每個伺服器
        ServerStats serverStats = loadBalancerStats.getSingleServerStat(server); // 獲取各個伺服器的狀態
        if (!serverStats.isCircuitBreakerTripped(currentTime)) { // 沒有觸發斷路器的話繼續執行
            int concurrentConnections = serverStats.getActiveRequestsCount(currentTime); // 獲取當前伺服器的請求個數
            if (concurrentConnections < minimalConcurrentConnections) { // 比較各個伺服器之間的請求數，然後選取請求數最少的伺服器並放到chosen變數中
                minimalConcurrentConnections = concurrentConnections;
                chosen = server;
            }
        }
    }
    if (chosen == null) { // 如果沒有選上，呼叫父類ClientConfigEnabledRoundRobinRule的choose方法，也就是使用RoundRobinRule輪詢的方式進行負載均衡        
        return super.choose(key);
    } else {
        return chosen;
    }
}

複製程式碼

加權響應時間負載均衡 （WeightedResponseTime）

區域感知輪詢負載均衡（ZoneAware）：

區域感知負載均衡內建電路跳閘邏輯，可被配置基於區域同源關係（Zone Affinity，也就是更傾向於選擇發出呼叫的服務所在的託管區域內，這樣可以降低延遲，節省成本）選擇目標服務例項。它監控每個區域中執行例項的行為，而且能夠實時的快速丟棄一整個區域。這樣在面對整個區域故障時，幫我們提升了彈性。

例項驗證Ribbon中的LoadBalance功能

ServerList中提供了3個instance，分別是：

compute-service:2222
compute-service:2223
compute-service:2224

然後使用不同的IRule策略檢視負載均衡的實現。

複製程式碼

package org.springframework.cloud.client.loadbalancer;

import org.springframework.beans.factory.annotation.Qualifier;

import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * 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 {
}

複製程式碼

LoadBalancerAutoConfiguration.java為實現客戶端負載均衡器的自動化配置類。

複製程式碼

package org.springframework.cloud.client.loadbalancer;



/**
 * Auto configuration for Ribbon (client side load balancing).
 *
 * @author Spencer Gibb
 * @author Dave Syer
 * @author Will Tran
 */
@Configuration
@ConditionalOnClass(RestTemplate.class)
@ConditionalOnBean(LoadBalancerClient.class)
@EnableConfigurationProperties(LoadBalancerRetryProperties.class)
public class LoadBalancerAutoConfiguration {

    @LoadBalanced
    @Autowired(required = false)
    private List<RestTemplate> restTemplates = Collections.emptyList();

    @Bean
    public SmartInitializingSingleton loadBalancedRestTemplateInitializer(
            final List<RestTemplateCustomizer> customizers) {
        return new SmartInitializingSingleton() {
            @Override
            public void afterSingletonsInstantiated() {
                for (RestTemplate restTemplate : LoadBalancerAutoConfiguration.this.restTemplates) {
                    for (RestTemplateCustomizer customizer : customizers) {
                        customizer.customize(restTemplate);
                    }
                }
            }
        };
    }

複製程式碼

Ribbon要實現負載均衡自動化配置需要滿足如下兩個條件：

  • @ConditionalOnClass(RestTemplate.class)：RestTemplate類必須存在於當前工程的環境中。
  • @ConditionalOnBean(LoadBalancerClient.class)：在spring的Bean工程中必須有LoadBalancerClient.class的實現Bean。

在自動化配置中主要做三件事：

  • 建立一個LoadBalancerInterceptor的Bean，用於實現對客戶端發起請求時進行攔截，以實現客戶端負載均衡。
  • 建立一個RestTemplateCustomizer的Bean，用於給RestTemplate增加LoadBalancerInterceptor攔截器。
  • 維護了一個被@LoadBalanced註解修飾的RestTemplate物件列表，並在這裡進行初始化，通過呼叫RestTemplateCustomizer的例項來給需要客戶端負載均衡的RestTemplate增加LoadBalancerInterceptor攔截器。

LoadBalancerAutoConfiguration.java裡面有2個內部類，如下：

複製程式碼

@Configuration
    @ConditionalOnMissingClass("org.springframework.retry.support.RetryTemplate")
    static class LoadBalancerInterceptorConfig {
        @Bean
        public LoadBalancerInterceptor ribbonInterceptor(
                LoadBalancerClient loadBalancerClient,
                LoadBalancerRequestFactory requestFactory) {
            return new LoadBalancerInterceptor(loadBalancerClient, requestFactory);
        }

        @Bean
        @ConditionalOnMissingBean
        public RestTemplateCustomizer restTemplateCustomizer(
                final LoadBalancerInterceptor loadBalancerInterceptor) {
            return new RestTemplateCustomizer() {
                @Override
                public void customize(RestTemplate restTemplate) {
                    List<ClientHttpRequestInterceptor> list = new ArrayList<>(
                            restTemplate.getInterceptors());
                    list.add(loadBalancerInterceptor);
                    restTemplate.setInterceptors(list);
                }
            };
        }
    }

    @Configuration
    @ConditionalOnClass(RetryTemplate.class)
    static class RetryAutoConfiguration {
        @Bean
        public RetryTemplate retryTemplate() {
            RetryTemplate template =  new RetryTemplate();
            template.setThrowLastExceptionOnExhausted(true);
            return template;
        }

        @Bean
        @ConditionalOnMissingBean
        public LoadBalancedRetryPolicyFactory loadBalancedRetryPolicyFactory() {
            return new LoadBalancedRetryPolicyFactory.NeverRetryFactory();
        }

        @Bean
        public RetryLoadBalancerInterceptor ribbonInterceptor(
                LoadBalancerClient loadBalancerClient, LoadBalancerRetryProperties properties,
                LoadBalancedRetryPolicyFactory lbRetryPolicyFactory,
                LoadBalancerRequestFactory requestFactory) {
            return new RetryLoadBalancerInterceptor(loadBalancerClient, retryTemplate(), properties,
                    lbRetryPolicyFactory, requestFactory);
        }

        @Bean
        @ConditionalOnMissingBean
        public RestTemplateCustomizer restTemplateCustomizer(
                final RetryLoadBalancerInterceptor loadBalancerInterceptor) {
            return new RestTemplateCustomizer() {
                @Override
                public void customize(RestTemplate restTemplate) {
                    List<ClientHttpRequestInterceptor> list = new ArrayList<>(
                            restTemplate.getInterceptors());
                    list.add(loadBalancerInterceptor);
                    restTemplate.setInterceptors(list);
                }
            };
        }
    }

複製程式碼

 接下來，我們看看LoadBalancerInterceptor攔截器是如何將一個普通的RestTemplate變成客戶度負載均衡的：

複製程式碼

/*
 * Copyright 2013-2017 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.springframework.cloud.client.loadbalancer;

import java.io.IOException;
import java.net.URI;

import org.springframework.http.HttpRequest;
import org.springframework.http.client.ClientHttpRequestExecution;
import org.springframework.http.client.ClientHttpRequestInterceptor;
import org.springframework.http.client.ClientHttpResponse;

/**
 * @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();
        return this.loadBalancer.execute(serviceName, requestFactory.createRequest(request, body, execution)); //看這裡
    }
}

複製程式碼

LoadBalancerRequestFactory .java  

複製程式碼

/*
 * Copyright 2017 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.springframework.cloud.client.loadbalancer;

import java.util.List;

import org.springframework.cloud.client.ServiceInstance;
import org.springframework.http.HttpRequest;
import org.springframework.http.client.ClientHttpRequestExecution;
import org.springframework.http.client.ClientHttpResponse;

/**
 * Creates {@link LoadBalancerRequest}s for {@link LoadBalancerInterceptor} and
 * {@link RetryLoadBalancerInterceptor}. Applies
 * {@link LoadBalancerRequestTransformer}s to the intercepted
 * {@link HttpRequest}.
 * 
 * @author William Tran
 *
 */
public class LoadBalancerRequestFactory {

    private LoadBalancerClient loadBalancer;
    private List<LoadBalancerRequestTransformer> transformers;

    public LoadBalancerRequestFactory(LoadBalancerClient loadBalancer,
            List<LoadBalancerRequestTransformer> transformers) {
        this.loadBalancer = loadBalancer;
        this.transformers = transformers;
    }

    public LoadBalancerRequestFactory(LoadBalancerClient loadBalancer) {
        this.loadBalancer = loadBalancer;
    }

    public LoadBalancerRequest<ClientHttpResponse> createRequest(final HttpRequest request,
            final byte[] body, final ClientHttpRequestExecution execution) {
        return new LoadBalancerRequest<ClientHttpResponse>() {

            @Override
            public ClientHttpResponse apply(final ServiceInstance instance)
                    throws Exception {
                HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance, loadBalancer);
                if (transformers != null) {
                    for (LoadBalancerRequestTransformer transformer : transformers) {
                        serviceRequest = transformer.transformRequest(serviceRequest, instance);
                    }
                }
                return execution.execute(serviceRequest, body);
            }

        };
    }

}

複製程式碼

ServiceRequestWrapper.java

複製程式碼

package org.springframework.cloud.client.loadbalancer;

import java.net.URI;

import org.springframework.cloud.client.ServiceInstance;
import org.springframework.http.HttpRequest;
import org.springframework.http.client.support.HttpRequestWrapper;

/**
 * @author Ryan Baxter
 */
public class ServiceRequestWrapper extends HttpRequestWrapper {
    private final ServiceInstance instance;
    private final LoadBalancerClient loadBalancer;

    public ServiceRequestWrapper(HttpRequest request, ServiceInstance instance,
                                 LoadBalancerClient loadBalancer) {
        super(request);
        this.instance = instance;
        this.loadBalancer = loadBalancer;
    }

    @Override
    public URI getURI() {
        URI uri = this.loadBalancer.reconstructURI(
                this.instance, getRequest().getURI());
        return uri;
    }
}

複製程式碼

 spring cloud中對應的實現類：

複製程式碼

package org.springframework.cloud.netflix.ribbon;


/**
 * @author Spencer Gibb
 * @author Dave Syer
 * @author Ryan Baxter
 */
public class RibbonLoadBalancerClient implements LoadBalancerClient {

    private SpringClientFactory clientFactory;

    public RibbonLoadBalancerClient(SpringClientFactory clientFactory) {
        this.clientFactory = clientFactory;
    }

    @Override
    public URI reconstructURI(ServiceInstance instance, URI original) {
        Assert.notNull(instance, "instance can not be null");
        String serviceId = instance.getServiceId();
        RibbonLoadBalancerContext context = this.clientFactory
                .getLoadBalancerContext(serviceId);
        Server server = new Server(instance.getHost(), instance.getPort());
        IClientConfig clientConfig = clientFactory.getClientConfig(serviceId);
        ServerIntrospector serverIntrospector = serverIntrospector(serviceId);
        URI uri = RibbonUtils.updateToHttpsIfNeeded(original, clientConfig,
                serverIntrospector, server);
        return context.reconstructURIWithServer(server, uri);
    }

    @Override
    public ServiceInstance choose(String serviceId) {
        Server server = getServer(serviceId);
        if (server == null) {
            return null;
        }
        return new RibbonServer(serviceId, server, isSecure(server, serviceId),
                serverIntrospector(serviceId).getMetadata(server));
    }

    @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);
    }

    @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;
    }

    private ServerIntrospector serverIntrospector(String serviceId) {
        ServerIntrospector serverIntrospector = this.clientFactory.getInstance(serviceId,
                ServerIntrospector.class);
        if (serverIntrospector == null) {
            serverIntrospector = new DefaultServerIntrospector();
        }
        return serverIntrospector;
    }

    private boolean isSecure(Server server, String serviceId) {
        IClientConfig config = this.clientFactory.getClientConfig(serviceId);
        ServerIntrospector serverIntrospector = serverIntrospector(serviceId);
        return RibbonUtils.isSecure(config, serverIntrospector, server);
    }

    protected Server getServer(String serviceId) {
        return getServer(getLoadBalancer(serviceId));
    }

    protected Server getServer(ILoadBalancer loadBalancer) {
        if (loadBalancer == null) {
            return null;
        }
        return loadBalancer.chooseServer("default"); // TODO: better handling of key
    }

    protected ILoadBalancer getLoadBalancer(String serviceId) {
        return this.clientFactory.getLoadBalancer(serviceId);
    }

複製程式碼

getServer方法中並沒有使用LoadBalancerClient中的choose方法，而是使用Netflix Rion自身的ILoadBalancer介面中定義的chooseServer方法。再看ILoadBalancer 介面：

複製程式碼

package com.netflix.loadbalancer;

public interface ILoadBalancer {
    //向負載均衡器的例項列表中增加例項
    public void addServers(List<Server> newServers);

    //通過某種策略，從負載均衡器中選擇一個具體的例項    
    public Server chooseServer(Object key);
    //用來通知和標識負載均衡器中某個具體例項已經停止服務，不然負載均衡器在下一次獲取服務例項清單前都會認為服務例項均是正常服務的。
    public void markServerDown(Server server);
    
   //獲取正常服務列表
    public List<Server> getReachableServers();

    //所有已知例項列表
    public List<Server> getAllServers();

複製程式碼

 再看實現類，BaseLoadBalancer類是實現了基礎的負載均衡，而DynamicServerListLoadBalancer和ZoneAwareLoadBalancer在負載均衡基礎上做了一些功能的擴充套件。

 

 那麼Spring cloud在整合Ribbon的時候採用的哪個具體實現，可以看RibbonClientConfiguration配置類中的程式碼片段如下，採用的是ZoneAwareLoadBalancer來實現負載均衡器。

複製程式碼

    @Bean
    @ConditionalOnMissingBean
    public ILoadBalancer ribbonLoadBalancer(IClientConfig config,
            ServerList<Server> serverList, ServerListFilter<Server> serverListFilter,
            IRule rule, IPing ping) {
        if (this.propertiesFactory.isSet(ILoadBalancer.class, name)) {
            return this.propertiesFactory.get(ILoadBalancer.class, config, name);
        }
        ZoneAwareLoadBalancer<Server> balancer = LoadBalancerBuilder.newBuilder()
                .withClientConfig(config).withRule(rule).withPing(ping)
                .withServerListFilter(serverListFilter).withDynamicServerList(serverList)
                .buildDynamicServerListLoadBalancer();
        return balancer;
    }

複製程式碼

 在回到主方法RibbonLoadBalancerClient.execute()

LoadBalancerClient的execute()
-->1、ZoneAwareLoadBalancer.chooseServer()獲取了負載均衡策略分配到的服務例項物件Server
-->2、將Server物件封裝成RibbonService例項
-->3、呼叫LoadBalancerRequest的apply()
-->4-1、在apply()中，先將request包裝成ServiceRequestWrapper，在Wrapper中拼接URI
-->4-2、拼接URI中，呼叫RibbonLoadBalancerClient.reconstructURI()
-->4-3、拼接URI中，呼叫RibbonLoadBalancerContext.reconstructURIWithServer()
-->4-4、拼接URI中，呼叫RibbonLoadBalancerContext.reconstructURIWithServer()
-->5、攔截器呼叫

-->6、執行完成後，Ribbon還通過RibbonStatsRecorder物件對服務的請求進行了記錄

複製程式碼

    @Override
    public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException {
        ILoadBalancer loadBalancer = getLoadBalancer(serviceId); //在springcloud中是ZoneAwareLoadBalancer例項
        Server server = getServer(loadBalancer); //1、ZoneAwareLoadBalancer獲取了負載均衡策略分配到的服務例項物件Server
        if (server == null) {
            throw new IllegalStateException("No instances available for " + serviceId);
        }
        //2、將Server例項封裝成RibbonService例項
        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);  //3、request看LoadBalancerRequestFactory中的createRequest()方法返回的匿名類
            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;
    }

複製程式碼

 

複製程式碼

    public LoadBalancerRequest<ClientHttpResponse> createRequest(final HttpRequest request,
            final byte[] body, final ClientHttpRequestExecution execution) {
        return new LoadBalancerRequest<ClientHttpResponse>() {

            @Override
            public ClientHttpResponse apply(final ServiceInstance instance)
                    throws Exception {
                HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance, loadBalancer);  //4-1、中拼接URI，見下面的ServiceRequestWrapper中的getURI()
                if (transformers != null) {
                    for (LoadBalancerRequestTransformer transformer : transformers) {
                        serviceRequest = transformer.transformRequest(serviceRequest, instance);
                    }
                }
                return execution.execute(serviceRequest, body);  //5、攔截器呼叫見InterceptingClientHttpRequest的內部類InterceptingRequestExecution.execute()
            }

        };
    }

複製程式碼

    @Override
    public URI getURI() {
        URI uri = this.loadBalancer.reconstructURI(
                this.instance, getRequest().getURI()); //4-2、reconstructURI被RibbonLoadBalancerClient過載，看RibbonLoadBalancerClient.reconstructURI()
        return uri;
    }

複製程式碼

    @Override
    public URI reconstructURI(ServiceInstance instance, URI original) {
        Assert.notNull(instance, "instance can not be null");
        String serviceId = instance.getServiceId();
        RibbonLoadBalancerContext context = this.clientFactory
                .getLoadBalancerContext(serviceId);
        Server server = new Server(instance.getHost(), instance.getPort());
        IClientConfig clientConfig = clientFactory.getClientConfig(serviceId);
        ServerIntrospector serverIntrospector = serverIntrospector(serviceId);
        URI uri = RibbonUtils.updateToHttpsIfNeeded(original, clientConfig,
                serverIntrospector, server);
        return context.reconstructURIWithServer(server, uri);//4-3、構建服務例項的URI，RibbonLoadBalancerContext.reconstructURIWithServer()見下面的LoadBalancerContext.reconstructURIWithServer()
    }

複製程式碼

複製程式碼

    public URI reconstructURIWithServer(Server server, URI original) {//4-4、LoadBalancerContext的reconstructURIWithServer()
        String host = server.getHost();
        int port = server .getPort();
        if (host.equals(original.getHost()) 
                && port == original.getPort()) {
            return original;
        }
        String 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);
        }
    }

複製程式碼

複製程式碼

        @Override
        public ClientHttpResponse execute(HttpRequest request, byte[] body) throws IOException {//5、攔截器呼叫，InterceptingClientHttpRequest的內部類InterceptingRequestExecution.execute()
            if (this.iterator.hasNext()) {
                ClientHttpRequestInterceptor nextInterceptor = this.iterator.next();
                return nextInterceptor.intercept(request, body, this);
            }
            else {
                ClientHttpRequest delegate = requestFactory.createRequest(request.getURI(), request.getMethod());
                for (Map.Entry<String, List<String>> entry : request.getHeaders().entrySet()) {
                    List<String> values = entry.getValue();
                    for (String value : values) {
                        delegate.getHeaders().add(entry.getKey(), value);
                    }
                }
                if (body.length > 0) {
                    StreamUtils.copy(body, delegate.getBody());
                }
                return delegate.execute();
            }
        }

複製程式碼

複製程式碼

public class RibbonStatsRecorder {

    private RibbonLoadBalancerContext context;
    private ServerStats serverStats;
    private Stopwatch tracer;

    public RibbonStatsRecorder(RibbonLoadBalancerContext context, Server server) {
        this.context = context;
        if (server != null) {
            serverStats = context.getServerStats(server);
            context.noteOpenConnection(serverStats);
            tracer = context.getExecuteTracer().start();
        }
    }

    public void recordStats(Object entity) {
        this.recordStats(entity, null);
    }

    public void recordStats(Throwable t) {
        this.recordStats(null, t);
    }

    protected void recordStats(Object entity, Throwable exception) {
        if (this.tracer != null && this.serverStats != null) {
            this.tracer.stop();
            long duration = this.tracer.getDuration(TimeUnit.MILLISECONDS);
            this.context.noteRequestCompletion(serverStats, entity, exception, duration, null/* errorHandler */);
        }
    }
}

複製程式碼

https://www.cnblogs.com/duanxz/p/7504947.html

相關推薦

Ribbon之Loadbalance的原始碼分析

Load Balance負載均衡是用於解決一臺機器(一個程序)無法解決所有請求而產生的一種演算法。 像nginx可以使用負載均衡分配流量，ribbon為客戶端提供負載均衡，dubbo服務呼叫裡的負載均衡等等，很多地方都使用到了負載均衡。 使用負載均衡帶來的好處很明顯： 當叢集裡的1

【kubernetes/k8s原始碼分析】kubelet原始碼分析之cdvisor原始碼分析

  資料流 UnsecuredDependencies -> run   1. cadvisor.New初始化 if kubeDeps.CAdvisorInterface == nil { imageFsInfoProvider := cadv

【kubernetes/k8s原始碼分析】 controller-manager之replicaset原始碼分析

ReplicaSet簡介     Kubernetes 中建議使用 ReplicaSet來取代 ReplicationController。ReplicaSet 跟 ReplicationController 沒有本質的不同， ReplicaSet 支援集合式的

【kubernetes/k8s原始碼分析】 client-go包之Informer原始碼分析

Informer 簡介        Informer 是 Client-go 中的一個核心工具包。如果 Kubernetes 的某個元件，需要 List/Get Kubernetes 中的 Object（包括pod，service等等），可以直接使用

【go原始碼分析】go原始碼之slice原始碼分析

Go 語言切片是對陣列的抽象。 Go 陣列的長度不可改變，與陣列相比切片的長度是不固定的，可以追加元素，在追加時可能使切片的容量增大。 len() 和 cap() 函式     切片是可索引的，並且可以由 len() 方法獲取長度。    

【go原始碼分析】go原始碼之list原始碼分析

本文針對go 1.11版本，路徑src/container/list/list.go 資料結構 Element結構體 Value 前驅 後繼 // Element is an element of a linked list. type Element st

RecyclerView之SnapHelper原始碼分析

很久沒有寫Android控制元件了，正好最近專案有個自定義控制元件的需求，整理了下做個總結，主要是實現類似於抖音翻頁的效果，但是有有點不同，需要在底部漏出後面的view，這樣說可能不好理解，看下Demo，按頁滑動，後面的View有放大縮放的動畫，滑動速度過小時會有回到原位的效果，下滑也是按頁滑動的效果。

springMVC原始碼學習之addFlashAttribute原始碼分析

本文主要從falshMap初始化，存，取，消毀來進行原始碼分析，springmvc版本4.3.18。關於使用及驗證請參考另一篇https://www.cnblogs.com/pu20065226/p/10032048.html 1.初始化和呼叫，首先是入springMVC 入口webmvc包中org.spr

java集合之----ArrayList原始碼分析（基於jdk1.8）

一、ArrayList 1、ArrayList是什麼： ArrayList就是動態陣列，用MSDN中的說法，就是Array的複雜版本，它提供了動態的增加和減少元素，實現了ICollection和IList介面，靈活的設定陣列的大小等好處，實現了Randomaccess介面，支援快速隨

java集合之----HashMap原始碼分析（基於JDK1.7與1.8）

一、什麼是HashMap 百度百科這樣解釋： 簡而言之，HashMap儲存的是鍵值對（key和value），通過key對映到value，具有很快的訪問速度。HashMap是非執行緒安全的，也就是說在多執行緒併發環境下會出現問題（死迴圈） 二、內部實現 （1）結構 HashM

ROS Navigation Stack之dwa_local_planner原始碼分析

DWA和base_local_planner的關係 在base_local_planner包中有兩個檔案叫trajectory_planner.cpp 以及對應的ros實現，其和DWA是同一層的。 由於nav_core提供了統一的介面，因此我們可以先看看統一的介面有哪些，那我們便知

go 原始碼學習之---Tail 原始碼分析

已經有兩個月沒有寫部落格了，也有好幾個月沒有看go相關的內容了，由於工作原因最近在做java以及大資料相關的內容，導致最近工作較忙，部落格停止了更新，正好想撿起之前go的東西，所以找了一個原始碼學習 這個也是之前用go寫日誌收集的時候用到的一個包 :github.com/hpcloud/tail, 這次就學

Java執行緒池之ThreadPoolExecutor原始碼分析

一、引言 Java併發工具包自帶了很多常用的執行緒池，程式可以將定義的Runnable、Callable任務提交到執行緒池當中執行，由執行緒池負責非同步執行其中的任務。 Java執行緒池框架結構圖： 其中，Executors是一個執行緒池靜態工廠類，可以呼叫其

【轉】Java併發程式設計筆記之CopyOnWriteArrayList原始碼分析

併發包中併發List只有CopyOnWriteArrayList這一個，CopyOnWriteArrayList是一個執行緒安全的ArrayList,對其進行修改操作和元素迭代操作都是在底層建立一個拷貝陣列（快照）上進行的，也就是寫時拷貝策略。 我們首先看一下CopyOnW

實戰java高併發程式設計之CountDownLatch原始碼分析

首先看第一個！ CountDownLatch 使用場景 CountDownLatch類是常見的併發同步控制類，適用於某一執行緒的執行在其他多個執行緒執行完成之後，比如火箭發射前需要各項指標檢查，只有當各項指標檢查完才能發射，再比如解析多個excel文件，只有當

原始碼分析篇之String原始碼分析二

    前面已經分析過String原始碼為什麼是不可變的，同時通過我們常用的String的相關的類StringBuffer和StringBuilder，我們可以發現String類中欄位名被定義為了final型別，這樣的話將只能被賦值一次。接下來，繼續看String原始碼實現的

實戰java高併發程式設計之ReentrantReadWriteLoc原始碼分析

前面分析了併發工具類CountDownLatch和CyclicBarrier，本文分享分析比較重要的ReentrantReadWriteLock。 使用場景 以前的同步方式需要對讀、寫操作進行同步，讀讀之間，讀寫之間，寫寫之間等；工程師們發現讀讀之間並不會影響資

原始碼分析篇之String原始碼分析三

    上期已經討論了比較器的相關的問題。同時也提出問題明明父類已經實現了某個介面，子類為什麼還是要實現同樣的介面呢。這究竟有什麼意義呢。通過使用度孃的多次探究之後最終我還是沒有發現說它有什麼特別的含義。我認為的最大的含義就是告訴使用者這個類實現了這樣的一個介面。方便使用者進

SpringMVC原始碼學習之request處理流程 springMVC原始碼學習地址 springMVC原始碼學習之addFlashAttribute原始碼分析 java reflect反射呼叫方法invoke

目的：為看原始碼提供呼叫地圖，最長呼叫邏輯深度為8層，反正我是springMVC原始碼學習地址看了兩週才理出來的。 1.處理流程（版本為4.3.18） 入口為spring-webmvc-4.3.18.RELEASE.jar中org.springframework.web.servlet.Dispatche

Java集合框架之ArrayList原始碼分析

分析： java集合框架主要包括兩種型別的容器，集合（Collection）,儲存元素集合，圖(Map)，儲存鍵值對對映，而Collection介面又有三種子型別：List，Set,Queue,然後是一些抽象類，最後是一些實現類，常用ArrayList,Lin