摘要：客戶端負(fù)載均衡需要客戶端自己維護(hù)自己要訪問的服務(wù)實(shí)例清單，這些服務(wù)清單來源于注冊(cè)中心在使用進(jìn)行服務(wù)治理時(shí)。使用從負(fù)載均衡器中挑選出的服務(wù)實(shí)例來執(zhí)行請(qǐng)求內(nèi)容。

?Spring Cloud Ribbon是一個(gè)基于HTTP和TCP的客戶端負(fù)載均衡工具，基于Netflix Ribbon實(shí)現(xiàn)。

客戶端負(fù)載均衡(本文重點(diǎn))

源碼分析(本文重點(diǎn))

負(fù)載均衡器

負(fù)載均衡策略

配置詳解

自動(dòng)化配置

?負(fù)載均衡是對(duì)系統(tǒng)的高可用、網(wǎng)絡(luò)壓力的緩解和處理內(nèi)容擴(kuò)容的重要手段之一。

?負(fù)載均衡可以分為客戶端負(fù)載均衡和服務(wù)端負(fù)載均衡。

?負(fù)載均衡按設(shè)備來分為硬件負(fù)載均衡和軟件負(fù)載均衡，都屬于服務(wù)端負(fù)載均衡。

?硬件負(fù)載均衡主要通過在服務(wù)器節(jié)點(diǎn)之間安裝專門用于負(fù)載均衡的設(shè)備，例如F5等。

?軟件負(fù)載均衡通過在服務(wù)器上安裝一些具有負(fù)載均衡功能或模塊的軟件來完成請(qǐng)求的轉(zhuǎn)發(fā)工作，例如Nginx等。

?硬件負(fù)載均衡和軟件負(fù)載均衡都會(huì)維護(hù)一個(gè)可用的服務(wù)清單，然后通過心跳檢測來剔除故障節(jié)點(diǎn)以保證服務(wù)清單中的節(jié)點(diǎn)都正常可用。當(dāng)客戶端發(fā)出請(qǐng)求時(shí)，負(fù)載均衡器會(huì)按照某種算法(線性輪詢、按權(quán)重負(fù)載、按流量負(fù)載等)從服務(wù)清單中取出一臺(tái)服務(wù)器的地址，然后將請(qǐng)求轉(zhuǎn)發(fā)到該服務(wù)器上。

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

?基于Spring Cloud Ribbon實(shí)現(xiàn)客戶端負(fù)載均衡非常簡單，主要由以下步驟：

服務(wù)提供者需要啟動(dòng)多個(gè)服務(wù)實(shí)例并注冊(cè)到一個(gè)或多個(gè)相關(guān)聯(lián)的服務(wù)注冊(cè)中心上；

服務(wù)消費(fèi)者直接通過帶有@LoadBalanced注解的RestTemplate向服務(wù)提供者發(fā)送請(qǐng)求以實(shí)現(xiàn)客戶端的負(fù)載均衡。

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

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

?該類是一個(gè)接口類，代碼如下：

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

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

ServiceInstance choose(String serviceId) : 根據(jù)傳入的服務(wù)名稱，從負(fù)載均衡器中挑選一個(gè)對(duì)應(yīng)服務(wù)的實(shí)例。

T execute(String serviceId, LoadBalancerRequest request) throws IOException : 使用從負(fù)載均衡器中挑選出的服務(wù)實(shí)例來執(zhí)行請(qǐng)求內(nèi)容。

URI reconstructURI(ServiceInstance instance, URI original) : 構(gòu)建一個(gè)符合host:port格式的URI。在分布式系統(tǒng)中，我們都使用邏輯上的服務(wù)名作為host來構(gòu)建URI(代替服務(wù)實(shí)例的host:port形式)進(jìn)行請(qǐng)求。在該操作的定義中，前者ServiceInstance對(duì)象是帶有host和port的具體服務(wù)實(shí)例，后者URI對(duì)象則是使用邏輯服務(wù)名的URI，返回的URI是根據(jù)這兩者轉(zhuǎn)換后的host:port形式的URI。

?通過分析org.springframework.cloud.client.loadbalancer包中的類，可以找出org.springframework.cloud.client.loadbalancer.LoadBalancerAutoConfiguration是實(shí)現(xiàn)客戶端負(fù)載均衡的自動(dòng)配置類

?從LoadBalancerAutoConfiguration類的頭部注解上可以看出，Ribbon實(shí)現(xiàn)客戶端負(fù)載均衡的自動(dòng)配置需要滿足下面兩個(gè)條件：

@ConditionalOnClass(RestTemplate.class) : RestTemplate類必須存在于當(dāng)前工程的環(huán)境中

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

?在該自動(dòng)化配置的任務(wù)中，主要完成以下三個(gè)任務(wù)：

創(chuàng)建一個(gè)LoadBalancerInterceptor實(shí)例，用于客戶端發(fā)起請(qǐng)求時(shí)進(jìn)行攔截，進(jìn)而實(shí)現(xiàn)客戶端的負(fù)載均衡

創(chuàng)建一個(gè)RestTemplateCustomizer實(shí)例，用于給RestTemplate增加LoadBalancerInterceptor攔截器

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

?通過上面可以看出，真正實(shí)現(xiàn)客戶端負(fù)載均衡是因?yàn)橛蠰oadBalancerInterceptor攔截器的存在，那么下面看一下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方法還用到另外兩個(gè)類LoadBalancerRequestFactory和ServiceRequestWrapper。

