项目3：从0开始的RPC框架（扩展版）-3

七. 负载均衡

1. 需求分析

目前我们的RPC框架仅允许消费者读取第一个服务提供者的服务节点，但在实际应用中，同一个服务会有多个服务提供者上传节点信息。如果消费者只读取第一个，势必会增大单个节点的压力，并且也浪费了其它节点资源。

//获取第一个服务节点
ServiceMetaInfo selectedServiceMetaInfo = serviceMetaInfoList.get(0);

因此，RPC框架应该允许对服务节点进行选择，而非调用固定的一个。采用负载均衡能够有效解决这个问题。

2. 设计方案

（1）什么是负载均衡

负载，可以理解为要处理的工作和压力，比如网络请求、事务、数据处理任务等。

均衡，就是将工作和压力平均地分配给多个工作者，从而分摊每个工作者的压力。

即将工作负载均匀分配到多个计算资源（如服务器、网络链路、硬盘驱动器等）的技术，以优化资源使用、最大化吞吐量、最小化响应时间和避免过载。

在RPC框架中，负载均衡的作用是从一组可用的服务节点中选择一个进行调用。

（2）常见的负载均衡算法

所谓负载均衡算法，就是按照何种策略选择资源。不同的算法有不同的适用场景，要根据实际情况选择。

• 轮询（Round Robin）：按照循环的顺序将请求分配给每个服务器，适用于各服务器性能相近的情况。

//5台服务器节点，请求调用顺序：
1，2，3，4，5，1，2，3，4，5

• 随机（Random）：随机选择一个服务器处理请求，适用于各服务器性能相近且当前负载均匀的情况。

//5台服务器节点，请求调用顺序：
2，4，1，3，2，5，1，4，4，1

• 加权轮询（Weighted Round Robin）：根据服务器的性能或权重来分配请求，权重更高的服务器会获得更多的请求，适用于服务器性能不均匀的情况。

//假设有1台千兆带宽的服务器节点和4台百兆带宽的服务器节点，请求调用顺序：
1，1，1，2， 1，1，1，3， 1，1，1，4， 1，1，1，5

• 加权随机（Weighted Random）：根据服务器的权重随机选择一个处理请求，适用于服务器性能不均匀的情况。

//假设有2台千兆带宽，3台百兆带宽的服务器节点，请求调用顺序：
1，2，2，1，3， 1，1，1，2，4， 2，2，2，1，5

• 最小连接数（Least Connections）：选择当前连接数最少的服务器处理请求，适用于长连接场景。
• IP Hash：根据客户端IP地址的哈希值选择服务器处理请求，确保同一客户端的请求始终被分配到同一台服务器上，适用于需要保持会话一致性的场景。

另一种比较经典的哈希算法，

一致性哈希（Consistent Hashing）：将整个哈希值空间划分成一个环状结构，每个节点或服务器在环上占据一个位置，每个请求根据其哈希值映射到环上的一个点，然后顺时针寻找第一个遇见的节点，将请求路由到该节点上。

一致性哈希结构图如下图所示，并且请求A会在服务器C进行处理。

此外，一致性哈希还能有效解决节点下线和倾斜问题。

• 节点下线：某个节点下线后，其负载会被平均分摊到其它节点上，不会影响到整个系统的稳定性，只会产生局部变动。

当服务器C下线后，请求A会交由服务器A处理。服务器B接收到的请求不变。

• 倾斜问题：如果服务节点在哈希环上分布不均匀，可能会导致大部分请求全都集中在某一台服务器上，造成数据分布不均匀。通过引入虚拟节点，对每个服务节点计算多个哈希，每个计算结果位置都放置该服务节点，即一个实际物理节点对应多个虚拟节点，使得请求能够被均匀分布，减少节点间的负载差异。

如上图所示节点分布情况，大部分请求都会落在服务器C中，服务器A中的请求会很少。

