摘要：客戶端負載均衡需要客戶端自己維護自己要訪問的服務實例清單，這些服務清單來源于注冊中心在使用進行服務治理時。使用從負載均衡器中挑選出的服務實例來執行請求內容。

?Spring Cloud Ribbon是一個基于HTTP和TCP的客戶端負載均衡工具，基于Netflix Ribbon實現。

客戶端負載均衡(本文重點)

源碼分析(本文重點)

負載均衡器

負載均衡策略

配置詳解

自動化配置

?負載均衡是對系統的高可用、網絡壓力的緩解和處理內容擴容的重要手段之一。

?負載均衡可以分為客戶端負載均衡和服務端負載均衡。

?負載均衡按設備來分為硬件負載均衡和軟件負載均衡，都屬于服務端負載均衡。

?硬件負載均衡主要通過在服務器節點之間安裝專門用于負載均衡的設備，例如F5等。

?軟件負載均衡通過在服務器上安裝一些具有負載均衡功能或模塊的軟件來完成請求的轉發工作，例如Nginx等。

?硬件負載均衡和軟件負載均衡都會維護一個可用的服務清單，然后通過心跳檢測來剔除故障節點以保證服務清單中的節點都正常可用。當客戶端發出請求時，負載均衡器會按照某種算法(線性輪詢、按權重負載、按流量負載等)從服務清單中取出一臺服務器的地址，然后將請求轉發到該服務器上。

?客戶端負載均衡需要客戶端自己維護自己要訪問的服務實例清單， 這些服務清單來源于注冊中心(在使用Eureka進行服務治理時)。

?基于Spring Cloud Ribbon實現客戶端負載均衡非常簡單，主要由以下步驟：

服務提供者需要啟動多個服務實例并注冊到一個或多個相關聯的服務注冊中心上；

服務消費者直接通過帶有@LoadBalanced注解的RestTemplate向服務提供者發送請求以實現客戶端的負載均衡。

?既然Ribbon客戶端負載均衡需要為RestTemplate增加@LoadBalanced注解，那么下面我們就從這個注解開始分析。

?通過該注解的頭部注釋可以得知，該注解的作用是使用LoadBalancerClient來對RestTemplate進行配置，下面接著看LoadBalancerClient

?該類是一個接口類，代碼如下：

package org.springframework.cloud.client.loadbalancer;

import org.springframework.cloud.client.ServiceInstance;

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

public interface LoadBalancerClient extends ServiceInstanceChooser {
    
    /**
     * 注意該方法是從父類ServiceInstanceChooser接口中繼承過來的
     */
    ServiceInstance choose(String serviceId);

     T execute(String serviceId, LoadBalancerRequest request) throws IOException;

     T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest request) throws IOException;

    URI reconstructURI(ServiceInstance instance, URI original);
}

?從接口的方法和注釋可以看出，一個客戶端負載均衡器應該具有以下幾種重要功能：

ServiceInstance choose(String serviceId) : 根據傳入的服務名稱，從負載均衡器中挑選一個對應服務的實例。

T execute(String serviceId, LoadBalancerRequest request) throws IOException : 使用從負載均衡器中挑選出的服務實例來執行請求內容。

URI reconstructURI(ServiceInstance instance, URI original) : 構建一個符合host:port格式的URI。在分布式系統中，我們都使用邏輯上的服務名作為host來構建URI(代替服務實例的host:port形式)進行請求。在該操作的定義中，前者ServiceInstance對象是帶有host和port的具體服務實例，后者URI對象則是使用邏輯服務名的URI，返回的URI是根據這兩者轉換后的host:port形式的URI。

?通過分析org.springframework.cloud.client.loadbalancer包中的類，可以找出org.springframework.cloud.client.loadbalancer.LoadBalancerAutoConfiguration是實現客戶端負載均衡的自動配置類

?從LoadBalancerAutoConfiguration類的頭部注解上可以看出，Ribbon實現客戶端負載均衡的自動配置需要滿足下面兩個條件：

@ConditionalOnClass(RestTemplate.class) : RestTemplate類必須存在于當前工程的環境中

@ConditionalOnBean(LoadBalancerClient.class) : 在Spring的Bean工程中必須要有LoadBalancerClient的實現Bean

?在該自動化配置的任務中，主要完成以下三個任務：

創建一個LoadBalancerInterceptor實例，用于客戶端發起請求時進行攔截，進而實現客戶端的負載均衡

創建一個RestTemplateCustomizer實例，用于給RestTemplate增加LoadBalancerInterceptor攔截器

