Dubbo负载均衡:一致性Hash的实现分析

LoadBalance负责从多个Invoker中选出具体的一个用于本次调用,以分摊压力。Dubbo中LoadBalance结构如下图。

1com.alibaba.dubbo.rpc.cluster.LoadBalance 2接口提供了 3<T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException; 4通过该方法,进行结点选择。 5
1com.alibaba.dubbo.rpc.cluster.loadbalance.AbstractLoadBalance 2实现了一些公共方法,并定义抽象方法 3protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation); 4该方法由具体的负载均衡实现类去实现。 5

一致性哈希负载均衡配置

具体的负载均衡实现类包括4种。分别是随机、轮训、最少活跃、一致性Hash。

一致性Hash负载均衡配置,有如下几种形式。

1配置如: 2 3<dubbo:service interface="..." loadbalance="consistenthash" /> 4或: 5 6<dubbo:reference interface="..." loadbalance="consistenthash" /> 7或: 8 9<dubbo:service interface="..."> 10 <dubbo:method name="..." loadbalance="consistenthash"/> 11</dubbo:service> 12或: 13 14<dubbo:reference interface="..."> 15 <dubbo:method name="..." loadbalance="consistenthash"/> 16</dubbo:reference> 17

一致性Hash负载均衡涉及到两个主要的配置参数为hash.argumentshash.nodes

hash.arguments : 当进行调用时候根据调用方法的哪几个参数生成key,并根据key来通过一致性hash算法来选择调用结点。例如调用方法invoke(String s1,String s2); 若hash.arguments为1(默认值),则仅取invoke的参数1(s1)来生成hashCode。

hash.nodes: 为结点的副本数。

1 2缺省只对第一个参数Hash,如果要修改,请配置 3<dubbo:parameter key="hash.arguments" value="0,1" /> 4 5缺省用160份虚拟节点,如果要修改,请配置 6<dubbo:parameter key="hash.nodes" value="320" /> 7 8

Dubbo中一致性Hash的实现分析

dubbo的一致性哈希通过ConsistentHashLoadBalance类来实现。

ConsistentHashLoadBalance内部定义ConsistentHashSelector类,最终通过该类进行结点选择。ConsistentHashLoadBalance实现的doSelect方法来利用所创建的ConsistentHashSelector对象选择结点。

doSelect的实现如下。当调用该方法时,如果选择器不存在则去创建。随后通过ConsistentHashSelector的select方法选择结点。

