前言
版本
作者：韩数
Github：https://github.com/hanshuaikang
完成日期：2019-06-16日
jdk：1.8
springboot版本：2.1.3.RELEASE
SpringCould版本：Greenwich.SR1
声明：
身为一个刚入门的计算机菜佬，阅读源码自然离不开参考书籍和视频的引导，本篇文章的分析过程中"严重"借鉴了 翟永超 前辈的《SpringCloud微服务实战》这本书籍，在这里也向准备学习微服务的小伙伴们强烈推荐这本书，大家可以把这篇文章理解为《SpringCloud微服务实战》Ribbon部分的精简版和电子版，因为个人水平的原因，很多问题不敢妄下定论，以免误人子弟，所有书上很多内容都是精简过后直接放上去的，由于SpringCloud已经迭代到了Greenwich.SR1版本，Ribbon也和书上有了略微的差别，本篇文章的源码采用的是Ribbon新版本，同时，因为时间原因，有很多额外的子类实现并没有完全顾上，例如PredicateBasedRule类的ZoneAvoidanceRule和AvailabilityFilteringRule 感兴趣的读者可以买《SpringCloud微服务实战》这本书细看，同时强烈推荐小马哥的微服务直播课系列《小马哥微服务实战》。
致谢
翟永超：博客地址：
http://blog.didispace.com/aboutme/
小马哥： Java 微服务实践 - Spring Boot / Spring Cloud购买链接：
https://segmentfault.com/ls/1650000011387052
电子版及相关代码下载（欢迎Star）
Github：https://github.com/hanshuaikang/Spring-Note
微信公众号：码上marson
负载均衡策略
通过上面的分析，我们发现当一个请求过来时，会被拦截交给相应的负载均衡器，然后不同的负载均衡器根据不同的策略来选择合适的服务实例。在这里我们是知道Ribbon是根据不同的Rule来实现对实例的一个选择的，那么Ribbon具体提供了哪些规则供我们使用呢？通过查看Ribbon的IRule接口的实现集成关系图，我们终可以发现，Ribbon主要提供了以下几个规则实现的。
- RandomRule 类：该策略实现了从服务实例清单中随机选择一个服务实例的功能
- RoundRobinRule类：该策略实现了轮询的方式从服务实例清单中依次选择服务实例的功能RetryRule
- RetryRule类：该策略实现了具备重试机制的实例选择功能
- WeightedResponseTimeRule类：根据权重来选择实例
- BestAvailableRule类：选择一个空闲的实例
- PredicateBasedRule 类：先过滤，然后再以轮询的方式选择实例
...
IRule接口:
public interface IRule{
public Server choose(Object key);
public void setLoadBalancer(ILoadBalancer lb);
public ILoadBalancer getLoadBalancer();
}
AbstractLoadBalancerRule抽象类:
public abstract class AbstractLoadBalancerRule implements IRule, IClientConfigAware {
private ILoadBalancer lb;
@Override
public void setLoadBalancer(ILoadBalancer lb){
this.lb = lb;
}
@Override
public ILoadBalancer getLoadBalancer(){
return lb;
}
}RandomRule类
功能：该策略实现了从服务实例清单中随机选择一个服务实例的功能。
查看代码发现具体的实例选择并没有由默认的choose(Object key)来实现，而是委托给了同类下的choose(ILoadBalancer lb, Object key)方法来完成实际的实例选择工作。
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 == ) {
/*
* No servers. End regardless of pass, because subsequent passes
* only get more restrictive.
*/
return null;
}
int index = chooseRandomInt(serverCount);
server = upList.get(index);
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;
}
注：如果获取不到服务实例，则可能存在并发的bug
RoundRobinRule类
功能：该策略实现了轮询的方式从服务实例清单中依次选择服务实例的功能
public Server choose(ILoadBalancer lb, Object key) {
if (lb == null) {
log.warn("no load balancer");
return null;
}
Server server = null;
int count = ;
while (server == null && count++ < 10) {
//reachableServers 可用的服务实例清单
List<Server> reachableServers = lb.getReachableServers();
//allServers 获取所有可用的服务列表
List<Server> allServers = lb.getAllServers();
int upCount = reachableServers.size();
int serverCount = allServers.size();
if ((upCount == ) || (serverCount == )) {
log.warn("No up servers available from load balancer: " + lb);
return null;
}
int nextServerIndex = incrementAndGetModulo(serverCount);
server = allServers.get(nextServerIndex);
if (server == null) {
/* Transient. */
Thread.yield();
continue;
}
if (server.isAlive() && (server.isReadyToServe())) {
return (server);
}
// Next.
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;
}
}
源码分析：可以发现RoundRobinRule的实现逻辑和RandomRule非常类似，我们可以看出来，RoundRobinRule定义了一个计数器变量count，该计数器会在每次循环后自动叠加，当获取不到Server的次数超过十次时，会结束尝试，并发出警告：No available alive servers after 10 tries from load balancer。
而线性轮询的实现则是通过 incrementAndGetModulo(int modulo)来实现的.
RetryRule类:
功能：该策略实现了具备重试机制的实例选择功能
public Server choose(ILoadBalancer lb, Object key) {
//请求时间
long requestTime = System.currentTimeMillis();
//deadline 截止期限
long deadline = requestTime + maxRetryMillis;
Server answer = null;
answer = subRule.choose(key);
if (((answer == null) || (!answer.isAlive()))
&& (System.currentTimeMillis() < deadline)) {
InterruptTask task = new InterruptTask(deadline
- System.currentTimeMillis());
while (!Thread.interrupted()) {
answer = subRule.choose(key);
if (((answer == null) || (!answer.isAlive()))
&& (System.currentTimeMillis() < deadline)) {
/* pause and retry hoping it's transient */
Thread.yield();
} else {
break;
}
}
task.cancel();
}
if ((answer == null) || (!answer.isAlive())) {
return null;
} else {
return answer;
}
}
默认使用的是RoundRobinRule策略。期间如果能选择到实例就返回，如果选择不到就根据设置的尝试结束时间为阈值，如果超过截止期限则直接返回null。
WeightedResponseTimeRule类
功能：根据权重来选择实例
主要有以下三个核心内容：
- 定时任务
- 权重计算
- 实例选择
1. 定时任务
void initialize(ILoadBalancer lb) {
if (serverWeightTimer != null) {
serverWeightTimer.cancel();
}
serverWeightTimer = new Timer("NFLoadBalancer-serverWeightTimer-"
+ name, true);
//启动定时任务
serverWeightTimer.schedule(new DynamicServerWeightTask(), ,
serverWeightTaskTimerInterval);
// do a initial run
ServerWeight sw = new ServerWeight();
sw.maintainWeights();
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
public void run() {
logger
.info("Stopping NFLoadBalancer-serverWeightTimer-"
+ name);
serverWeightTimer.cancel();
}
}));
}
WeightedResponseTimeRule类在初始化的时候会先定义一个计时器，然后会启动一个定时任务，用来为每个服务实例计算权重，该任务默认每30秒执行一次。
class DynamicServerWeightTask extends TimerTask {
public void run() {
ServerWeight serverWeight = new ServerWeight();
try {
serverWeight.maintainWeights();
} catch (Exception e) {
logger.error("Error running DynamicServerWeightTask for {}", name, e);
}
}
}2.权重计算
通过上面的DynamicServerWeightTask的代码呢，我们可以大致了解到，权重计算的功能呢实际是由ServerWeight的maintainWeights()来执行的。少废话，上代码。
public void maintainWeights() {
ILoadBalancer lb = getLoadBalancer();
if (lb == null) {
return;
}
if (!serverWeightAssignmentInProgress.compareAndSet(false, true)) {
return;
}
try {
logger.info("Weight adjusting job started");
AbstractLoadBalancer nlb = (AbstractLoadBalancer) lb;
LoadBalancerStats stats = nlb.getLoadBalancerStats();
if (stats == null) {
// no statistics, nothing to do
return;
}
double totalResponseTime = ;
// find maximal 95% response time
for (Server server : nlb.getAllServers()) {
// this will automatically load the stats if not in cache
ServerStats ss = stats.getSingleServerStat(server);
totalResponseTime += ss.getResponseTimeAvg();
}
// weight for each server is (sum of responseTime of all servers - responseTime)
// so that the longer the response time, the less the weight and the less likely to be chosen
Double weightSoFar = .;
// create new list and hot swap the reference
List<Double> finalWeights = new ArrayList<Double>();
for (Server server : nlb.getAllServers()) {
ServerStats ss = stats.getSingleServerStat(server);
double weight = totalResponseTime - ss.getResponseTimeAvg();
weightSoFar += weight;
finalWeights.add(weightSoFar);
}
setWeights(finalWeights);
} catch (Exception e) {
logger.error("Error calculating server weights", e);
} finally {
serverWeightAssignmentInProgress.set(false);
}
}
}
那WeightedResponseTimeRule是如何计算权重的呢？主要分为以下两步：
- 先遍历服务器列表，并得到每个服务器的平均响应时间，遍历过程中对其求和，遍历结束后得到总响应时间totalResponseTime。
- 再一次遍历服务器列表，并将总响应时间totalResponseTime减去每个服务器的平均响应时间作为权重weight，再将这之前的所以权重累加到weightSoFar 变量中，并且保存到finalWeights供choose使用。
3.实例选择
public Server choose(ILoadBalancer lb, Object key) {
if (lb == null) {
return null;
}
Server server = null;
while (server == null) {
//获取当前引用，以防它被其他线程更改
List<Double> currentWeights = accumulatedWeights;
if (Thread.interrupted()) {
return null;
}
List<Server> allList = lb.getAllServers();
int serverCount = allList.size();
if (serverCount == ) {
return null;
}
int serverIndex = ;
// 列表中的后一个是所有权重的和
double maxTotalWeight = currentWeights.size() == ? : currentWeights.get(currentWeights.size() - 1);
//尚未命中任何服务器，且未初始化总重量
//使用循环操作
if (maxTotalWeight < .001d || serverCount != currentWeights.size()) {
server = super.choose(getLoadBalancer(), key);
if(server == null) {
return server;
}
} else {
//生成一个从0(含)到maxTotalWeight(不含)之间的随机权重
double randomWeight = random.nextDouble() * maxTotalWeight;
//根据随机索引选择服务器索引
int n = ;
for (Double d : currentWeights) {
if (d >= randomWeight) {
serverIndex = n;
break;
} else {
n++;
}
}
server = allList.get(serverIndex);
}
if (server == null) {
/* Transient. */
Thread.yield();
continue;
}
if (server.isAlive()) {
return (server);
}
// Next.
server = null;
}
return server;
}执行步骤：
- 生成一个从0(含)到maxTotalWeight(不含)之间的随机权重
- 遍历权重列表，比较权重值与随机数的大小，如果权重值大于等于随机数，就当前权重列表的索引值去服务实例列表中列表中获取具体的实例。
BestAvailableRule类
功能：选择一个空闲的实例
@Override
public Server choose(Object key) {
if (loadBalancerStats == null) {
return super.choose(key);
}
List<Server> serverList = getLoadBalancer().getAllServers();
//minimalConcurrentConnections：小并发连接数
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)) {
//concurrentConnections:并发连接数
int concurrentConnections = serverStats.getActiveRequestsCount(currentTime);
if (concurrentConnections < minimalConcurrentConnections) {
minimalConcurrentConnections = concurrentConnections;
chosen = server;
}
}
}
if (chosen == null) {
return super.choose(key);
} else {
return chosen;
}
}通过查看源码可以得知BestAvailableRule大致采用了如下策略来选择服务实例，根据loadBalancerStats中的统计信息通过遍历负载均衡器维护的所有服务实例 选出并发连接数少的那一个，即空闲的实例。
如果loadBalancerStats为空的话，则直接调用父类ClientConfigEnabledRoundRobinRule的实现，即RoundRobinRule，线性轮询的方式。
PredicateBasedRule 类
功能：先过滤，然后再以轮询的方式选择实例
@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;
}
}实现逻辑：通过子类中实现的predicate逻辑来过滤一部分服务实例，然后再以线性轮询的方式从过滤之后的服务实例清单中选择一个。
当然，PredicateBasedRule本身是一个抽象类，必然Ribbon提供了相应的子类实现，我们看到有ZoneAvoidanceRule和AvailabilityFilteringRule，分别对PredicateBasedRule做了相应的扩展，有兴趣的小伙伴可以下去自行研究。
配置详解：
自动化配置：
同样，得益于Springboot的自动化配置，大大降低了开发者上手的难度，在引入Spring-Clould-Ribbon依赖之后，便能够自动构建下面这些接口的实现。
- IClientConfig：Ribbon客户端配置接口类，默认实现：com.netflix.client.config.DefaultClientConfigImpl
- IRule: Ribbon：服务实例选择策略接口类，默认采用的实现：com.netflix.loadbalancer.ZoneAvoidanceRule
- IPing:Ribbon：实例检查策略接口类,默认实现：NoOpPing 即不检查
- ServerList<T extends Server>：服务实例清单维护机制接口类，默认实现ConfigurationBasedServerList 当整合Eureka的情况下，则使用DiscoveryEnabledNIWSServerList类
- ServerListFilter<T extends Server>：服务实例过滤策略接口类，默认实现：ZoneAffinityServerListFilter 根据区域过滤，
- ILoadBalancer：负载均衡器接口类，默认实现：ZoneAwareLoadBalancer 具备区域感知
替换默认配置
Ribbon同时支持部分默认配置的替换，这为使用针对不同场景的定制化方案提供了可能。目前的话支持两种方式的替换（我只知道这两种）。
- 创建实例覆盖默认实现
- 配置文件配置
创建实例覆盖默认实现
例：将默认的负载均衡策略替换成自己自定义的策略。
@Bean
public IRule myRule() {
return new MyRule();
}配置文件配置
通过使用<service-name>.ribbon.<key> = value 方式
在application.properties中添加如下代码，即可以将默认的IPing策略替换成自己自定义的策略。
### 扩展 IPing 实现
user-service-provider.ribbon.NFLoadBalancerPingClassName = \
com.xxxx.demo.user.ribbon.client.ping.MyPingMyPing代码（小马哥微服务实战版）：
public class MyPing implements IPing {
@Override
public boolean isAlive(Server server) {
String host = server.getHost();
int port = server.getPort();
// /health endpoint
// 通过 Spring 组件来实现URL 拼装
UriComponentsBuilder builder = UriComponentsBuilder.newInstance();
builder.scheme("http");
builder.host(host);
builder.port(port);
builder.path("/actuator/health");
URI uri = builder.build().toUri();
RestTemplate restTemplate = new RestTemplate();
ResponseEntity responseEntity = restTemplate.getForEntity(uri, String.class);
// 当响应状态等于 200 时，返回 true ，否则 false
return HttpStatus.OK.equals(responseEntity.getStatusCode());
}
}
MyRule代码（小马哥微服务实战版）：
public class MyRule extends AbstractLoadBalancerRule {
@Override
public void initWithNiwsConfig(IClientConfig clientConfig) {
}
@Override
public Server choose(Object key) {
ILoadBalancer loadBalancer = getLoadBalancer();
//获取所有可达服务器列表
List<Server> servers = loadBalancer.getReachableServers();
if (servers.isEmpty()) {
return null;
}
// 永远选择后一台可达服务器
Server targetServer = servers.get(servers.size() - 1);
return targetServer;
}
}
总结：
通过本次对Ribbon源码的一个简单初探，慢慢明白一个的框架的之处了，再看看自己之前写的代码就有些难以直视了，一个框架的设计往往不仅仅是实现了某些功能，也同时考虑到了各种不同的使用场景，这样可以保证框架可以胜任大多数简单的项目和大型项目。同时框架内部有很多实现都很高效，很少出现有什么极度不合理的地方，同时代码复用性也很高，看似几十上百个类实则职责分明，井井有条，在保证功能的情况下同时又有良好的扩展性。因为平常学业繁忙（主要是懒还爱玩儿），刻苦学习（期末全靠水过去），所以Ribbon这篇磕磕绊绊写了有半个多月的时间。好在自己终于坚持把它给看完了。后面的打算呢，将会陆续把自己学习java微服务的笔记整理好开源至本人的github上，希望可以帮助到一些刚开始入门的小伙伴们，也骗一些star（滑稽），后，我是韩数，计算机小白，本科在读，我喜欢唱，跳...
