[系统运维]负载均衡原理，探究@LoadBalanced注解都做了什么（Ribbon）

RPC-百度百科

RPC(Remote Procedure Call Protocol)–远程过程调用协议，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。RPC协议假定某些传输协议的存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层。RPC使得开发包括网络分布式多程序在内的应用程序更加容易。

RPC采用客户机/服务器模式。请求程序就是一个客户机，而服务提供程序就是一个服务器。首先，客户机调用进程发送一个有进程参数的调用信息到服务进程，然后等待应答信息。在服务器端，进程保持睡眠状态直到调用信息的到达为止。当一个调用信息到达，服务器获得进程参数，计算结果，发送答复信息，然后等待下一个调用信息，最后，客户端调用进程接收答复信息，获得进程结果，然后调用执行继续进行。

RPC是远程过程调用（Remote Procedure Call）的缩写形式。SAP系统RPC调用的原理其实很简单，有一些类似于三层构架的C/S系统，第三方的客户程序通过接口调用SAP内部的标准或自定义函数，获得函数返回的数据进行处理后显示或打印。

负载均衡原理

定义

负载均衡建立在现有网络结构之上，它提供了一种廉价有效透明的方法扩展网络设备和服务器的带宽、增加吞吐量、加强网络数据处理能力、提高网络的灵活性和可用性。
负载均衡（Load Balance）其意思就是分摊到多个操作单元上进行执行，例如Web服务器、FTP服务器、企业关键应用服务器和其它关键任务服务器等，从而共同完成工作任务。

负载均衡是一个通用的特性，所有的RPC框架都会有这个概念的实现。

平时说负载均衡一般都是指服务端负载均衡，但因为分布式spring cloud分布式框架出现，也出现了客户端负载均衡这一概念

服务端负载均衡

最常见的就是Nginx，客户端发送请求，由Nginx服务器接收，根据使用的负载均衡算法，再将请求发送给相应的应用服务器。

由Nginx分配到不同的服务端

客户端负载均衡

客户端的负载均衡是在spring-cloud后出现的，在spring-cloud中有ribbon组件来负责负载均衡。spring的负载均衡需要用到服务注册中心eruka。

服务提供方：将自己注册到服务注册中心eruka

服务消费方：从服务注册中心中获取服务列表，使用服务

客户端的负载均衡流程如下:

服务消费方通过ribbon先从服务注册中心获取服务列表，根据一定的负载均衡算法，分发请求到不同的服务提供方

客户端从服务中心选择一个，去调用

参考自：http://t.csdn.cn/zBP3x

常见的负载均衡算法

当然，我们可以去自定义负载均衡算法。

Ribbon负载均衡组件

Ribbon（spring-cloud-starter-ribbon）

我们都知道使用RestTemplate时，可以直接使用服务名进行服务调用，只需要在定义RestTemplate时加上@LoadBalanced注解就可以了

@LoadBalanced //配置负载均衡实现RestTemplate
@Bean
public RestTemplate getRestTemplate(){
    return new RestTemplate();
}

接下来，我们重点分析下@LoadBalanced这个注解底层都干了什么

@LoadBalanced

在自动配置类里我们可以看到LoadBalancerAutoConfiguration的存在。

LoadBalancerAutoConfiguration

我们先看下这个类的源码：

@Configuration
@ConditionalOnClass({RestTemplate.class})
@ConditionalOnBean({LoadBalancerClient.class})
@EnableConfigurationProperties({LoadBalancerRetryProperties.class})
public class LoadBalancerAutoConfiguration {
    @LoadBalanced
    @Autowired(
        required = false
    )
    //这里是从ApplicationContext中获取所有被@LoadBalanced修饰的RestTemplate
    private List<RestTemplate> restTemplates = Collections.emptyList();
    @Autowired(
        required = false
    )
    
    private List<LoadBalancerRequestTransformer> transformers = Collections.emptyList();

    public LoadBalancerAutoConfiguration() {
    }

    @Bean
    public SmartInitializingSingleton loadBalancedRestTemplateInitializerDeprecated(final ObjectProvider<List<RestTemplateCustomizer>> restTemplateCustomizers) {
    //这里的 List<RestTemplateCustomizer> 是 ApplicationContext 存在的 RestTemplateCustomizer集合
        return () -> {
            restTemplateCustomizers.ifAvailable((customizers) -> {
                Iterator var2 = this.restTemplates.iterator();

                while(var2.hasNext()) {
                    RestTemplate restTemplate = (RestTemplate)var2.next();
                    Iterator var4 = customizers.iterator();

                    while(var4.hasNext()) {
                        RestTemplateCustomizer customizer = (RestTemplateCustomizer)var4.next();
                        //遍历restTemplates集合，使用RestTemplateCustomizer给每个restTemplate定制一下
                        customizer.customize(restTemplate);
                    }
                }

            });
        };
    }
    ...

}