?當(dāng)使用被@LoadBalanced注解的RestTemplate發(fā)送請(qǐng)求時(shí)，會(huì)被LoadBalancerInterceptor攔截器攔截，執(zhí)行LoadBalancerInterceptor的intercept方法,最終在該方法中選擇合適的服務(wù)實(shí)例通過LoadBalancerClient的execute方法進(jìn)行調(diào)用。

?因?yàn)長oadBalancerClient是抽象的負(fù)載均衡器接口，下面我們可以通過一個(gè)具體的負(fù)載均衡器RibbonLoadBalancerClient來進(jìn)行分析。

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)方法的執(zhí)行步驟:

?1.獲取負(fù)載均衡器，默認(rèn)的負(fù)載均衡器是ZoneAwareLoadBalancer，這個(gè)后面再講

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

?3.將獲取到的具體服務(wù)實(shí)例包裝成RibbonServer對(duì)象(該對(duì)象存儲(chǔ)了服務(wù)實(shí)例信息、服務(wù)名serviceId、是否需要https等其他信息)，最后調(diào)用LoadBalancerRequest的apply方法，向一個(gè)實(shí)際的的具體服務(wù)實(shí)例發(fā)起請(qǐng)求，請(qǐng)看下面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();
}

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

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

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

void markServerDown(Server server) : 通知和標(biāo)識(shí)負(fù)載均衡器中的某個(gè)具體的服務(wù)實(shí)例已經(jīng)停止服務(wù)，防止負(fù)載均衡器在下一次獲取服務(wù)實(shí)例清單前認(rèn)為該服務(wù)實(shí)例是正常服務(wù)

List getReachableServers() : 獲取當(dāng)前正常的服務(wù)實(shí)例列表

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

?該接口中使用到的Server對(duì)象定義是一個(gè)傳統(tǒng)的服務(wù)端節(jié)點(diǎn)，在該類中存儲(chǔ)了服務(wù)端節(jié)點(diǎn)的一些元數(shù)據(jù)信息，包括host、port以及一些部署信息等。

?整理該接口的實(shí)現(xiàn)類主要有：

AbstractLoadBalancer(抽象類)

BaseLoadBalancer:繼承自AbstractLoadBalancer

DynamicServerListLoadBalancer:繼承自BaseLoadBalancer

NoOpLoadBalancer:繼承自AbstractLoadBalancer

ZoneAwareLoadBalancer:繼承自DynamicServerListLoadBalancer

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

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

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

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

?該接口暴露了服務(wù)實(shí)例的一些基本信息，如：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的一個(gè)具體實(shí)現(xiàn)，查看源碼可以知道RibbonServer除了包含Server對(duì)象之外，還存儲(chǔ)了服務(wù)名、是否使用HTTPS以及一個(gè)Map類型的元數(shù)據(jù)集合。

?看完上面兩個(gè)類，我們來看一下具體的apply方法的實(shí)現(xiàn)，源碼中使用Lamada表達(dá)式實(shí)現(xiàn)，下面我只抽出具體的功能實(shí)現(xiàn)的源碼如下:

            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對(duì)象，下面看一下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方法，重寫后的該方法通過調(diào)用LoadBalancerClient中的reconstructURI方法來構(gòu)建一個(gè)host:port形式的URI對(duì)外發(fā)起請(qǐng)求。

?在apply方法的最后在調(diào)用ClientHttpRequestExecution的execute方法時(shí)，實(shí)際會(huì)去執(zhí)行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();
            }
        }
    }

?分析上面的代碼可以看出，在調(diào)用requestFactory.createRequest(request.getURI(), method)創(chuàng)建請(qǐng)求時(shí)會(huì)調(diào)用getURI方法，此時(shí)調(diào)用的getURI方法就是ServiceRequestWrapper類中的getURI方法，進(jìn)而調(diào)用LoadBalancerClient中的reconstructURI方法，下面我們看一下RibbonLoadBalancerClient中的reconstructURI是如何實(shí)現(xiàn)的

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

?分析上面的代碼，首先根據(jù)serviceId從SpringClientFactory對(duì)象中獲取serviceId對(duì)應(yīng)的負(fù)載均衡器的上下文RibbonLoadBalancerContext對(duì)象，然后再使用該上下文對(duì)象的reconstructURIWithServer方法和server對(duì)象來構(gòu)建具體的URI。

?備注：

SpringClientFactory : 一個(gè)用來創(chuàng)建客戶端負(fù)載均衡器的工廠類，該工廠類會(huì)為每一個(gè)不同名的Ribbon客戶端生成不同的上下文

RibbonLoadBalancerContext : LoadBalancerContext的子類，該類用于存儲(chǔ)一些被負(fù)載均衡器使用的上下文內(nèi)容和API操作。

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

?使用被@LoadBalanced注解的RestTemplate發(fā)起請(qǐng)求時(shí)，會(huì)被LoadBalancerInterceptor攔截，然后借助負(fù)載均衡器LoadBalancerClient將邏輯服務(wù)名轉(zhuǎn)換為host:port的具體的服務(wù)實(shí)例地址，在使用RibbonLoadBalancerClient(Ribbon實(shí)現(xiàn)的負(fù)載均衡器)時(shí)實(shí)際使用的是Ribbon中定義的ILoadBalancer，默認(rèn)自動(dòng)化配置的負(fù)載均衡器是ZoneAwareLoadBalancer。

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

?后面會(huì)介紹負(fù)載均衡器的源碼分析，請(qǐng)繼續(xù)關(guān)注?。?！

?前往微信公眾號(hào)閱讀文章，點(diǎn)擊這里，或者直接掃碼關(guān)注公眾號(hào)