引入虚拟节点后，每个服务器会有多个节点，使得请求分布更加均匀。

3. 具体实现

（1）实现三种负载均衡器

根据轮询、随机和一致性哈希三种负载均衡算法实现对应的负载均衡器。

创建loadbalancer包，将所有负载均衡器相关的代码都放在该包下。

创建LoadBalancer负载均衡器通用接口。

提供一个选择服务方法，接收请求参数和可用服务列表，可以根据这些信息进行选择：

package com.khr.krpc.loadbalancer;import com.khr.krpc.model.ServiceMetaInfo;import java.util.List;
import java.util.Map;/*** 负载均衡器（消费端使用）*/
public interface LoadBalancer {/*** 选择服务调用** @param requestParams 请求参数* @param serviceMetaInfoList 可用服务列表* @return*/ServiceMetaInfo select(Map<String,Object> requestParams,List<ServiceMetaInfo>serviceMetaInfoList);
}

实现轮询负载均衡器。

使用到了JUC包的AtomicInteger实现原子计数器，防止并发冲突。

package com.khr.krpc.loadbalancer;import com.khr.krpc.model.ServiceMetaInfo;import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;/*** 轮询负载均衡器*/public class RoundRobinLoadBalancer implements LoadBalancer{/*** 当前轮询的下标*/private final AtomicInteger currentIndex = new AtomicInteger(0);@Overridepublic ServiceMetaInfo select(Map<String,Object> requestParams,List<ServiceMetaInfo> serviceMetaInfoList){if (serviceMetaInfoList.isEmpty()){return null;}//只有一个服务，无需轮询int size = serviceMetaInfoList.size();if (size == 1){return serviceMetaInfoList.get(0);}//取模算法轮询int index = currentIndex.getAndIncrement() % size;return  serviceMetaInfoList.get(index);}
}

currentIndex是一个AtomixInteger，它是一个线程安全的整数变量。getAndIncrement() 方法会返回当前值，然后将其递增。例如 currentIndex 当前值是0，那么 getAndIncrement() 会返回0，并将currentIndex设置为1，依次类推。

size是 serviceMetaInfoList 的大小，取模运算大致流程如下：

//假设size=5
第一次调用：currentIndex.getAndIncrement() % 5 -> 0 % 5 = 0 返回服务1。
第二次调用：currentIndex.getAndIncrement() % 5 -> 1 % 5 = 1 返回服务2。
第三次调用：currentIndex.getAndIncrement() % 5 -> 2 % 5 = 2 返回服务3。
第四次调用：currentIndex.getAndIncrement() % 5 -> 3 % 5 = 3 返回服务4。
第五次调用：currentIndex.getAndIncrement() % 5 -> 4 % 5 = 4 返回服务5。
第六次调用：currentIndex.getAndIncrement() % 5 -> 5 % 5 = 0 返回服务1。
（循环开始）

实现随机负载均衡器。

使用Java自带的Random类实现随机选取：

package com.khr.krpc.loadbalancer;import com.khr.krpc.model.ServiceMetaInfo;import java.util.List;
import java.util.Map;
import java.util.Random;/*** 随机负载均衡器*/
public class RandomLoadBalancer implements LoadBalancer{private final Random random = new Random();@Overridepublic ServiceMetaInfo select(Map<String,Object> requestParams,List<ServiceMetaInfo> serviceMetaInfoList){int size = serviceMetaInfoList.size();if (size == 0){return null;}//只有一个服务，不用随机if (size == 1){return serviceMetaInfoList.get(0);}return serviceMetaInfoList.get(random.nextInt(size));}
}

实现一致性哈希负载均衡器。

使用TreeMap实现一致性哈希环，该数据结构提供了 ceilingEntry 和 firstEntry 两个方法，便于获取符合算法要求的节点：

package com.khr.krpc.loadbalancer;import com.khr.krpc.model.ServiceMetaInfo;import java.util.List;
import java.util.Map;
import java.util.TreeMap;/*** 一致性哈希负载均衡器*/public class ConsistentHashLoadBalancer implements LoadBalancer{/*** 一致性 Hash 环，存放虚拟节点*/private final TreeMap<Integer, ServiceMetaInfo> virtualNodes = new TreeMap<>();/*** 虚拟节点数*/private static final int VIRTUAL_NODE_NUM = 100;@Overridepublic ServiceMetaInfo select(Map<String, Object> requestParams,List<ServiceMetaInfo> serviceMetaInfoList){if (serviceMetaInfoList.isEmpty()){return null;}//构建虚拟节点for (ServiceMetaInfo serviceMetaInfo : serviceMetaInfoList){for (int i = 0; i < VIRTUAL_NODE_NUM; i++){int hash = getHash(serviceMetaInfo.getServiceAddress() + "#" + i);virtualNodes.put(hash,serviceMetaInfo);}}//获取调用请求的 hash 值int hash = getHash(requestParams);//选择最接近且大于等于调用请求 hash 值的虚拟节点Map.Entry<Integer,ServiceMetaInfo> entry = virtualNodes.ceilingEntry(hash);if (entry == null){//如果没有大于等于调用请求 hash 值的虚拟节点，则返回环首部的节点entry = virtualNodes.firstEntry();}return  entry.getValue();}/*** Hash 算法** @param key* @return*/private int getHash(Object key){return key.hashCode();}
}

该均衡器给每个服务实例创建100个虚拟节点，并且每个节点的哈希值由其服务地址 getServiceAddress() 和虚拟节点索引 i 组成一个字符串构成。

//某服务1地址为"192.168.0.1"：
虚拟节点 0 的哈希值：getHash("192.168.0.1#0")
虚拟节点 1 的哈希值：getHash("192.168.0.1#1")
虚拟节点 2 的哈希值：getHash("192.168.0.1#2")
……//某服务2地址为"192.168.0.2"：
虚拟节点 0 的哈希值：getHash("192.168.0.2#0")
虚拟节点 1 的哈希值：getHash("192.168.0.2#1")
虚拟节点 2 的哈希值：getHash("192.168.0.2#2")
……依次类推

此外，该一致性哈希均衡器中的 Hash 算法只是简单调用了对象的 hashCode 方法，可以自定义更复杂的哈希算法。

（2）支持配置和扩展负载均衡器

像注册中心和序列化器一样，负载均衡器同样支持自定义，让开发者能够填写配置来指定使用的负载均衡器。依旧使用工厂创建对象、使用SPI动态加载。

在loadbalancer包下创建LoadBalancerKeys类，列举所有支持的负载均衡器键名：

package com.khr.krpc.loadbalancer;/*** 负载均衡器键名常量*/
public interface LoadBalancerKeys {/*** 轮询*/String ROUND_ROBIN = "roundRobin";/*** 随机*/String RANDOM = "random";/*** 一致性哈希*/String CONSISTENT_HASH = "consistentHash";
}

在loadbalancer包下创建LoadBalancerFactory类。

使用工厂模式，支持根据 key 从SPI获取负载均衡器对象实例（和SerializerFactory几乎一致）：

package com.khr.krpc.loadbalancer;import com.khr.krpc.spi.SpiLoader;/*** 负载均衡器工厂（工厂模式，用于获取负载均衡器对象）*/
public class LoadBalancerFactory {static {SpiLoader.load(LoadBalancer.class);}/*** 默认负载均衡器*/private static final LoadBalancer DEFAULT_LOAD_BALANCER = new RoundRobinLoadBalancer();/*** 获取实例** @param key* @return*/public static LoadBalancer getInstance(String key){return SpiLoader.getInstance(LoadBalancer.class, key);}
}

在META-INF的rpc/system目录下编写负载均衡器接口的SPI配置文件，文件名称为com.khr.krpc.loadbalancer.LoadBalancer：

roundRobin=com.khr.krpc.loadbalancer.RoundRobinLoadBalancer
random=com.khr.krpc.loadbalancer.RandomLoadBalancer
consistentHash=com.khr.krpc.loadbalancer.ConsistentHashLoadBalancer

在RpcConfig类中新增负载均衡器全局配置：

package com.khr.krpc.config;import com.khr.krpc.loadbalancer.LoadBalancerKeys;
import com.khr.krpc.serializer.SerializerKeys;
import lombok.Data;/*** RPC框架配置*/
@Data
public class RpcConfig {……/*** 负载均衡器*/private String loadBalancer = LoadBalancerKeys.ROUND_ROBIN;
}

（3）应用负载均衡器

修改ServiceProxy类中针对服务节点调用的代码，改为调用负载均衡器获取节点：

//负载均衡器LoadBalancer loadBalancer = LoadBalancerFactory.getInstance(rpcConfig.getLoadBalancer());
//将调用方法名（请求路径）作为负载均衡参数
Map<String, Object> requestParams = new HashMap<>();
requestParams.put("methodName", rpcRequest.getMethodName());
ServiceMetaInfo selectedServiceMetaInfo = loadBalancer.select(requestParams,serviceMetaInfoList);

4. 测试

（1）测试负载均衡算法

编写单元测试类LoadBalancerTest：

package com.khr.rpc.loadbalancer;import com.khr.krpc.loadbalancer.ConsistentHashLoadBalancer;
import com.khr.krpc.loadbalancer.LoadBalancer;
import com.khr.krpc.loadbalancer.RandomLoadBalancer;
import com.khr.krpc.loadbalancer.RoundRobinLoadBalancer;
import com.khr.krpc.model.ServiceMetaInfo;
import org.junit.Assert;
import org.junit.Test;import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;import static org.junit.Assert.*;/*** 负载均衡器测试*/
public class LoadBalancerTest {final LoadBalancer loadBalancer = new RandomLoadBalancer();@Testpublic void select(){//请求参数Map<String, Object> requestParams = new HashMap<>();requestParams.put("methodName","apple");//服务列表ServiceMetaInfo serviceMetaInfo1 = new ServiceMetaInfo();serviceMetaInfo1.setServiceName("MyService");serviceMetaInfo1.setServiceVersion("1.0");serviceMetaInfo1.setServiceHost("localhost");serviceMetaInfo1.setServicePort(1234);ServiceMetaInfo serviceMetaInfo2 = new ServiceMetaInfo();serviceMetaInfo2.setServiceName("MyService");serviceMetaInfo2.setServiceVersion("1.0");serviceMetaInfo2.setServiceHost("khr.icu");serviceMetaInfo2.setServicePort(80);List<ServiceMetaInfo> serviceMetaInfoList = Arrays.asList(serviceMetaInfo1,serviceMetaInfo2);//连续调用3次ServiceMetaInfo serviceMetaInfo = loadBalancer.select(requestParams,serviceMetaInfoList);System.out.println(serviceMetaInfo);Assert.assertNotNull(serviceMetaInfo);serviceMetaInfo = loadBalancer.select(requestParams,serviceMetaInfoList);System.out.println(serviceMetaInfo);Assert.assertNotNull(serviceMetaInfo);serviceMetaInfo = loadBalancer.select(requestParams,serviceMetaInfoList);System.out.println(serviceMetaInfo);Assert.assertNotNull(serviceMetaInfo);}
}

替换loadBalancer对象为不同的负载均衡器，有不同的运行结果：

Random

RoundRobin

 Consistent Hashing

（2）测试负载均衡调用

在不同的端口启动两个服务提供者，然后启动消费者，通过Debug或控制台输出来观察每次请求的节点地址即可。这里不再做详细演示。

至此，扩展功能，负载均衡完成。