Jedis一致性hash与sharding

  一、Jedis一致性hash

    利用缓存技术，不仅可以提升系统性能，还能缓解系统故障。对于redis 3.0以下的版本，redis-server没有sharding的功能，只有master-slave模式。目前企业用的普遍都是只有m/s模式的redis多实例部署，无论是master还是slave挂掉，都需要调整程序配置(或代码)。Jedis为我们提供了编程级别的sharding方式，本文主要介绍相关API使用方法。

     Jedis中sharding基于一致性hash算法，hash值计算采取MD5作为辅助，此算法似乎已成事实上的标准，不过较新的版本采用的是谷歌的murmur_hash算法(MD5 is really not good?)。

Java代码  

1. public interface Hashing {  
2.   public static final Hashing MURMUR_HASH = new MurmurHash();  
3.   public ThreadLocal<MessageDigest> md5Holder = new ThreadLocal<MessageDigest>();  
4.          // 基于MD5的一致性hash算法实现  
5.   public static final Hashing MD5 = new Hashing() {  
6.     public long hash(String key) {  
7.       return hash(SafeEncoder.encode(key));  
8.     }  
9.   
10.     public long hash(byte[] key) {  
11.       try {  
12.         if (md5Holder.get() == null) {  
13.           md5Holder.set(MessageDigest.getInstance("MD5"));  
14.         }  
15.       } catch (NoSuchAlgorithmException e) {  
16.         throw new IllegalStateException("++++ no md5 algorythm found");  
17.       }  
18.       MessageDigest md5 = md5Holder.get();  
19.   
20.       md5.reset();  
21.       md5.update(key);  
22.       byte[] bKey = md5.digest(); // 获得MD5字节序列  
23.       // 前四个字节作为计算参数，最终获得一个32位int值.    
24.       // 此种计算方式，能够确保key的hash值更加“随即”/“离散”    
25.       // 如果hash值过于密集，不利于一致性hash的实现(特别是有“虚拟节点”设计时)   
26.       long res = ((long) (bKey[3] & 0xFF) << 24) | ((long) (bKey[2] & 0xFF) << 16)  
27.           | ((long) (bKey[1] & 0xFF) << 8) | (long) (bKey[0] & 0xFF);  
28.       return res;  
29.     }  
30.   };  
31.   
32.   public long hash(String key);  
33.   
34.   public long hash(byte[] key);  
35. }  

    node构建过程：

Java代码  

1. //shards列表为客户端提供了所有redis-server配置信息，包括：ip，port，weight，name  
2. //其中weight为权重，将直接决定“虚拟节点”的“比例”(密度)，权重越高，在存储是被hash命中的概率越高  
3. //--其上存储的数据越多。  
4. //其中name为“节点名称”，jedis使用name作为“节点hash值”的一个计算参数。  
5. //---  
6. //一致性hash算法，要求每个“虚拟节点”必须具备“hash值”，每个实际的server可以有多个“虚拟节点”(API级别)  
7. //其中虚拟节点的个数= “逻辑区间长度” * weight，每个server的“虚拟节点”将会以“hash”的方式分布在全局区域中  
8. //全局区域总长为2^32.每个“虚拟节点”以hash值的方式映射在全局区域中。  
9. // 环形：0-->vnode1(:1230)-->vnode2(:2800)-->vnode3(400000)---2^32-->0  
10. //所有的“虚拟节点”将按照其”节点hash“顺序排列(正序/反序均可)，因此相邻两个“虚拟节点”之间必有hash值差，  
11. //那么此差值，即为前一个(或者后一个，根据实现而定)“虚拟节点”所负载的数据hash值区间。  
12. //比如hash值为“2000”的数据将会被vnode1所接受。  
13. //---  
14. private void initialize(List<S> shards) {  
15.     nodes = new TreeMap<Long, S>();//虚拟节点，采取TreeMap存储:排序，二叉树  
16.   
17.     for (int i = 0; i != shards.size(); ++i) {  
18.         final S shardInfo = shards.get(i);  
19.         if (shardInfo.getName() == null)  
20.                 //当没有设置“name”是，将“SHARD-NODE”作为“虚拟节点”hash值计算的参数  
21.                 //"逻辑区间步长"为160,为什么呢??  
22.                 //最终多个server的“虚拟节点”将会交错布局，不一定非常均匀。  
23.             for (int n = 0; n < 160 * shardInfo.getWeight(); n++) {  
24.                 nodes.put(this.algo.hash("SHARD-" + i + "-NODE-" + n), shardInfo);  
25.             }  
26.         else  
27.             for (int n = 0; n < 160 * shardInfo.getWeight(); n++) {  
28.                 nodes.put(this.algo.hash(shardInfo.getName() + "*" + shardInfo.getWeight() + n), shardInfo);  
29.             }  
30.         resources.put(shardInfo, shardInfo.createResource());  
31.     }  
32. }  

    node选择方式：