1 @SuppressWarnings("unchecked") 2 @Override 3 protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) { 4 // 获取调用方法名 5 String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName(); 6 // 生成调用列表hashCode 7 int identityHashCode = System.identityHashCode(invokers); 8 // 以调用方法名为key,获取一致性hash选择器 9 ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key); 10 // 若不存在则创建新的选择器 11 if (selector == null || selector.getIdentityHashCode() != identityHashCode) { 12 // 创建ConsistentHashSelector时会生成所有虚拟结点 13 selectors.put(key, new ConsistentHashSelector<T>(invokers, invocation.getMethodName(), identityHashCode)); 14 // 获取选择器 15 selector = (ConsistentHashSelector<T>) selectors.get(key); 16 } 17 // 选择结点 18 return selector.select(invocation); 19 } 20

ConsistentHashSelector在构造函数内部会创建replicaNumber个虚拟结点,并将这些虚拟结点存储于TreeMap。随后根据调用方法的参数来生成key,并在TreeMap中选择一个结点进行调用。

1 private static final class ConsistentHashSelector<T> { 2 3 private final TreeMap<Long, Invoker<T>> virtualInvokers; // 虚拟结点 4 5 private final int replicaNumber; // 副本数 6 7 private final int identityHashCode;// hashCode 8 9 private final int[] argumentIndex; // 参数索引数组 10 11 public ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) { 12 // 创建TreeMap 来保存结点 13 this.virtualInvokers = new TreeMap<Long, Invoker<T>>(); 14 // 生成调用结点HashCode 15 this.identityHashCode = System.identityHashCode(invokers); 16 // 获取Url 17 // dubbo://169.254.90.37:20880/service.DemoService?anyhost=true&application=srcAnalysisClient&check=false&dubbo=2.8.4&generic=false&interface=service.DemoService&loadbalance=consistenthash&methods=sayHello,retMap&pid=14648&sayHello.timeout=20000&side=consumer&timestamp=1493522325563 18 URL url = invokers.get(0).getUrl(); 19 // 获取所配置的结点数,如没有设置则使用默认值160 20 this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160); 21 // 获取需要进行hash的参数数组索引,默认对第一个参数进行hash 22 String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0")); 23 argumentIndex = new int[index.length]; 24 for (int i = 0; i < index.length; i ++) { 25 argumentIndex[i] = Integer.parseInt(index[i]); 26 } 27 // 创建虚拟结点 28 // 对每个invoker生成replicaNumber个虚拟结点,并存放于TreeMap中 29 for (Invoker<T> invoker : invokers) { 30 31 for (int i = 0; i < replicaNumber / 4; i++) { 32 // 根据md5算法为每4个结点生成一个消息摘要,摘要长为16字节128位。 33 byte[] digest = md5(invoker.getUrl().toFullString() + i); 34 // 随后将128位分为4部分,0-31,32-63,64-95,95-128,并生成4个32位数,存于long中,long的高32位都为0 35 // 并作为虚拟结点的key。 36 for (int h = 0; h < 4; h++) { 37 long m = hash(digest, h); 38 virtualInvokers.put(m, invoker); 39 } 40 } 41 } 42 } 43 44 public int getIdentityHashCode() { 45 return identityHashCode; 46 } 47 48 // 选择结点 49 public Invoker<T> select(Invocation invocation) { 50 // 根据调用参数来生成Key 51 String key = toKey(invocation.getArguments()); 52 // 根据这个参数生成消息摘要 53 byte[] digest = md5(key); 54 //调用hash(digest, 0),将消息摘要转换为hashCode,这里仅取0-31位来生成HashCode 55 //调用sekectForKey方法选择结点。 56 Invoker<T> invoker = sekectForKey(hash(digest, 0)); 57 return invoker; 58 } 59 60 private String toKey(Object[] args) { 61 StringBuilder buf = new StringBuilder(); 62 // 由于hash.arguments没有进行配置,因为只取方法的第1个参数作为key 63 for (int i : argumentIndex) { 64 if (i >= 0 && i < args.length) { 65 buf.append(args[i]); 66 } 67 } 68 return buf.toString(); 69 } 70 71 //根据hashCode选择结点 72 private Invoker<T> sekectForKey(long hash) { 73 Invoker<T> invoker; 74 Long key = hash; 75 // 若HashCode直接与某个虚拟结点的key一样,则直接返回该结点 76 if (!virtualInvokers.containsKey(key)) { 77 // 若不一致,找到一个最小上届的key所对应的结点。 78 SortedMap<Long, Invoker<T>> tailMap = virtualInvokers.tailMap(key); 79 // 若存在则返回,例如hashCode落在图中[1]的位置 80 // 若不存在,例如hashCode落在[2]的位置,那么选择treeMap中第一个结点 81 // 使用TreeMap的firstKey方法,来选择最小上界。 82 if (tailMap.isEmpty()) { 83 key = virtualInvokers.firstKey(); 84 } else { 85 86 key = tailMap.firstKey(); 87 } 88 } 89 invoker = virtualInvokers.get(key); 90 return invoker; 91 } 92 93 private long hash(byte[] digest, int number) { 94 return (((long) (digest[3 + number * 4] & 0xFF) << 24) 95 | ((long) (digest[2 + number * 4] & 0xFF) << 16) 96 | ((long) (digest[1 + number * 4] & 0xFF) << 8) 97 | (digest[0 + number * 4] & 0xFF)) 98 & 0xFFFFFFFFL; 99 } 100 101 private byte[] md5(String value) { 102 MessageDigest md5; 103 try { 104 md5 = MessageDigest.getInstance("MD5"); 105 } catch (NoSuchAlgorithmException e) { 106 throw new IllegalStateException(e.getMessage(), e); 107 } 108 md5.reset(); 109 byte[] bytes = null; 110 try { 111 bytes = value.getBytes("UTF-8"); 112 } catch (UnsupportedEncodingException e) { 113 throw new IllegalStateException(e.getMessage(), e); 114 } 115 md5.update(bytes); 116 return md5.digest(); 117 } 118 119 } 120

上述代码中 hash(byte[] digest, int number)方法用来生成hashCode。该函数将生成的结果转换为long类,这是因为生成的结果是一个32位数,若用int保存可能会产生负数。而一致性hash生成的逻辑环其hashCode的范围是在 0 - MAX_VALUE之间。因此为正整数,所以这里要强制转换为long类型,避免出现负数。

进行结点选择的方法为select,最后通过sekectForKey方法来选择结点。

1 // 选择结点 2 public Invoker<T> select(Invocation invocation) { 3 // 根据调用参数来生成Key 4 String key = toKey(invocation.getArguments()); 5 // 根据这个参数生成消息摘要 6 byte[] digest = md5(key); 7 //调用hash(digest, 0),将消息摘要转换为hashCode,这里仅取0-31位来生成HashCode 8 //调用sekectForKey方法选择结点。 9 Invoker<T> invoker = sekectForKey(hash(digest, 0)); 10 return invoker; 11 } 12

sekectForKey方法的实现如下。

1 private Invoker<T> sekectForKey(long hash) { 2 Invoker<T> invoker; 3 Long key = hash; 4 // 若HashCode直接与某个虚拟结点的key一样,则直接返回该结点 5 if (!virtualInvokers.containsKey(key)) { 6 // 若不在,找到一个最小上届的key所对应的结点。 7 SortedMap<Long, Invoker<T>> tailMap = virtualInvokers.tailMap(key); 8 // 若存在则返回,例如hashCode落在图中[1]的位置 9 // 若不存在,例如hashCode落在[2]的位置,那么选择treeMap中第一个结点 10 // 使用TreeMap的firstKey方法,来选择最小上界。 11 if (tailMap.isEmpty()) { 12 key = virtualInvokers.firstKey(); 13 } else { 14 15 key = tailMap.firstKey(); 16 } 17 } 18 invoker = virtualInvokers.get(key); 19 return invoker; 20 } 21

在进行选择时候若HashCode直接与某个虚拟结点的key一样,则直接返回该结点,例如hashCode落在某个结点上(圆圈所表示)。若不在,找到一个最小上届的key所对应的结点。例如进行选择时的key落在图中1所标注的位置。由于利用TreeMap存储,key所落在的位置可能无法找到最小上界,例如图中2所标注的位置。那么需要返回TreeMap中的最小值(构成逻辑环状结构,找不到,则返回最开头的结点)。

参考

dubbo 2.8.4源码

上一篇:Dubbo负载均衡

代码交流 2021