用 Java 代码实现负载均衡的五种常见算法
在几年前，负载均衡还是“高端玩家”的游戏，我这种小白还难以触及，现在“负载均衡”已经有点看似烂大街的趋势了。
提起负载均衡，首先要理解负载均衡到底是想解决什么样的问题，维基百科有这么一段描述：
主要作用是将大量作业合理地分摊到多个操作单元上进行执行，用于解决互联网架构中的高并发和高可用的问题。
在现在的互联网系统中，为了避免出现单点问题，最常见的做法就是将系统部署到多台机器上，也就是集群。但是这样会出现两个问题：
-
如何让机器均衡或者相对均衡的接收到到流量； -
当集群的某个节点宕机，让流量不会打到这个节点上；
负载均衡技术就是来解决这两个问题。但这时候还要考虑两个问题：
-
避免负载均衡系统成为单点； -
负载均衡系统是否需要关注服务器自身的状态差异；
第一点比如可以基于 standby 模式搭建负载均衡系统集群，现在也有基于 Gossip 实现去中心化的软件负载解决方案。
关于第二点，首先当服务器宕机或者与负载均衡系统连接中断的时候，负载均衡系统是必须要能够感知到的。但是在负载均衡算法中是否需要加入服务器自身状态的指标也是值得考虑的，如根据负载均衡系统和服务器之间的连接数、服务器的 CPU 负载、I/O 负载、服务器的响应时间等，但是这样负载均衡系统需要接收服务器的指标状态，而指标收集可能还会涉及到收集的时间节点，这会大大增加系统的复杂度。
真正的大型高并发系统负载均衡是非常复杂的，比如还需要考虑系统预热的情况，要想深入了解还是需要多看分布式相关的书籍和进行实践。本文主要是通过代码实现几种简单的算法，借此体会负载均衡的思想。
先定义一个 Invoker，存储一些基本的信息，一般来说这个类会抽象处理：
package dongguabai.demo;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import lombok.ToString;
/**
* @author Dongguabai
* @Description
* @Date 创建于 2020-06-24 09:16
*/
@NoArgsConstructor
@Getter
@Setter
@ToString
@EqualsAndHashCode(of = "address")
public class Invoker {
private String address;
/**
* 权重
*/
private int weight = 1;
private int currentWeight = 0;
public Invoker(String address) {
this.address = address;
}
public Invoker(String address, int weight) {
this.address = address;
this.weight = weight;
}
}
定义一个抽象类：
package dongguabai.demo;
import com.google.common.collect.HashMultiset;
import com.google.common.collect.Multiset;
import org.springframework.util.CollectionUtils;
import java.util.Comparator;
import java.util.List;
import java.util.Set;
import java.util.stream.IntStream;
/**
* @author Dongguabai
* @Description 抽象类
* @Date 创建于 2020-06-23 13:41
*/
public abstract class AbstractLoadBalance {
protected List<Invoker> invokers;
public AbstractLoadBalance(List<Invoker> invokers) {
this.invokers = invokers;
}
protected final Multiset<Invoker> results = HashMultiset.create();
public Invoker selectHost(){
List<Invoker> list = getInvokers();
if (CollectionUtils.isEmpty(list)) {
return null;
}
if (list.size() == 1) {
return list.get(0);
}
return doSelect();
}
public Invoker selectHost(Object client){
List<Invoker> list = getInvokers();
if (CollectionUtils.isEmpty(list)) {
return null;
}
if (list.size() == 1) {
return list.get(0);
}
return doSelect(client);
}
protected Invoker doSelect(){
return null;
}
protected Invoker doSelect(Object client){
return null;
}
//todo 优化refesh
protected boolean addInvoker(Invoker invoker){
if (getInvokers() != null){
return invokers.add(invoker);
}
return false;
}
//todo 优化refesh
protected boolean removeInvoker(Invoker invoker){
if (getInvokers() != null){
return invokers.remove(invoker);
}
return false;
}
//todo 不要返回集合
protected List<Invoker> getInvokers(){
return invokers;
}
public void result(int loop) {
results.clear();
if (loop < 1) {
throw new IllegalArgumentException();
}
IntStream.range(0, loop).forEach(i -> results.add(selectHost()));
Set<Multiset.Entry<Invoker>> entrySet = results.entrySet();
entrySet.stream().sorted(Comparator.comparingInt(Multiset.Entry::getCount)).forEach(System.out::println);
}
}
完全随机
随机算法是用得比较多，也是比较简单的一种算法，属于任务数量的绝对平分。在调用次数比较多的时候会相对平均分布到每台机器上。
package dongguabai.demo.random;
import dongguabai.demo.AbstractLoadBalance;
import dongguabai.demo.Invoker;
import java.util.List;
import java.util.Random;
/**
* @author Dongguabai
* @Description 随机
* @Date 创建于 2020-06-23 18:57
*/
public class SimpleRandomLoadBalance extends AbstractLoadBalance {
public SimpleRandomLoadBalance(List<Invoker> invokers) {
super(invokers);
}
@Override
protected Invoker doSelect() {
return getInvokers().get(new Random().nextInt(getInvokers().size()));
}
@Override
public List<Invoker> getInvokers() {
return invokers;
}
}
测试：
@Test
public void testSimpleRandomLoadBalance(){
List<Invoker> invokers = Lists.newArrayList(
new Invoker("176.170.209.1"),
new Invoker("176.170.209.2"),
new Invoker("176.170.209.3"),
new Invoker("176.170.209.4"),
new Invoker("176.170.209.5"),
new Invoker("176.170.209.6"),
new Invoker("176.170.209.7"),
new Invoker("176.170.209.8"),
new Invoker("176.170.209.9"),
new Invoker("176.170.209.10"));
AbstractLoadBalance randomLoadBalance = new SimpleRandomLoadBalance(invokers);
randomLoadBalance.result(20000);
}
运行结果：
Invoker(address=176.170.209.8, weight=1) x 1933
Invoker(address=176.170.209.9, weight=1) x 1953
Invoker(address=176.170.209.10, weight=1) x 1953
Invoker(address=176.170.209.7, weight=1) x 1983
Invoker(address=176.170.209.3, weight=1) x 2001
Invoker(address=176.170.209.6, weight=1) x 2003
Invoker(address=176.170.209.2, weight=1) x 2017
Invoker(address=176.170.209.5, weight=1) x 2027
Invoker(address=176.170.209.1, weight=1) x 2058
Invoker(address=176.170.209.4, weight=1) x 2072
样本数量越大，越平均。
加权随机
实际线上可能机器的性能各不相同，为了让性能更好的机器分配更多的流量，可以为每台机器设置一个权重，按照一定的比例进行任务的均分。
package dongguabai.demo.random;
import dongguabai.demo.Invoker;
import dongguabai.demo.random.SimpleRandomLoadBalance;
import org.springframework.util.CollectionUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.IntStream;
/**
* @author Dongguabai
* @Description 加权随机（复制）
* @Date 创建于 2020-06-24 10:30
*/
public class CopyWeightRandomLoadBalance extends SimpleRandomLoadBalance {
private List<Invoker> weightInvokers = new ArrayList<>();
public CopyWeightRandomLoadBalance(List<Invoker> invokers) {
super(invokers);
if (!CollectionUtils.isEmpty(invokers)){
invokers.forEach(invoker -> {
if (invoker.getWeight() < 2) {
weightInvokers.add(invoker);
return;
}
IntStream.range(0,invoker.getWeight()).forEach(i->weightInvokers.add(invoker));
});
}
}
@Override
public final List<Invoker> getInvokers() {
return weightInvokers;
}
}
测试：
@Test
public void testCopyWeightRandomLoadBalance(){
List<Invoker> invokers = Lists.newArrayList(
new Invoker("176.170.209.1",1),
new Invoker("176.170.209.2",9),
new Invoker("176.170.209.3",1),
new Invoker("176.170.209.4",9),
new Invoker("176.170.209.5",4),
new Invoker("176.170.209.6",6),
new Invoker("176.170.209.7",1),
new Invoker("176.170.209.8",9),
new Invoker("176.170.209.9",1),
new Invoker("176.170.209.10",9));
AbstractLoadBalance simpleWeightRandomLoadBalance1 = new CopyWeightRandomLoadBalance(invokers);
simpleWeightRandomLoadBalance1.result(20000);
}
运行结果：
Invoker(address=176.170.209.1, weight=1) x 376
Invoker(address=176.170.209.3, weight=1) x 386
Invoker(address=176.170.209.7, weight=1) x 397
Invoker(address=176.170.209.9, weight=1) x 398
Invoker(address=176.170.209.5, weight=4) x 1559
Invoker(address=176.170.209.6, weight=6) x 2380
Invoker(address=176.170.209.8, weight=9) x 3567
Invoker(address=176.170.209.10, weight=9) x 3622
Invoker(address=176.170.209.4, weight=9) x 3632
Invoker(address=176.170.209.2, weight=9) x 3683
可以发现是按照权重比例进行分配。
但是这种算法需要对调用信息进行复制，当数量较大时对内存的消耗相对是比较大的，所以一般会使用另外一种方式实现。
还是以这个权重为例：
List<Invoker> invokers = Lists.newArrayList(
new Invoker("176.170.209.1",1),
new Invoker("176.170.209.2",9),
new Invoker("176.170.209.3",1),
new Invoker("176.170.209.4",9),
new Invoker("176.170.209.5",4),
new Invoker("176.170.209.6",6),
new Invoker("176.170.209.7",1),
new Invoker("176.170.209.8",9),
new Invoker("176.170.209.9",1),
new Invoker("176.170.209.10",9));
首先计算出总的权重，我这里是 50，生成 [0,50) 的随机数 n；根据这个权重为每一个 Invoker 分配一个区间：
0，1，10，11，20，24，30，31，40，41，50
这个区间包左不包右，比如生成的随机数 0<= n <1，那么就是“176.170.209.1”这个机器，以此类推。
具体实现如下：
package dongguabai.demo.random;
import dongguabai.demo.Invoker;
import java.util.List;
import java.util.Random;
/**
* @author Dongguabai
* @Description
* @Date 创建于 2020-06-24 17:10
*/
public class SectionWeightRandomLoadBalance extends SimpleRandomLoadBalance {
private boolean averageWeight = true;
private int totalWeight;
public SectionWeightRandomLoadBalance(List<Invoker> invokers) {
super(invokers);
for (int i = 0; i < invokers.size(); i++) {
Invoker invoker = invokers.get(i);
if (averageWeight && i > 0 && invoker.getWeight() != invokers.get(i - 1).getWeight()) {
averageWeight = false;
}
totalWeight += invoker.getWeight();
}
@Override
protected Invoker doSelect() {
if (averageWeight || totalWeight < 1) {
return super.doSelect();
}
int index = new Random().nextInt(totalWeight);
for (Invoker invoker : invokers) {
if (index < invoker.getWeight()) {
return invoker;
}
index -= invoker.getWeight();
}
return super.doSelect();
}
}
测试：
@Test
public void testWeightRandomLoadBalance(){
List<Invoker> invokers = Lists.newArrayList(
new Invoker("176.170.209.1",1),
new Invoker("176.170.209.2",9),
new Invoker("176.170.209.3",1),
new Invoker("176.170.209.4",9),
new Invoker("176.170.209.5",4),
new Invoker("176.170.209.6",6),
new Invoker("176.170.209.7",1),
new Invoker("176.170.209.8",9),
new Invoker("176.170.209.9",1),
new Invoker("176.170.209.10",9));
AbstractLoadBalance simpleWeightRandomLoadBalance1 = new SectionWeightRandomLoadBalance(invokers);
simpleWeightRandomLoadBalance1.result(20000);
}
运行结果：
Invoker(address=176.170.209.3, weight=1) x 386
Invoker(address=176.170.209.9, weight=1) x 405
Invoker(address=176.170.209.7, weight=1) x 411
Invoker(address=176.170.209.1, weight=1) x 430
Invoker(address=176.170.209.5, weight=4) x 1598
Invoker(address=176.170.209.6, weight=6) x 2450
Invoker(address=176.170.209.4, weight=9) x 3543
Invoker(address=176.170.209.10, weight=9) x 3557
Invoker(address=176.170.209.2, weight=9) x 3573
Invoker(address=176.170.209.8, weight=9) x 3647
轮询
轮询算法也很好理解，就是轮流执行每台服务器，所以有一个请求序号的概念。
package dongguabai.demo.roundrobin;
import dongguabai.demo.AbstractLoadBalance;
import dongguabai.demo.Invoker;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @author Dongguabai
* @Description
* @Date 创建于 2020-06-25 13:48
*/
public class SimpleRoundRobinLoadBalance extends AbstractLoadBalance {
protected AtomicInteger offset = new AtomicInteger(-1);
public SimpleRoundRobinLoadBalance(List<Invoker> invokers) {
super(invokers);
}
@Override
protected Invoker doSelect() {
return getInvokers().get(offset.addAndGet(1) % getInvokers().size());
}
}
测试：
@Test
public void testSimpleRoundRobinLoadBalance(){
List<Invoker> invokers = Lists.newArrayList(
new Invoker("176.170.209.1"),
new Invoker("176.170.209.2"),
new Invoker("176.170.209.3"),
new Invoker("176.170.209.4"),
new Invoker("176.170.209.5"),
new Invoker("176.170.209.6"),
new Invoker("176.170.209.7"),
new Invoker("176.170.209.8"),
new Invoker("176.170.209.9"),
new Invoker("176.170.209.10"));
AbstractLoadBalance roundRobinLoadBalance = new SimpleRoundRobinLoadBalance(invokers);
IntStream.range(0,22).parallel().forEach(i-> System.out.println(roundRobinLoadBalance.selectHost()));
}
执行结果：
Invoker(address=176.170.209.7, weight=1)
Invoker(address=176.170.209.5, weight=1)
Invoker(address=176.170.209.9, weight=1)
Invoker(address=176.170.209.2, weight=1)
Invoker(address=176.170.209.6, weight=1)
Invoker(address=176.170.209.8, weight=1)
Invoker(address=176.170.209.3, weight=1)
Invoker(address=176.170.209.4, weight=1)
Invoker(address=176.170.209.1, weight=1)
Invoker(address=176.170.209.4, weight=1)
Invoker(address=176.170.209.8, weight=1)
Invoker(address=176.170.209.9, weight=1)
Invoker(address=176.170.209.10, weight=1)
Invoker(address=176.170.209.7, weight=1)
Invoker(address=176.170.209.2, weight=1)
Invoker(address=176.170.209.6, weight=1)
Invoker(address=176.170.209.5, weight=1)
Invoker(address=176.170.209.3, weight=1)
Invoker(address=176.170.209.2, weight=1)
Invoker(address=176.170.209.1, weight=1)
Invoker(address=176.170.209.10, weight=1)
Invoker(address=176.170.209.1, weight=1)
加权轮询
有一个方式是根据权重复制，同上文介绍的加权随机的第一种方式。
package dongguabai.demo.roundrobin;
import dongguabai.demo.Invoker;
import org.springframework.util.CollectionUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.IntStream;
/**
* @author Dongguabai
* @Description
* @Date 创建于 2020-06-25 17:33
*/
public class CopyWeightRoundRobinLoadBalance extends SimpleRoundRobinLoadBalance{
private List<Invoker> weightInvokers = new ArrayList<>();
public CopyWeightRoundRobinLoadBalance(List<Invoker> invokers) {
super(invokers);
if (!CollectionUtils.isEmpty(invokers)){
invokers.forEach(invoker -> {
if (invoker.getWeight() < 2) {
weightInvokers.add(invoker);
return;
}
IntStream.range(0,invoker.getWeight()).forEach(i->weightInvokers.add(invoker));
});
}
}
@Override
public final List<Invoker> getInvokers() {
return weightInvokers;
}
}
测试：
@Test
public void testCopyWeightRoundRobinLoadBalance(){
List<Invoker> invokers = Lists.newArrayList(
new Invoker("176.170.209.1",1),
new Invoker("176.170.209.2",9),
new Invoker("176.170.209.3",1),
new Invoker("176.170.209.4",9),
new Invoker("176.170.209.5",4),
new Invoker("176.170.209.6",6),
new Invoker("176.170.209.7",1),
new Invoker("176.170.209.8",9),
new Invoker("176.170.209.9",1),
new Invoker("176.170.209.10",9));
AbstractLoadBalance copyWeightRoundRobinLoadBalance = new CopyWeightRoundRobinLoadBalance(invokers);
IntStream.range(0,50).forEach(i-> System.out.println(copyWeightRoundRobinLoadBalance.selectHost()));
// copyWeightRoundRobinLoadBalance.result(20000);
}
输出结果：
Invoker(address=176.170.209.1, weight=1)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.3, weight=1)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.5, weight=4)
Invoker(address=176.170.209.5, weight=4)
Invoker(address=176.170.209.5, weight=4)
Invoker(address=176.170.209.5, weight=4)
Invoker(address=176.170.209.6, weight=6)
Invoker(address=176.170.209.6, weight=6)
Invoker(address=176.170.209.6, weight=6)
Invoker(address=176.170.209.6, weight=6)
Invoker(address=176.170.209.6, weight=6)
Invoker(address=176.170.209.6, weight=6)
Invoker(address=176.170.209.7, weight=1)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.9, weight=1)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
上面是通过复制进行加权，同随机加权一样，还有一种通过权重区间进行轮询。
package dongguabai.demo.roundrobin;
import dongguabai.demo.Invoker;
import java.util.List;
/**
* @author Dongguabai
* @Description
* @Date 创建于 2020-06-27 14:22
*/
public class SectionWeightRoundRobinLoadBalance extends SimpleRoundRobinLoadBalance {
private boolean averageWeight = true;
private int totalWeight;
public SectionWeightRoundRobinLoadBalance(List<Invoker> invokers) {
super(invokers);
for (int i = 0; i < invokers.size(); i++) {
Invoker invoker = invokers.get(i);
if (averageWeight && i > 0 && invoker.getWeight() != invokers.get(i - 1).getWeight()) {
averageWeight = false;
}
totalWeight += invoker.getWeight();
}
}
@Override
protected Invoker doSelect() {
if (averageWeight || totalWeight < 1) {
return super.doSelect();
}
int index = offset.addAndGet(1) % totalWeight;
for (Invoker invoker : invokers) {
if (index < invoker.getWeight()) {
return invoker;
}
index -= invoker.getWeight();
}
return super.doSelect();
}
}
测试：
@Test
public void testSectionWeightRoundRobinLoadBalance(){
List<Invoker> invokers = Lists.newArrayList(
new Invoker("176.170.209.1",1),
new Invoker("176.170.209.2",9),
new Invoker("176.170.209.3",1),
new Invoker("176.170.209.4",9),
new Invoker("176.170.209.5",4),
new Invoker("176.170.209.6",6),
new Invoker("176.170.209.7",1),
new Invoker("176.170.209.8",9),
new Invoker("176.170.209.9",1),
new Invoker("176.170.209.10",9));
AbstractLoadBalance sectionWeightRoundRobinLoadBalance = new SectionWeightRoundRobinLoadBalance(invokers);
IntStream.range(0,50).forEach(i-> System.out.println(sectionWeightRoundRobinLoadBalance.selectHost()));
}
运行结果：
Invoker(address=176.170.209.1, weight=1)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.2, weight=9)
Invoker(address=176.170.209.3, weight=1)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.4, weight=9)
Invoker(address=176.170.209.5, weight=4)
Invoker(address=176.170.209.5, weight=4)
Invoker(address=176.170.209.5, weight=4)
Invoker(address=176.170.209.5, weight=4)
Invoker(address=176.170.209.6, weight=6)
Invoker(address=176.170.209.6, weight=6)
Invoker(address=176.170.209.6, weight=6)
Invoker(address=176.170.209.6, weight=6)
Invoker(address=176.170.209.6, weight=6)
Invoker(address=176.170.209.6, weight=6)
Invoker(address=176.170.209.7, weight=1)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.8, weight=9)
Invoker(address=176.170.209.9, weight=1)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
Invoker(address=176.170.209.10, weight=9)
平滑加权轮询
上面两种轮询的实现有一个很大的问题就是会造成某台权重较高的机器在短时间内负载较高，其实轮询我们希望的是，比如 A(1),B(2),C(3) 三台机器我们希望的是每 6 次调用，有 1 次是 A，2 次是 B，3 次是 C 即可。当然也可以进行改动，比如复制法可以将复制后的机器列表打乱。
在 Nginx 中有一种平滑加权的轮询算法。在上面的其他权重算法中，每台机器会有一个权重 weight 属性，这个属性是配置好之后就不变的，在平滑加权轮询中每台机器还有一个当前权重的属性 currentWeight，这个属性是会变化的。比如 A(1),B(2),C(3) 三台机器，它们的初始化 currentWeight 都是 0，发生调用的时候，就会每台机器的 currentWeight += weight，然后选择所有机器中最大的 currentWeight 所在的机器（如果出现相同，则按照顺序选择第一个），随后这台机器的 currentWeight 减去总 weight 权重之和 totalWeight。
| 调用序号 | 选择前当前权重 | Max（currentWeight） | 选择机器 | 选择后当前权重 |
|---|---|---|---|---|
| 1 | 1，2，3 | 3 | C | 1，2，-3 |
| 2 | 2，4，0 | 4 | B | 2，-2，0 |
| 3 | 3，0，3 | 3 | A | -3，0，3 |
| 4 | -2，2，6 | 6 | C | -2，2，0 |
| 5 | -1，4，3 | 4 | B | -1，-2，3 |
| 6 | 0，0，6 | 6 | C | 0，0，0 |
可以看到经过 6 次调用，每台机器被调用的次数与权重的占比是一致的。
这个算法中 currentWeight 总和是不变的，因为被选择的机器的 currentWeight 减去 totalWeight 后每台机器的 currentWeight 又会分别加上自身的 weight，即 weight 越大的机器被选择的次数就会越多。这个算法的数学证明可以参见：https://www.fanhaobai.com/2018/12/load-balance-smooth-weighted-round-robin.html
代码实现：
package dongguabai.demo.roundrobin;
import dongguabai.demo.Invoker;
import java.util.List;
/**
* @author Dongguabai
* @Description 平滑加权轮询
* @Date 创建于 2020-06-27 16:35
*/
public class SmoothWeightRoundRobinLoadBalance extends SimpleRoundRobinLoadBalance {
private boolean averageWeight = true;
private int totalWeight;
private final Object lock = new Object();
public SmoothWeightRoundRobinLoadBalance(List<Invoker> invokers) {
super(invokers);
for (int i = 0; i < invokers.size(); i++) {
Invoker invoker = invokers.get(i);
if (averageWeight && i > 0 && invoker.getWeight() != invokers.get(i - 1).getWeight()) {
averageWeight = false;
}
totalWeight += invoker.getWeight();
}
}
@Override
protected Invoker doSelect() {
if (averageWeight || totalWeight < 1) {
return super.doSelect();
}
synchronized (lock) {
Invoker doInvoker = selectInvoker(invokers);
before(doInvoker);
return doInvoker;
}
}
private Invoker selectInvoker(List<Invoker> invokers) {
Invoker maxCurrentInvoker = invokers.get(0);
for (Invoker invoker : invokers) {
invoker.setCurrentWeight(invoker.getWeight() + invoker.getCurrentWeight());
if (maxCurrentInvoker.getCurrentWeight() < invoker.getCurrentWeight()) {
maxCurrentInvoker = invoker;
}
}
return maxCurrentInvoker;
}
private void before(Invoker doInvoker) {
doInvoker.setCurrentWeight(doInvoker.getCurrentWeight() - totalWeight);
}
}
测试：
@Test
public void testSmoothWeightRoundRobinLoadBalance(){
List<Invoker> invokers = Lists.newArrayList(
new Invoker("A",1),
new Invoker("B",2),
new Invoker("C",3));
AbstractLoadBalance smoothWeightRoundRobinLoadBalance = new SmoothWeightRoundRobinLoadBalance(invokers);
IntStream.range(0,6).forEach(i-> System.out.println(smoothWeightRoundRobinLoadBalance.selectHost()));
}
运行结果：
Invoker(address=C, weight=3, currentWeight=-3)
Invoker(address=B, weight=2, currentWeight=-2)
Invoker(address=A, weight=1, currentWeight=-3)
Invoker(address=C, weight=3, currentWeight=0)
Invoker(address=B, weight=2, currentWeight=-2)
Invoker(address=C, weight=3, currentWeight=0)
哈希
一个最简单的想法是直接用哈希来计算， 对机器数取模。在请求信息足够分散的情况下，是可以满足均衡性的，但一旦有机器加入或者宕机时，所有的原有机器都会受到影响。无法保证稳定性。
这时候就有了一致性哈希算法。一致性哈希算法会将哈希值空间组织成一个虚拟的圆环，即哈希环，普通哈希算法仅仅只针对请求信息进行哈希运算，但是一致性哈希会对服务器也进行哈希运算，从而将两者的哈希值映射在一个圆环上，即哈希环：
当请求哈希后落在 Node1 和 Node2 之间，就会根据顺时针选择到 Node2 机器上。这样保证了稳定性，机器的增加或者退出不会影响所有的机器，但是还是会存在一个问题，假如 Node3 机器宕机，那么 Node1 到 Node4 的请求都会落在 Node1 机器上，这样 Node1 机器的压力就会变大。
于是又有了虚拟节点，即对每一个服务节点计算多个哈希，每个计算结果位置都放置一个此服务节点，通过这种方式将机器节点分散开来，通过虚拟节点均衡各个机器的请求量。
在网上找到了一个哈希的算法实现：
package dongguabai.demo.hash;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
/**
* @author Dongguabai
* @Description
* @Date 创建于 2020-06-27 21:45
*/
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class HashUtils {
/**
* 引用：https://www.jianshu.com/p/4660a8a1f132
* 计算Hash值, 使用FNV1_32_HASH算法
*
* @param str
* @return
*/
public static int getHash(String str) {
final int p = 16777619;
int hash = (int) 2166136261L;
for (int i = 0; i < str.length(); i++) {
hash = (hash ^ str.charAt(i)) * p;
}
hash += hash << 13;
hash ^= hash >> 7;
hash += hash << 3;
hash ^= hash >> 17;
hash += hash << 5;
if (hash < 0) {
hash = Math.abs(hash);
}
return hash;
}
}
一致性哈希算法实现，这里是不关心权重的：
package dongguabai.demo.hash;
import com.google.common.collect.Multiset;
import dongguabai.demo.AbstractLoadBalance;
import dongguabai.demo.Invoker;
import java.util.Comparator;
import java.util.List;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.stream.IntStream;
/**
* @author Dongguabai
* @Description
* @Date 创建于 2020-06-27 21:38
*/
public class ConsistentHashLoadBalance extends AbstractLoadBalance {
/**
* 虚拟节点数量
*/
private static final int VIRTUAL_NODE_NUM = 1000;
/**
* hash 和 虚拟节点映射关系
*/
private SortedMap<Integer, Invoker> virtualNodesMap = new TreeMap<>();
public ConsistentHashLoadBalance(List<Invoker> invokers) {
super(invokers);
putVirtualNodes(invokers);
}
/**
* 添加虚拟节点映射(环)
* @param invokers
*/
private void putVirtualNodes(List<Invoker> invokers) {
for (Invoker invoker : invokers) {
for (int i = 0; i < VIRTUAL_NODE_NUM; i++) {
int hash = HashUtils.getHash(invoker.getAddress() + "-" + i);
virtualNodesMap.put(hash,invoker);
}
}
}
/**
* 根据客户端的信息获取 Invoker
* @param client
* @return
*/
public Invoker getInvoker(Object client){
if (client == null){
return null;
}
//获取大于等于 hash 的第一个 Node
SortedMap<Integer, Invoker> subMap = virtualNodesMap.tailMap(HashUtils.getHash(client.toString()));
if (subMap.isEmpty()) {
// hash值在最尾部，映射到第一个Node上
return virtualNodesMap.get(virtualNodesMap.firstKey());
}
return subMap.get(subMap.firstKey());
}
@Override
public void result(int loop) {
results.clear();
if (loop < 1) {
throw new IllegalArgumentException();
}
IntStream.range(0, loop).forEach(i -> results.add(selectHost("test-client-"+i)));
Set<Multiset.Entry<Invoker>> entrySet = results.entrySet();
entrySet.stream().sorted(Comparator.comparingInt(Multiset.Entry::getCount)).forEach(System.out::println);
}
@Override
protected boolean addInvoker(Invoker invoker) {
if (super.addInvoker(invoker)){
List<Invoker> invokers = getInvokers();
virtualNodesMap.clear();
putVirtualNodes(invokers);
return true;
}
return false;
}
@Override
protected boolean removeInvoker(Invoker invoker) {
if (super.removeInvoker(invoker)){
List<Invoker> invokers = getInvokers();
virtualNodesMap.clear();
putVirtualNodes(invokers);
return true;
}
return false;
}
@Override
protected Invoker doSelect(Object client) {
return getInvoker(client);
}
}
测试：
@Test
public void testConsistentHashLoadBalance(){
List<Invoker> invokers = Lists.newArrayList(
new Invoker("176.170.209.1",1),
new Invoker("176.170.209.2",9),
new Invoker("176.170.209.3",1),
new Invoker("176.170.209.4",9),
new Invoker("176.170.209.5",4),
new Invoker("176.170.209.6",6),
new Invoker("176.170.209.7",1),
new Invoker("176.170.209.8",9),
new Invoker("176.170.209.9",1),
new Invoker("176.1709.10",9));
AbstractLoadBalance consistentHashLoadBalance = new ConsistentHashLoadBalance(invokers);
consistentHashLoadBalance.result(1000000);
Invoker changeInvoker = new Invoker("176.1709.11", 1);
System.out.println("add......");
consistentHashLoadBalance.addInvoker(changeInvoker);
consistentHashLoadBalance.result(1000000);
System.out.println("remove.....");
consistentHashLoadBalance.removeInvoker(changeInvoker);
consistentHashLoadBalance.result(1000000);
}
运行结果：
Invoker(address=176.170.209.8, weight=9, currentWeight=0) x 93728
Invoker(address=176.1709.10, weight=9, currentWeight=0) x 95566
Invoker(address=176.170.209.1, weight=1, currentWeight=0) x 98669
Invoker(address=176.170.209.5, weight=4, currentWeight=0) x 99376
Invoker(address=176.170.209.4, weight=9, currentWeight=0) x 99596
Invoker(address=176.170.209.6, weight=6, currentWeight=0) x 101373
Invoker(address=176.170.209.7, weight=1, currentWeight=0) x 101706
Invoker(address=176.170.209.3, weight=1, currentWeight=0) x 102884
Invoker(address=176.170.209.9, weight=1, currentWeight=0) x 103137
Invoker(address=176.170.209.2, weight=9, currentWeight=0) x 103965
add......
Invoker(address=176.1709.10, weight=9, currentWeight=0) x 85667
Invoker(address=176.170.209.8, weight=9, currentWeight=0) x 86560
Invoker(address=176.170.209.1, weight=1, currentWeight=0) x 90289
Invoker(address=176.170.209.7, weight=1, currentWeight=0) x 90905
Invoker(address=176.170.209.4, weight=9, currentWeight=0) x 90932
Invoker(address=176.1709.11, weight=1, currentWeight=0) x 91236
Invoker(address=176.170.209.5, weight=4, currentWeight=0) x 92081
Invoker(address=176.170.209.9, weight=1, currentWeight=0) x 92808
Invoker(address=176.170.209.2, weight=9, currentWeight=0) x 92847
Invoker(address=176.170.209.6, weight=6, currentWeight=0) x 93085
Invoker(address=176.170.209.3, weight=1, currentWeight=0) x 93590
remove.....
Invoker(address=176.170.209.8, weight=9, currentWeight=0) x 93728
Invoker(address=176.1709.10, weight=9, currentWeight=0) x 95566
Invoker(address=176.170.209.1, weight=1, currentWeight=0) x 98669
Invoker(address=176.170.209.5, weight=4, currentWeight=0) x 99376
Invoker(address=176.170.209.4, weight=9, currentWeight=0) x 99596
Invoker(address=176.170.209.6, weight=6, currentWeight=0) x 101373
Invoker(address=176.170.209.7, weight=1, currentWeight=0) x 101706
Invoker(address=176.170.209.3, weight=1, currentWeight=0) x 102884
Invoker(address=176.170.209.9, weight=1, currentWeight=0) x 103137
Invoker(address=176.170.209.2, weight=9, currentWeight=0) x 103965
可以看到增加或者减少节点也是可以保证负载均衡的。
负载最低优先
前面几种算法主要是站在负载均衡服务的角度，保证每个机器节点获得的调用次数均衡或者相对均衡，但是实际生产环境调用次数相同并不一定真的能够让每台机器真的实现均衡，于是就有了负载最低优先的策略，负载最低优先可以使用如最小连接数、CPU/IO 负载，这里引入《从零开始学架构》中的介绍：
负载最低优先算法基本上能够比较完美地解决轮询算法的缺点，因为采用这种算法后，负载均衡系统需要感知服务器当前的运行状态。当然，其代价是复杂度大幅上升。通俗来讲，轮询可能是 5 行代码就能实现的算法，而负载最低优先算法可能要 1000 行才能实现，甚至需要负载均衡系统和服务器都要开发代码。负载最低优先算法如果本身没有设计好，或者不适合业务的运行特点，算法本身就可能成为性能的瓶颈，或者引发很多莫名其妙的问题。所以负载最低优先算法虽然效果看起来很美好，但实际上真正应用的场景反而没有轮询（包括加权轮询）那么多。
总结
本文主要简单的用 Java 代码实现了几种常见的负载均衡策略及其代码实现，但是有一个值得思考的问题是，要尽可能的将系统关键的节点进行简化，如机器权重，在容器化技术流行的今天，是否可以保证每台机器的性能一致，从而避免因为权重问题增加负载均衡的复杂度；同时负载均衡策略的选择可以具体根据业务场景进行不同的选择，在一些复杂的业务处理场景，如订单业务中根据订单号进行哈希从而将相同的订单交由相同的机器进行处理，避免出现多台机器间数据的不同步，也能够保证业务的处理顺序。
References
-
《从零开始学架构》 -
《分布式 Java 应用》 -
https://www.fanhaobai.com/2018/12/load-balance-smooth-weighted-round-robin.html -
https://zh.wikipedia.org/wiki/%E8%B4%9F%E8%BD%BD%E5%9D%87%E8%A1%A1 -
https://www.jianshu.com/p/4660a8a1f132