Java代码  

1. public R getShard(String key) {  
2.     return resources.get(getShardInfo(key));  
3. }  
4. //here:  
5. public S getShardInfo(byte[] key) {  
6.         //获取>=key的“虚拟节点”的列表  
7.     SortedMap<Long, S> tail = nodes.tailMap(algo.hash(key));  
8.         //如果不存在“虚拟节点”，则将返回首节点。  
9.     if (tail.size() == 0) {  
10.         return nodes.get(nodes.firstKey());  
11.     }  
12.         //如果存在，则返回符合(>=key)条件的“虚拟节点”的第一个节点  
13.     return tail.get(tail.firstKey());  
14. }  

    Jedis sharding默认的一致性hash算法比较适合cache-only的情景，不太适合数据持久化情况。在持久存储情况下，我们可以使用“强hash”分片，需要重写hash算法(参加后面的InnerHashing)。强hash算法下，如果某个虚拟节点所在的物理节点故障，将导致数据无法访问，即无法从虚拟节点列表中删除失效的server。

    二、API

    ShardedJedis

Java代码  

1.         JedisShardInfo sd1 = new JedisShardInfo("127.0.0.1", 6379, 15000);  
2.         sd1.setPassword("123456");  
3.         JedisShardInfo sd2 = new JedisShardInfo("127.0.0.1", 6479, 15000);  
4.         sd2.setPassword("123456");  
5.         List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();  
6.         shards.add(sd1);  
7.         shards.add(sd2);  
8.         ShardedJedis shardedJedis = new ShardedJedis(shards, new InnerHashing());  
9.         String key = "k2sdjowejjroer3";  
10.         shardedJedis.set(key, "v2");  
11.         Charset charset = Charset.forName("utf-8");  
12.         // 注意此处对key的字节转换时，一定要和Innerhashing.hash(String)保持一致  
13.         System.out.println(shardedJedis.get("k2").getBytes(charset));  
14.   
15. // Jedis的一致性hash算法已经足够良好，程序员建议不要重写  
16. public class InnerHashing implements Hashing {  
17.     static Charset charset = Charset.forName("utf-8");  
18.   
19.     @Override  
20.     public long hash(String key) {  
21.         return hash(key.getBytes(charset));  
22.     }  
23.   
24.     @Override  
25.     public long hash(byte[] key) {  
26.         int hashcode = new HashCodeBuilder().append(key).toHashCode();  
27.         return hashcode & 0x7FFFFFFF;  
28.     }  
29. }  

Xml代码  

1. <bean id="shardedJedis" class="redis.clients.jedis.ShardedJedis">  
2.     <constructor-arg>  
3.         <list>  
4.             <bean class="redis.clients.jedis.JedisShardInfo">  
5.                 <constructor-arg value="127.0.0.1"></constructor-arg>  
6.                 <constructor-arg value="6379"></constructor-arg>  
7.                 <property name="password" value="0123456"></property>  
8.             </bean>  
9.             <bean class="redis.clients.jedis.JedisShardInfo">  
10.                 <constructor-arg value="127.0.0.1"></constructor-arg>  
11.                 <constructor-arg value="6379"></constructor-arg>  
12.                 <property name="password" value="0123456"></property>  
13.             </bean>  
14.         </list>  
15.     </constructor-arg>  
16. </bean>  

     ShardedJedisPool & ShardedJedisPipeline

Java代码  