維護一個被@LoadBalanced注解修飾的RestTemplate集合(List)，并在這里進行初始化，通過調用RestTemplateCustomizer實例來給需要客戶端負載均衡的RestTemplate增加LoadBalancerInterceptor攔截器

?通過上面可以看出，真正實現客戶端負載均衡是因為有LoadBalancerInterceptor攔截器的存在，那么下面看一下LoadBalancerInterceptor類

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;
import org.springframework.util.Assert;

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

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;

public class LoadBalancerRequestFactory {

    private LoadBalancerClient loadBalancer;
    private List transformers;

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

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

    public LoadBalancerRequest createRequest(final HttpRequest request,
            final byte[] body, final ClientHttpRequestExecution execution) {
        return instance -> {
            HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance, loadBalancer);
            if (transformers != null) {
                for (LoadBalancerRequestTransformer transformer : transformers) {
                    serviceRequest = transformer.transformRequest(serviceRequest, instance);
                }
            }
            return execution.execute(serviceRequest, body);
        };
    }
}

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;

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

?攔截器的intercept方法還用到另外兩個類LoadBalancerRequestFactory和ServiceRequestWrapper。

?當使用被@LoadBalanced注解的RestTemplate發送請求時，會被LoadBalancerInterceptor攔截器攔截，執行LoadBalancerInterceptor的intercept方法,最終在該方法中選擇合適的服務實例通過LoadBalancerClient的execute方法進行調用。

?因為LoadBalancerClient是抽象的負載均衡器接口，下面我們可以通過一個具體的負載均衡器RibbonLoadBalancerClient來進行分析。

package org.springframework.cloud.netflix.ribbon;

import java.io.IOException;
import java.net.URI;
import java.util.Collections;
import java.util.Map;

import com.netflix.client.config.IClientConfig;
import com.netflix.loadbalancer.ILoadBalancer;
import com.netflix.loadbalancer.Server;

import org.springframework.cloud.client.DefaultServiceInstance;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.LoadBalancerClient;
import org.springframework.cloud.client.loadbalancer.LoadBalancerRequest;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;

import static org.springframework.cloud.netflix.ribbon.RibbonUtils.updateToSecureConnectionIfNeeded;

public class RibbonLoadBalancerClient implements LoadBalancerClient {

    private SpringClientFactory clientFactory;

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

    @Override
    public  T execute(String serviceId, LoadBalancerRequest 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 execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest 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;
    }

}

?分析execute(String serviceId, LoadBalancerRequest request)方法的執行步驟:

?1.獲取負載均衡器，默認的負載均衡器是ZoneAwareLoadBalancer，這個后面再講

?2.獲取具體的服務實例，這里并沒有調用LoadBalancerClient的ServiceInstance choose(String serviceId)方法，而是調用的ILoadBalancer的Server chooseServer(Object key)方法來獲取具體的服務實例，想知道ILoadBalancer干啥用，請看下面的接口介紹

?3.將獲取到的具體服務實例包裝成RibbonServer對象(該對象存儲了服務實例信息、服務名serviceId、是否需要https等其他信息)，最后調用LoadBalancerRequest的apply方法，向一個實際的的具體服務實例發起請求，請看下面LoadLoadBalancerRequest的分析

package com.netflix.loadbalancer;

import java.util.List;

public interface ILoadBalancer {

    public void addServers(List newServers);
    
    public Server chooseServer(Object key);
    
    public void markServerDown(Server server);
    
    public List getReachableServers();

    public List getAllServers();
}

?該類也是負載均衡器的一個抽象接口，主要定義了一系列的抽象操作：

void addServers(List newServers) : 向負載均衡器維護的服務實例列表中增加服務實例

Server chooseServer(Object key) : 根據某種負載策略，從負載均衡器中挑選一個具體的服務實例

void markServerDown(Server server) : 通知和標識負載均衡器中的某個具體的服務實例已經停止服務，防止負載均衡器在下一次獲取服務實例清單前認為該服務實例是正常服務

List getReachableServers() : 獲取當前正常的服務實例列表

List getAllServers() : 獲取所有已知的服務實例列表

?該接口中使用到的Server對象定義是一個傳統的服務端節點，在該類中存儲了服務端節點的一些元數據信息，包括host、port以及一些部署信息等。

?整理該接口的實現類主要有：

AbstractLoadBalancer(抽象類)

BaseLoadBalancer:繼承自AbstractLoadBalancer

DynamicServerListLoadBalancer:繼承自BaseLoadBalancer

NoOpLoadBalancer:繼承自AbstractLoadBalancer

ZoneAwareLoadBalancer:繼承自DynamicServerListLoadBalancer

