一致性哈希策略的实现方式:我们先把服务列表中的地址进行哈希计算,把计算后的值放到哈希环上,接收到请求后,根据请求的固定属性值来进行哈希计算,然后根据请求的哈希值在哈希环上顺时针寻找服务地址的哈希值,寻找到哪个服务地址的哈希值,就把请求分配给哪个服务。
步骤:
首先获取到节点列表
根据节点列表得到每一个节点的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()); } }