接着看这个RestTemplateCustomizer定制过程做了啥

RestTemplateCustomizer

RestTemplateCustomizer 是 LoadBalancerAutoConfiguration 的内部类

@Configuration
@ConditionalOnMissingClass({"org.springframework.retry.support.RetryTemplate"})
static class LoadBalancerInterceptorConfig {
    LoadBalancerInterceptorConfig() {
    }

	//定义LoadBalancerInterceptor Bean, 这个拦截器继承自 ClientHttpRequestInterceptor ,
	//可以被添加到RestTemplate的拦截器列表中
	//public class LoadBalancerInterceptor implements ClientHttpRequestInterceptor 
    @Bean
    public LoadBalancerInterceptor ribbonInterceptor(LoadBalancerClient loadBalancerClient, LoadBalancerRequestFactory requestFactory) {
        return new LoadBalancerInterceptor(loadBalancerClient, requestFactory);
    }

    @Bean
    @ConditionalOnMissingBean
    public RestTemplateCustomizer restTemplateCustomizer(final LoadBalancerInterceptor loadBalancerInterceptor) {
        return (restTemplate) -> {
            List<ClientHttpRequestInterceptor> list = new ArrayList(restTemplate.getInterceptors());
            //在RestTemplate的拦截器列表中加上LoadBalancerInterceptor拦截器
            list.add(loadBalancerInterceptor);
            restTemplate.setInterceptors(list);
        };
    }
}

这里就是 @LoadBalanced 直接修饰的 RestTemplate，会被机上一个LoadBalancerInterceptor拦截器

LoadBalancerInterceptor 拦截器

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) {
        this(loadBalancer, new LoadBalancerRequestFactory(loadBalancer));
    }

    public ClientHttpResponse intercept(final HttpRequest request, final byte[] body, final ClientHttpRequestExecution execution) throws IOException {
        // 服务名使用URI中的host信息
        URI originalUri = request.getURI();
        String serviceName = originalUri.getHost();
        Assert.state(serviceName != null, "Request URI does not contain a valid hostname: " + originalUri);
        //使用LoadBalancerClient 客户端负载均衡器做真正的服务调用
        return (ClientHttpResponse)this.loadBalancer.execute(serviceName, this.requestFactory.createRequest(request, body, execution));
    }
}

LoadBalancerClient

LoadBalancerClient (客户端负载均衡器)会根据负载均衡请求和服务名做真正的负载均衡。

public interface LoadBalancerClient extends ServiceInstanceChooser {
	//serviceId就是服务名，request就是请求
    <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException;

	// serviceInstance 服务实例
    <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException;

	//带有服务名的老的URI
    URI reconstructURI(ServiceInstance instance, URI original);
}

public interface LoadBalancerRequest<T> {
    T apply(ServiceInstance instance) throws Exception;
}

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 (instance) -> {
            HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance, this.loadBalancer);
            LoadBalancerRequestTransformer transformer;
            if (this.transformers != null) {
                for(Iterator var6 = this.transformers.iterator(); var6.hasNext(); serviceRequest = transformer.transformRequest((HttpRequest)serviceRequest, instance)) {
                    transformer = (LoadBalancerRequestTransformer)var6.next();
                }
            }

            return execution.execute((HttpRequest)serviceRequest, body);
        };
    }
}

RibbonLoadBalancerClient

Ribbon 中 LoadBalancerClient 的 默认实现类为 RibbonLoadBalancerClient

我们重点看下这个不带服务实例的execute 是如何 找到合适的 ServiceInstance
有一点可以明确，不带ServiceInstance 最后会去调用 带ServiceInstance的execute，上面也有所体现

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 <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint)
			throws IOException {
		//从这里开始的
		//通过serviceId 获取 ILoadBalancer 
		ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
		Server server = getServer(loadBalancer, hint);
		if (server == null) {
			throw new IllegalStateException("No instances available for " + serviceId);
		}
		//RibbonServer 是 ServiceInstance  的子类
		//public static class RibbonServer implements ServiceInstance {
		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();
		}
   }

通过serviceId 获取 ILoadBalancer

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

serviceId 也就是服务名

public ILoadBalancer getLoadBalancer(String name) {
		return getInstance(name, ILoadBalancer.class);
	}

@Override
	public <C> C getInstance(String name, Class<C> type) {
		C instance = super.getInstance(name, type);
		if (instance != null) {
			return instance;
		}
		IClientConfig config = getInstance(name, IClientConfig.class);
		return instantiateWithConfig(getContext(name), type, config);
	}

根据loadBalancer和hint（null）获取Service
因为上面hint为null，这里变成了default