?默認使用的負載均衡器是ZoneAwareLoadBalancer，要想知道這個結果很簡單，查看一下RibbonClientConfiguration配置類，該類中有一個方法如下:

    @Bean
    @ConditionalOnMissingBean
    public ILoadBalancer ribbonLoadBalancer(IClientConfig config,
            ServerList serverList, ServerListFilter 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);
    }

?此處的代碼邏輯如果沒有配置負載均衡器，那么默認的負載均衡器就是ZoneAwareLoadBalancer。

?該類是一個抽象的接口，只有一個apply(ServiceInstance instance)方法，先看一下ServiceInstance類。

?該接口暴露了服務實例的一些基本信息，如：serviceId、port、host等，源碼如下：

package org.springframework.cloud.client;

import java.net.URI;
import java.util.Map;

public interface ServiceInstance {

    String getServiceId();

    String getHost();

    int getPort();

    boolean isSecure();

    URI getUri();

    Map getMetadata();
    
    default String getScheme() {
        return null;
    }
}

?上面提及的RibbonServer是ServiceInstance的一個具體實現，查看源碼可以知道RibbonServer除了包含Server對象之外，還存儲了服務名、是否使用HTTPS以及一個Map類型的元數據集合。

?看完上面兩個類，我們來看一下具體的apply方法的實現，源碼中使用Lamada表達式實現，下面我只抽出具體的功能實現的源碼如下:

            HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance, loadBalancer);
            if (transformers != null) {
                for (LoadBalancerRequestTransformer transformer : transformers) {
                    serviceRequest = transformer.transformRequest(serviceRequest, instance);
                }
            }
            return execution.execute(serviceRequest, body);

?在apply方法中，首先將HttpRequest包裝成ServiceRequestWrapper對象，下面看一下ServiceRequestWrapper類，源碼如下：

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;

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

?在ServiceRequestWrapper類中重寫了getURI方法，重寫后的該方法通過調用LoadBalancerClient中的reconstructURI方法來構建一個host:port形式的URI對外發起請求。

?在apply方法的最后在調用ClientHttpRequestExecution的execute方法時，實際會去執行InterceptingClientHttpRequest類下面的InterceptingRequestExecution的execute方法，該方法的具體代碼如下：

        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(request.getURI(), method)創建請求時會調用getURI方法，此時調用的getURI方法就是ServiceRequestWrapper類中的getURI方法，進而調用LoadBalancerClient中的reconstructURI方法，下面我們看一下RibbonLoadBalancerClient中的reconstructURI是如何實現的

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

        URI uri;
        Server server;
        if (instance instanceof RibbonServer) {
            RibbonServer ribbonServer = (RibbonServer) instance;
            server = ribbonServer.getServer();
            uri = updateToSecureConnectionIfNeeded(original, ribbonServer);
        } else {
            server = new Server(instance.getScheme(), instance.getHost(), instance.getPort());
            IClientConfig clientConfig = clientFactory.getClientConfig(serviceId);
            ServerIntrospector serverIntrospector = serverIntrospector(serviceId);
            uri = updateToSecureConnectionIfNeeded(original, clientConfig,
                    serverIntrospector, server);
        }
        return context.reconstructURIWithServer(server, uri);
    }

?分析上面的代碼，首先根據serviceId從SpringClientFactory對象中獲取serviceId對應的負載均衡器的上下文RibbonLoadBalancerContext對象，然后再使用該上下文對象的reconstructURIWithServer方法和server對象來構建具體的URI。

?備注：

SpringClientFactory : 一個用來創建客戶端負載均衡器的工廠類，該工廠類會為每一個不同名的Ribbon客戶端生成不同的上下文

RibbonLoadBalancerContext : LoadBalancerContext的子類，該類用于存儲一些被負載均衡器使用的上下文內容和API操作。

?關于LoadBalancerContext類中的reconstructURIWithServer方法是如何組裝host:port形式的邏輯很容易理解，我們就不在這里解釋了，有興趣的可以自己去看一下。

?使用被@LoadBalanced注解的RestTemplate發起請求時，會被LoadBalancerInterceptor攔截，然后借助負載均衡器LoadBalancerClient將邏輯服務名轉換為host:port的具體的服務實例地址，在使用RibbonLoadBalancerClient(Ribbon實現的負載均衡器)時實際使用的是Ribbon中定義的ILoadBalancer，默認自動化配置的負載均衡器是ZoneAwareLoadBalancer。

https://gitee.com/petterheng/spring-cloud-eureka

?后面會介紹負載均衡器的源碼分析，請繼續關注！！！

?前往微信公眾號閱讀文章，點擊這里，或者直接掃碼關注公眾號