Jedis中的一致性hash

Jedis中的一致性hash
本文仅供大家参考，不保证正确性，有问题请及时指出
一致性hash就不多说了，网上有很多说的很好的文章，这里说说Jedis中的Shard是如何使用一致性hash的，也为大家在实现一致性hash提供些思路。

首先是hash函数，在Jedis中有两种Hash算法可供选择，分别是MurMurHash和MD5.
按照Jedis的说法MurMur Hash更快，效果更好些。

MurmurHash.java

package redis.clients.util;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
publicclass
MurmurHash implements Hashing {
    publicstaticint hash(byte[]
data, int seed) {
        return
hash(ByteBuffer.wrap(data), seed);
    }
    publicstaticint hash(byte[]
data, int offset,
int length,
int seed) {
        return
hash(ByteBuffer.wrap(data, offset, length), seed);
    }
    publicstaticint hash(ByteBuffer buf,
int seed) {
        ByteOrder byteOrder = buf.order();
        buf.order(ByteOrder.LITTLE_ENDIAN);
 
        int m = 0x5bd1e995;
        int r = 24;
 
        int h = seed ^ buf.remaining();
 
        int k;
        while (buf.remaining() >= 4) {
            k = buf.getInt();
 
            k *= m;
            k ^= k >>> r;
            k *= m;
 
            h *= m;
            h ^= k;
        }
        if (buf.remaining() > 0) {
            ByteBuffer finish = ByteBuffer.allocate(4).order(
                    ByteOrder.LITTLE_ENDIAN);
            finish.put(buf).rewind();
            h ^= finish.getInt();
            h *= m;
        }
 
        h ^= h >>> 13;
        h *= m;
        h ^= h >>> 15;
 
        buf.order(byteOrder);
        return h;
    }
 
    publicstaticlong hash64A(byte[]
data, int seed) {
        return
hash64A(ByteBuffer.wrap(data), seed);
    }
 
    publicstaticlong hash64A(byte[]
data, int offset,
int length,
int seed) {
        return
hash64A(ByteBuffer.wrap(data, offset, length), seed);
    }
 
    publicstaticlong hash64A(ByteBuffer buf,
int seed) {
        ByteOrder byteOrder = buf.order();
        buf.order(ByteOrder.LITTLE_ENDIAN);
 
        long m = 0xc6a4a7935bd1e995L;
        int r = 47;
 
        long h = seed ^ (buf.remaining() * m);
 
        long k;
        while (buf.remaining() >= 8) {
            k = buf.getLong();
 
            k *= m;
            k ^= k >>> r;
            k *= m;
 
            h ^= k;
            h *= m;
        }
 
        if (buf.remaining() > 0) {
            ByteBuffer finish = ByteBuffer.allocate(8).order(
                    ByteOrder.LITTLE_ENDIAN);
         finish.put(buf).rewind();
            h ^= finish.getLong();
            h *= m;
        }
        h ^= h >>> r;
        h *= m;
        h ^= h >>> r;
        buf.order(byteOrder);
        return h;
    }
 
    publiclong hash(byte[]
key) {
        return
hash64A(key, 0x1234ABCD);
    }
 
    publiclong hash(String key) {
        return hash(SafeEncoder.encode(key));
    }
}

而MD5从java提供的类中即可获得，但是需要将MD5值转换为long类型的常量。

publicinterface Hashing {
    publicstaticfinal Hashing
MURMUR_HASH =
new
MurmurHash();
    public ThreadLocal<MessageDigest>
md5Holder =
new ThreadLocal<MessageDigest>();
    publicstaticfinal Hashing
MD5 =
new Hashing() {
        publiclong hash(String key) {
            return hash(SafeEncoder.encode(key));
        }
        publiclong hash(byte[]
key) {
            try {
                if (md5Holder.get() ==
null) {
                    md5Holder.set(MessageDigest.getInstance("MD5"));
                }
            } catch (NoSuchAlgorithmException e) {
                thrownew IllegalStateException("++++ no md5
algorythm found");
            }
            MessageDigest md5 =
md5Holder.get();
            md5.reset();
            md5.update(key);
            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);
            return res;
        }
};

}

MD5中使用了ThreadLocal，这个就不多说了，网上也有很多相关的blog。或者查看我《如何写性能测试程序》那篇博客。

 

下面主要说说，Jedis中的一致性Hash部分。

 

Shard中定义了如下属性：

    private TreeMap<Long, S>
nodes;
    privatefinal Hashing
algo;
    privatefinal Map<ShardInfo<R>, R>
resources =
new LinkedHashMap<ShardInfo<R>, R>();
nodes属性类型为TreeMap<Long,S>，表示主机映射。每个主机被Hash后，对应多个Long值，而nodes维护了这些Long值对应的主机。

algo是hash算法，是MurMur或者MD5.

resources 完成主机信息和实际链接（Jedis）间的映射。

初始化主机hash如下：

   privatevoid initialize(List<S> shards) {
        nodes =
new TreeMap<Long, S>();
        for (int i = 0; i != shards.size(); ++i)
{
            final S
shardInfo = shards.get(i);
            if (shardInfo.getName() ==
null)
              for (int n = 0; n < 160 *
shardInfo.getWeight(); n++) {
                
nodes.put(this.algo.hash("SHARD-"
+ i + "-NODE-" + n),
shardInfo);
              }
            else
              for (int n = 0; n < 160 *
shardInfo.getWeight(); n++) {
                
nodes.put(this.algo.hash(shardInfo.getName()
+ "*" +
shardInfo.getWeight() + n), shardInfo);
              }
            resources.put(shardInfo,
shardInfo.createResource());
        }
    }

在for循环中，遍历主机列表（shards.get(i)），之后对每个主机按照单权重160的比例计算shard值，将shard值和主机信息（shardInfo）放到nodes中，将主机信息（shardInfo）和其对应的链接资源（Jedis）映射放入到resources中。

Weight是权重，用于调节单个主机被映射值个数，如果weight为1，那么当前主机将被映射为160个值，weight为2，当前主机将被映射为320个值，因此weight为2的节点被访问到的概率就会高一些。

获取某个key对应的主机链接：

   public S getShardInfo(byte[] key) {
        SortedMap<Long, S> tail =
nodes.tailMap(algo.hash(key));
        if (tail.isEmpty()) {
            returnnodes.get(nodes.firstKey());
        }
        return tail.get(tail.firstKey());
    }

在获取key时，先根据key做hash（algo.hash(key)），然后获取node中值大于algo.hash(key)的子树，子树的头结点即为整个子树的最小值，因此选中这个节点做为key的映射节点，取出key值对应的value（hostInfo），之后根据hostInfo，通过resources，获取最终的链接。

   public R getShard(byte[] key) {
        returnresources.get(getShardInfo(key));
    }
 

当然，一致性hash在存储时，一个key可以对应多个节点，一般这多个节点都是在nodes（MapTree）中连续的，因此，也可以改进这个算法，以便实现多节点存储同一个key的方式。

 

参考：jedis-2.1.0.jar