rpc项目中的负载均衡算法

2022-08-30 10:01:15 阅读：202 来源： 互联网

标签：负载 Exception String rpc SERVER 算法 client 地址节点

一致性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

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

rpc项目中的负载均衡算法

一致性hash算法（根据IP一致性hash）

基于最小连接数的负载均衡算法（）