rpc项目中的负载均衡算法
作者：互联网
一致性hash算法（根据IP一致性hash）
一致性哈希策略的实现方式：我们先把服务列表中的地址进行哈希计算，把计算后的值放到哈希环上，接收到请求后，根据请求的固定属性值来进行哈希计算，然后根据请求的哈希值在哈希环上顺时针寻找服务地址的哈希值，寻找到哪个服务地址的哈希值，就把请求分配给哪个服务。
步骤：
-
首先获取到节点列表
-
根据节点列表得到每一个节点的hash值，同时在增加一些虚拟节点，比如原始服务节点上192.168.0.3 那么虚拟节点就是192.168.0.3#1 和192.168.0.3#2 注意所有节点计算hash值都是用string类型的hashcode方法
-
根据消费者的ip计算出消费者的节点请求的hash值，根据Map.tailMap（）方法得到所有有大于 requestHash 的 key，取第一个
虚拟节点主要是解决服务节点hash分布不均匀的问题
/**
* 一致性哈希策略 Demo
*/
public class ConsistentHashingStrategy {
public static void main(String[] args) {
// 模拟 Server 地址列表
String[] serverList = {"192.168.0.15", "192.168.0.30", "192.168.0.45"};
// 新建 TreeMap 集合 ，以 Key，Value 的方式绑定 Hash 值与地址
SortedMap<Integer, String> serverHashMap = new TreeMap<>();
// 计算 Server 地址的 Hash 值
for (String address : serverList) {
int serverHash = Math.abs(address.hashCode());
// 绑定 Hash 值与地址
serverHashMap.put(serverHash, address);
}
// 模拟 Request 地址
String[] requestList = {"192.168.0.10", "192.168.0.20", "192.168.0.40", "192.168.0.50"};
// 计算 Request 地址的 Hash 值
for (String request : requestList) {
int requestHash = Math.abs(request.hashCode());
// 在 serverHashMap 中寻找所有大于 requestHash 的 key
SortedMap<Integer, String> tailMap = serverHashMap.tailMap(requestHash);
//如果有大于 requestHash 的 key， 第一个 key 就是离 requestHash 最近的 serverHash
if (!tailMap.isEmpty()) {
Integer key = tailMap.firstKey();
// 根据 key 获取 Server address
String address = serverHashMap.get(key);
System.out.println("请求 " + request + " 被分配给服务 " + address);
} else {
// 如果 serverHashMap 中没有比 requestHash 大的 key
// 则直接在 serverHashMap 取第一个服务
Integer key = serverHashMap.firstKey();
// 根据 key 获取 Server address
String address = serverHashMap.get(key);
System.out.println("请求 " + request + " 被分配给服务 " + address);
}
}
}
}
基于最小连接数的负载均衡算法（）
服务提供者
注册临时节点到zookeeper 的时候增加一个data字段，初始化为0
/**
* 最小连接数策略 Demo
* Server 服务端注册地址
*/
@Component
public class MinimumConnectionsStrategyServer implements ApplicationRunner {
@Autowired
private CuratorService curatorService;
// Curator 客户端
public CuratorFramework client;
// 当前服务地址的临时节点
public static String SERVER_IP;
// 当前服务地址临时节点的父节点，节点类型为持久节点
public static final String IMOOC_SERVER = "/imooc-server";
/**
* 服务启动后自动执行
*
* @param args args
* @throws Exception Exception
*/
@Override
public void run(ApplicationArguments args) throws Exception {
// Curator 客户端开启会话
client = curatorService.getCuratorClient();
client.start();
// 注册地址信息到 Zookeeper
registerAddressToZookeeper();
}
/**
* 注册地址信息到 Zookeeper
* 服务启动时和服务手动上线时调用此方法
*
* @throws Exception Exception
*/
public void registerAddressToZookeeper() throws Exception {
// 判断父节点是否存在，不存在则创建持久节点
Stat stat = client.checkExists().forPath(IMOOC_SERVER);
if (stat == null) {
client.create().creatingParentsIfNeeded().forPath(IMOOC_SERVER);
}
// 获取本机地址
String address = InetAddress.getLocalHost().getHostAddress();
// 创建临时节点，节点路径为 /IMOOC_SERVER/address，节点 data 为 请求会话数，初始化时为 0.
// /imooc-server/192.168.0.77
SERVER_IP = client.create()
.withMode(CreateMode.EPHEMERAL)
.forPath(IMOOC_SERVER + "/" + address, "0".getBytes());
}
/**
* 注销在 Zookeeper 上的注册的地址
* 服务手动下线时调用此方法
*
* @throws Exception Exception
*/
public void deregistrationAddress() throws Exception {
// 检查该节点是否存在
Stat stat = client.checkExists().forPath(SERVER_IP);
// 存在则删除
if (stat != null) {
client.delete().forPath(SERVER_IP);
}
}
}
消费者
在客户端的请求调用集群服务之前，先使用 Curator 获取 IMOOC_SERVER 下所有的临时节点，并寻找出 data 最小的临时节点，也就是最小连接数的服务。
在客户端发送请求时，我们可以让当前 Server 的请求会话数加 1，并更新到临时节点的 data，完成请求时，我们可以让当前 Server 的请求会话数减 1，并更新到临时节点的 data 。
/**
* 最小连接数策略 Demo
* Client 客户端发送请求
*/
@Component
public class MinimumConnectionsStrategyClient implements ApplicationRunner {
@Autowired
private CuratorService curatorService;
// Curator 客户端
public CuratorFramework client;
// 服务列表节点的 父节点
public static final String IMOOC_SERVER = "/imooc-server";
@Override
public void run(ApplicationArguments args) throws Exception {
// Curator 客户端开启会话
client = curatorService.getCuratorClient();
client.start();
}
/**
* 获取最小连接数的服务
* 发送请求前调用此方法，获取服务地址
*
* @return String
* @throws Exception Exception
*/
public String getTheMinimumNumberOfConnectionsService() throws Exception {
// 获取所有子节点
List<String> list = client.getChildren().forPath(IMOOC_SERVER);
// 新建 Map
Map<String, Integer> map = new HashMap<>();
// 遍历服务列表，保存服务地址与请求会话数的映射关系
for (String s : list) {
byte[] bytes = client.getData().forPath(IMOOC_SERVER + "/" + s);
int i = Integer.parseInt(new String(bytes));
map.put(s, i);
}
// 寻找 map 中会话数最小的值
Optional<Map.Entry<String, Integer>> min = map.entrySet().stream().min(Map.Entry.comparingByValue());
// 不为空的话
if (min.isPresent()) {
// 返回 服务地址 ip
Map.Entry<String, Integer> entry = min.get();
return entry.getKey();
} else {
// 没有则返回服务列表第一个服务地址 ip
return list.get(0);
}
}
/**
* 增加该服务的请求会话数量
* 使用服务地址处理业务前调用此方法
*
* @param ip 服务地址
* @throws Exception Exception
*/
public void increaseTheNumberOfRequestedSessions(String ip) throws Exception {
byte[] bytes = client.getData().forPath(IMOOC_SERVER + "/" + ip);
int i = Integer.parseInt(new String(bytes));
i++;
client.setData().forPath(IMOOC_SERVER + "/" + ip, String.valueOf(i).getBytes());
}
/**
* 减少该服务的请求会话数量
* 请求结束时调用此方法减少会话数量
*
* @param ip 服务地址
* @throws Exception Exception
*/
public void reduceTheNumberOfRequestedSessions(String ip) throws Exception {
byte[] bytes = client.getData().forPath(IMOOC_SERVER + "/" + ip);
int i = Integer.parseInt(new String(bytes));
i--;
client.setData().forPath(IMOOC_SERVER + "/" + ip, String.valueOf(i).getBytes());
}
}
标签：负载,Exception,String,rpc,SERVER,算法,client,地址,节点 来源： https://www.cnblogs.com/jjjmk/p/16638268.html