源码分析Memcached-Java-Client一致性hash算法

源码分析Memcached-Java-Client一致性hash算法

url: http://thrillerzw.iteye.com/blog/2055487
2014-04-25 23:27

memcached的分布式主要体现在client端,server之间没有关系。

一致性hash算法：cache不能命中的问题仍然存在，但是只存在于2个节点之间的位置。相对于取模的算法，一致性hash算法除了计算key的hash值外，还会计算每个server对应的hash值，然后将这些hash值映射到一个有限的值域上（比如0~2^32）。通过寻找hash值等于大于hash(key)的最小server作为存储该key数据的目标server。如果找不到，则直接把具有最小hash值的server作为目标server。
源码版本：Memcached-Java-Client-release_2.6.1

1、计算每个server对应的hash值，将值存入treemap 排序模拟圆，主要方法代码入下。

private void populateConsistentBuckets() {   
  // store buckets in tree map   
  consistentBuckets = new TreeMap<Long, String>();   
  MessageDigest md5 = MD5.get();   
  if (this.totalWeight <= 0 && this.weights != null) {   
   for (int i = 0; i < this.weights.length; i++)   
    this.totalWeight += (this.weights[i] == null) ? 1 : this.weights[i];   
  } else if (this.weights == null) {   
   this.totalWeight = this.servers.length;   
  }   
  for (int i = 0; i < servers.length; i++) {   
   int thisWeight = 1;   
   if (this.weights != null && this.weights[i] != null)   
    thisWeight = this.weights[i];   
   double factor = Math.floor(((double) (40 * this.servers.length * thisWeight)) / (double) this.totalWeight);   
      //根据权重控制虚拟节点数量 （factor * 4）。   
   for (long j = 0; j < factor; j++) {   
    byte[] d = md5.digest((servers[i] + "-" + j).getBytes());   
    //192.168.211.240:11212-0 。。。 192.168.211.240:11212-39   
//   System.out.println("server key="+servers[i] + "-" + j);   
    for (int h = 0; h < 4; h++) {   
     Long k = ((long) (d[3 + h * 4] & 0xFF) << 24) | ((long) (d[2 + h * 4] & 0xFF) << 16)   
       | ((long) (d[1 + h * 4] & 0xFF) << 8) | ((long) (d[0 + h * 4] & 0xFF));   
     //按Long值从小到大排序   
     consistentBuckets.put(k, servers[i]);   
    }   
   }   
   // Create a socket pool for each host   
   // Create an object pool to contain our active connections   
   GenericObjectPool gop;   
   SchoonerSockIOFactory factory;   
   if (authInfo != null) {   
    factory = new AuthSchoonerSockIOFactory(servers[i], isTcp, bufferSize, socketTO, socketConnectTO,   
      nagle, authInfo);   
   } else {   
    factory = new SchoonerSockIOFactory(servers[i], isTcp, bufferSize, socketTO, socketConnectTO, nagle);   
   }   
   gop = new GenericObjectPool(factory, maxConn, GenericObjectPool.WHEN_EXHAUSTED_BLOCK, maxIdle, maxConn);   
   factory.setSockets(gop);   
   socketPool.put(servers[i], gop);   
  }   
  System.out.println("consistentBuckets="+consistentBuckets);   
 }

输出：

consistentBuckets={7786957=192.168.211.240:11212, 13055238=192.168.211.240:11212, 15819052=192.168.211.240:11211,。。。 4294784513=192.168.211.240:11212}

2、计算存储数据key hash,寻找目标server。

consistentBuckets.get(bucket)   
 //计算客户端key hash值   
 private static long md5HashingAlg(String key) {   
  MessageDigest md5 = MD5.get();   
  md5.reset();   
  md5.update(key.getBytes());   
  byte[] bKey = md5.digest();   
  long res = ((long) (bKey[3] & 0xFF) << 24) | ((long) (bKey[2] & 0xFF) << 16) | ((long) (bKey[1] & 0xFF) << 8)   
    | (long) (bKey[0] & 0xFF);   
  System.out.println("key="+key+" hash result="+res);   
  return res;   
 }   
输出：   
 key=1 hash result=943901380  
 key=2 hash result=2373066440  
    
 //查找目标server   
 private final Long findPointFor(Long hv) {   
  // this works in java 6, but still want to release support for java5   
  // Long k = this.consistentBuckets.ceilingKey( hv );   
  // return ( k == null ) ? this.consistentBuckets.firstKey() : k;   
    //hash值等于大于hash(key)的最小server作为存储该key数据的目标server。   
  SortedMap<Long, String> tmap = this.consistentBuckets.tailMap(hv);   
  System.out.println("hash result="+hv+" tailMap="+tmap);   
  //如果找不到，则直接把具有最小hash值的server作为目标server。   
  return (tmap.isEmpty()) ? this.consistentBuckets.firstKey() : tmap.firstKey();   
 }   
    
输出：   
hash result=943901380 tailMap={948021698=192.168.211.240:11212, 973571166=192.168.211.240:11212, 。。。4294784513=192.168.211.240:11212}