protected Server getServer(ILoadBalancer loadBalancer, Object hint) {
		if (loadBalancer == null) {
			return null;
		}
		// Use 'default' on a null hint, or just pass it on?
		return loadBalancer.chooseServer(hint != null ? hint : "default");
	}

public Server chooseServer(Object key);

接着到BaseLoadBalancer

public Server chooseServer(Object key) {
        if (counter == null) {
            counter = createCounter();
        }
        counter.increment();
        if (rule == null) {
            return null;
        } else {
            try {
                //这里根据key进行选择
                return rule.choose(key);
            } catch (Exception e) {
                logger.warn("LoadBalancer [{}]:  Error choosing server for key {}", name, key, e);
                return null;
            }
        }
    }

public Server choose(Object key);

跟到这里我们就可以看到熟悉的负载均衡策略了，包括我们自定义的。
下面的就是具体的策略选择不同的服务了。

使用自定义规则

ConfigBean

package com.keafmd.springcloud.config;

import com.keafmd.myrule.MyRandomRule;
import com.netflix.loadbalancer.IRule;
import com.netflix.loadbalancer.RandomRule;
import com.netflix.loadbalancer.WeightedResponseTimeRule;
import org.springframework.cloud.client.loadbalancer.LoadBalanced;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.client.RestTemplate;

/**
 * Keafmd
 *
 * @ClassName: ConfigBean
 * @Description:
 * @author: 牛哄哄的柯南
 * @date: 2022-07-06 17:48
 */
@Configuration
public class ConfigBean {

    @LoadBalanced //配置负载均衡实现RestTemplate
    @Bean
    public RestTemplate getRestTemplate(){
        return new RestTemplate();
    }

    /**
     * IRule:
     * RoundRobinRule 轮询策略
     * RandomRule 随机策略
     * AvailabilityFilteringRule ： 会先过滤掉，跳闸，访问故障的服务~，对剩下的进行轮询~
     * RetryRule ： 会先按照轮询获取服务~，如果服务获取失败，则会在指定的时间内进行，重试
     */
    @Bean //注释掉 使用自定义规则
    public IRule myRule() {
        return new MyRandomRule();
//        return new RandomRule();//使用随机策略
        //return new RoundRobinRule();//使用轮询策略
        //return new AvailabilityFilteringRule();
        //return new RetryRule();
        //return new WeightedResponseTimeRule();
    }

}

MyRandomRule

继承 AbstractLoadBalancerRule

package com.keafmd.myrule;

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

import java.util.List;
import java.util.concurrent.ThreadLocalRandom;

/**
 * Keafmd
 *
 * @ClassName: MyRandomRule
 * @Description:
 * @author: 牛哄哄的柯南
 * @date: 2022-07-07 14:58
 */
public class MyRandomRule extends AbstractLoadBalancerRule {
    /**
     * 每个服务访问5次则换下一个服务(总共3个服务)
     * <p>
     * total=0,默认=0,如果=5,指向下一个服务节点
     * index=0,默认=0,如果total=5,index+1
     */
    private int total = 0;//被调用的次数
    private int currentIndex = 0;//当前是谁在提供服务
    //@edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "RCN_REDUNDANT_NULLCHECK_OF_NULL_VALUE")
    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) {
                /*
                 * No servers. End regardless of pass, because subsequent passes
                 * only get more restrictive.
                 */
                return null;
            }
            //int index = chooseRandomInt(serverCount);//生成区间随机数
            //server = upList.get(index);//从或活着的服务中,随机获取一个
            //=====================自定义代码=========================
            if (total < 5) {
                server = upList.get(currentIndex);
                total++;
            } else {
                total = 0;
                currentIndex++;
                if (currentIndex > upList.size()) {
                    currentIndex = 0;
                }
                server = upList.get(currentIndex);//从活着的服务中,获取指定的服务来进行操作
            }
            //======================================================
            if (server == null) {
                /*
                 * The only time this should happen is if the server list were
                 * somehow trimmed. This is a transient condition. Retry after
                 * yielding.
                 */
                Thread.yield();
                continue;
            }
            if (server.isAlive()) {
                return (server);
            }
            // Shouldn't actually happen.. but must be transient or a bug.
            server = null;
            Thread.yield();
        }
        return server;
    }
    protected int chooseRandomInt(int serverCount) {
        return ThreadLocalRandom.current().nextInt(serverCount);
    }
    @Override
    public Server choose(Object key) {
        return choose(getLoadBalancer(), key);
    }
    @Override
    public void initWithNiwsConfig(IClientConfig clientConfig) {
        // TODO Auto-generated method stub
    }
}

版权声明：
原创博主：牛哄哄的柯南
博主原文链接：https://keafmd.blog.csdn.net/
个人博客链接：https://www.keafmd.top/

看完如果对你有帮助，感谢点击下面的点赞支持！
[哈哈][抱拳]

加油！

共同努力！

Keafmd