1. JedisPoolConfig config = new JedisPoolConfig();  
2. config.setMaxTotal(32);  
3. config.setMaxIdle(6);  
4. config.setMinIdle(0);  
5. config.setMaxWaitMillis(15000);  
6.   
7. JedisShardInfo sd1 = new JedisShardInfo("127.0.0.1", 6379, 15000);  
8. sd1.setPassword("123456");  
9. JedisShardInfo sd2 = new JedisShardInfo("127.0.0.1", 6479, 15000);  
10. sd2.setPassword("123456");  
11. List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();  
12. shards.add(sd1);  
13. shards.add(sd2);  
14.   
15. ShardedJedisPool sjp = new ShardedJedisPool(config, shards);  
16. ShardedJedis shardedJedis = sjp.getResource();  
17. try {  
18.     System.out.println(shardedJedis.get("k2"));  
19.   
20.     ShardedJedisPipeline pipeline = new ShardedJedisPipeline();  
21.     pipeline.setShardedJedis(shardedJedis);  
22.     pipeline.set("k4", "v4");  
23.     pipeline.set("k5", "v5");  
24.     pipeline.get("k5");  
25.     List<Object> all = pipeline.syncAndReturnAll();  
26.     for (Object e : all) {  
27.         System.out.println(e);  
28.     }  
29. } catch (Exception e) {  
30.     e.printStackTrace();  
31. } finally {  
32.     sjp.returnResource(shardedJedis);  
33. }  

Xml代码  

1. <bean id="jedisPoolConfig" class="redis.clients.jedis.JedisPoolConfig">  
2.     <property name="maxActive" value="32"></property>  
3.     <property name="maxIdle" value="6"></property>  
4.     <property name="maxWait" value="15000"></property>  
5.     <property name="minEvictableIdleTimeMillis" value="300000"></property>  
6.     <property name="numTestsPerEvictionRun" value="3"></property>  
7.     <property name="timeBetweenEvictionRunsMillis" value="60000"></property>  
8.     <property name="whenExhaustedAction" value="1"></property>  
9. </bean>  
10. <bean id="shardedJedisPool" class="redis.clients.jedis.ShardedJedisPool" destroy-method="destroy">  
11.     <constructor-arg ref="jedisPoolConfig"></constructor-arg>  
12.     <constructor-arg>  
13.         <list>  
14.             <bean class="redis.clients.jedis.JedisShardInfo">  
15.                 <constructor-arg value="127.0.0.1"></constructor-arg>  
16.                 <constructor-arg value="6379"></constructor-arg>  
17.                 <property name="password" value="0123456"></property>  
18.             </bean>  
19.             <bean class="redis.clients.jedis.JedisShardInfo">  
20.                 <constructor-arg value="127.0.0.1"></constructor-arg>  
21.                 <constructor-arg value="6379"></constructor-arg>  
22.                 <property name="password" value="0123456"></property>  
23.             </bean>  
24.         </list>  
25.     </constructor-arg>  
26. </bean>  

    Something:

Something代码  

1. redis以及其他类似的网络IO server,实现绝对意义上的自动扩容(server端)和自动探测与rebalance,是很难的,同时也有一些风险.  
2.   
3. 我们现在的做法也比较土:  
4. 1) 有个web portal系统,当一个redis实例部署好之后,就是web系统上输入它的IP地址和探测脚本(脚本用来检测redis的内存负载情况,存活情况).  
5. 2) 录入之后可以将此redis"上线/下线",即将redis信息同步到zookeeper中(俗称configserver);  
6. 3) 所有redis-client端,都接入configserver,获取可用的redis列表;并初始化redis-client.  
7. 4) redis-client有一个额外的线程用来与configserver保持通讯,实时的跟踪redis列表的变更.  
8. 5) 如果redis列表变更,将导致redis-client端重新调整,主要是重建"一致性hash表".  
9. 6) 重建"一致性hash表"的过程,不需要调整代码或者重启服务,这个和hash的设计方式有些关系.  
10.   
11. 简单的来说,你可以使用任何方式(db,或者JMS订阅)来获取redis集群节点的变更数据即可..对于"客户端一致性hash表"的设计,也需要有些技巧,最好不要因为一个节点的join或者remove,导致大面积缓存的命中失败..  
12.   
13. 程序中通过合理的配置和编码，我们可以实现写master读slave。  

 本人通过查看公司应用系统的监控表明，redis几乎保持 set 2ms get 1ms, sql最快时 count 2ms select 3ms add/update 5ms

原文链接：[http://wely.iteye.com/blog/2275944]