redis一致性hash实践

之前在网上看过一致性hash原理，但是看过就忘记了，根本原因是没有理解透彻也没有实践，最近我负责了公司的会员业务，redis是开发自己搭建的，没有被DBA管理，最近要求迁移，这个工作就落在了我头上，为了确保迁移后缓存的key还能路由到原来的数据上，我把架构和源码研究了一番，发现竟然用了一致性hash,顿时眼前一亮，毕竟之前的系统没有这么用过，简单的就是主从+Sentinel,稍微复杂点用集群将数据分slot存储。

一、一致性hash要解决什么问题

一致性hash算法主要应用于分布式缓存系统中，可以有效地解决分布式存储结构下普通余数Hash算法伸缩性差的问题，可以保证在动态增加和删除节点的情况下尽量有多的请求命中原来路由到的节点。

对于分布式缓存，不同的节点存放不同的数据，先看一下普通余数算法的方式，假设有节点数n,缓存的健key,那么健为key的数据该存储到的节点序号index = hash(key) mod n ，但是当增加和删除节点时n会发生变化，造成命中率下降。

二、一致性hash算法原理

一致性hash算法将全量的缓存空间当作一个环型数据结构，总共有2^32个缓存区，环的起点是0，终点是2^32 - 1，并且起点与终点相连，整数按照顺时针分布，范围是[0, 2^32-1]。如下图所示：

那么如何将key和节点对应起来呢？

1）节点和key必须通过同一种hash算法映射到环上，即通过某种hash算法转换成一个32位的整数

将节点hash到环上，index = hash(node)，一般会用节点的ip或者name

将key也hash到环上，index = hash(key)

2）找到顺时针方向离key哈希值最近的节点，即大于等于key哈希值的第一个节点作为存储节点

那么我们来看一下增加、删除节点后对缓存系统的影响。

1.增加节点

如下图，增加node4节点，受到影响的仅仅是node1到node4之间的key，不再归node2管，归node4管理

2.删除节点

比如这张图，我们删除了node3，那么key将会存储到node1,所以受到影响的仅仅是node2到node3之间的key，即前一个节点到该节点之间的key。

如果仅仅是这样增加删除节点的命中率问题解决了，但是如果节点在环上分布不均匀，就会导致各个节点的负载不均衡，比如下图，所有的key都将存储到node2。

为了优化因为节点太少分布不均匀而导致的负载不均衡问题，一致性hash算法引入了虚拟节点的概念。
所谓虚拟节点,就是基于原来的物理节点映射出多个节点再映射到环形空间上，虚拟节点保存的机器信息是相同的，只不过hash到不不同的位置上。

三、jedis包的一致性hash实现源码分析

package redis.clients.util;

import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import redis.clients.util.Hashing;
import redis.clients.util.SafeEncoder;
import redis.clients.util.ShardInfo;

public class Sharded<R, S extends ShardInfo<R>> {
    public static final int DEFAULT_WEIGHT = 1;
    private TreeMap<Long, S> nodes;//hash环，保存所有节点，key就是hash后的值，value是节点的信息
    private final Hashing algo;//hash算法类
    private final Map<ShardInfo<R>, R> resources;
    private Pattern tagPattern;
    public static final Pattern DEFAULT_KEY_TAG_PATTERN = Pattern.compile("\\{(.+?)\\}");

    public Sharded(List<S> shards) {
        this(shards, Hashing.MURMUR_HASH);
    }

    public Sharded(List<S> shards, Hashing algo) {
        this.resources = new LinkedHashMap();
        this.tagPattern = null;
        this.algo = algo;
        this.initialize(shards);
    }

    public Sharded(List<S> shards, Pattern tagPattern) {
        this(shards, Hashing.MURMUR_HASH, tagPattern);
    }

    public Sharded(List<S> shards, Hashing algo, Pattern tagPattern) {
        this.resources = new LinkedHashMap();
        this.tagPattern = null;
        this.algo = algo;
        this.tagPattern = tagPattern;
        this.initialize(shards);
    }

    private void initialize(List<S> shards) {//节点的初始化 最终多个物理节点的“虚拟节点”将会在环上交错布局，不一定分布均匀。
        this.nodes = new TreeMap();

        for(int i = 0; i != shards.size(); ++i) {
            ShardInfo shardInfo = (ShardInfo)shards.get(i);
            int n;
            if(shardInfo.getName() == null) {//判断节点的名称为空 就给默认名称计算hash值
                for(n = 0; n < 160 * shardInfo.getWeight(); ++n) {//默认权重为1，每个物理节点有160个虚拟节点
                    this.nodes.put(Long.valueOf(this.algo.hash("SHARD-" + i + "-NODE-" + n)), shardInfo);
                }
            } else {//判断节点的名称不为空 就用名称计算hash值
                for(n = 0; n < 160 * shardInfo.getWeight(); ++n) {
                    this.nodes.put(Long.valueOf(this.algo.hash(shardInfo.getName() + "*" + shardInfo.getWeight() + n)), shardInfo);
                }
            }

            this.resources.put(shardInfo, shardInfo.createResource());
        }

    }

    public R getShard(byte[] key) {
        return this.resources.get(this.getShardInfo(key));
    }

    public R getShard(String key) {
        return this.resources.get(this.getShardInfo(key));
    }

    public S getShardInfo(byte[] key) {//获取key对应的存储节点
        SortedMap tail = this.nodes.tailMap(Long.valueOf(this.algo.hash(key)));//获取到大于等于key哈希值的子map
        return tail.isEmpty()?(ShardInfo)this.nodes.get(this.nodes.firstKey()):(ShardInfo)tail.get(tail.firstKey());//如果子map为空就取map的第一个节点，否则取子map的第一个节点
    }

    public S getShardInfo(String key) {
        return this.getShardInfo(SafeEncoder.encode(this.getKeyTag(key)));
    }

    public String getKeyTag(String key) {
        if(this.tagPattern != null) {
            Matcher m = this.tagPattern.matcher(key);
            if(m.find()) {
                return m.group(1);
            }
        }

        return key;
    }

    public Collection<S> getAllShardInfo() {
        return Collections.unmodifiableCollection(this.nodes.values());
    }

    public Collection<R> getAllShards() {
        return Collections.unmodifiableCollection(this.resources.values());
    }
}

再来看看hash算法的实现源码

package redis.clients.util;

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import redis.clients.util.MurmurHash;
import redis.clients.util.SafeEncoder;
/**
* 采用MD5加密算法的一种哈希算法实现，安全性高，但是速度慢。
*/
public interface Hashing {
    Hashing MURMUR_HASH = new MurmurHash();//这个变量放在这里感觉没用啊
    ThreadLocal<MessageDigest> md5Holder = new ThreadLocal();
    Hashing MD5 = new Hashing() {
        public long hash(String key) {
            return this.hash(SafeEncoder.encode(key));
        }

        public long hash(byte[] key) {
            try {
                if(md5Holder.get() == null) {
                    md5Holder.set(MessageDigest.getInstance("MD5"));
                }
            } catch (NoSuchAlgorithmException var6) {
                throw new IllegalStateException("++++ no md5 algorythm found");
            }

            MessageDigest md5 = (MessageDigest)md5Holder.get();
            md5.reset();
            md5.update(key);
            byte[] bKey = md5.digest();//先用md5计算摘要，一个字节数组
            //前四个字节作为计算参数，最终获得一个32位int值.  
            //此种计算方式，能够确保key的hash值更加“随即”/“离散”  
            //如果hash值过于密集，不利于一致性hash的实现(特别是有“虚拟节点”设计时)  
            long res = (long)(bKey[3] & 255) << 24 | (long)(bKey[2] & 255) << 16 | (long)(bKey[1] & 255) << 8 | (long)(bKey[0] & 255);//利用摘要的四个字节得到一个离散性较好的一个32位整数
            return res;
        }
    };

    long hash(String var1);

    long hash(byte[] var1);
}

package redis.clients.util;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import redis.clients.util.Hashing;
import redis.clients.util.SafeEncoder;
/**
* MurmurHash是一种非加密型哈希函数，适用于一般的哈希检索操作,与其它流行的哈希函数相比，对于规律 
* 性较强的key，MurmurHash的随机分布（离散性）特征表现更好，而且具有低碰撞率。Redis对节点进行  
* shard时采用的是这种算法。
*/

public class MurmurHash implements Hashing {
    public MurmurHash() {
    }

    public static int hash(byte[] data, int seed) {
        return hash(ByteBuffer.wrap(data), seed);
    }

    public static int hash(byte[] data, int offset, int length, int seed) {
        return hash(ByteBuffer.wrap(data, offset, length), seed);
    }

    public static int hash(ByteBuffer buf, int seed) {
        ByteOrder byteOrder = buf.order();
        buf.order(ByteOrder.LITTLE_ENDIAN);
        int m = 1540483477;
        byte r = 24;

        int h;
        int k;
        for(h = seed ^ buf.remaining(); buf.remaining() >= 4; h ^= k) {
            k = buf.getInt();
            k *= m;
            k ^= k >>> r;
            k *= m;
            h *= m;
        }

        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;
    }

    public static long hash64A(byte[] data, int seed) {
        return hash64A(ByteBuffer.wrap(data), seed);
    }

    public static long hash64A(byte[] data, int offset, int length, int seed) {
        return hash64A(ByteBuffer.wrap(data, offset, length), seed);
    }

    public static long hash64A(ByteBuffer buf, int seed) {
        ByteOrder byteOrder = buf.order();
        buf.order(ByteOrder.LITTLE_ENDIAN);
        long m = -4132994306676758123L;
        byte r = 47;

        long h;
        for(h = (long)seed ^ (long)buf.remaining() * m; buf.remaining() >= 8; h *= m) {
            long k = buf.getLong();
            k *= m;
            k ^= k >>> r;
            k *= m;
            h ^= k;
        }

        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;
    }

    public long hash(byte[] key) {
        return hash64A(key, 305441741);
    }

    public long hash(String key) {
        return this.hash(SafeEncoder.encode(key));
    }
}

四、我们项目中怎么用的

我们的架构是n台主服务器（带有从服务器），用m台Sentinel，程序是通过连接Sentinel的，Sentinel通过master name 获取到主服务器的ip和端口号信息，将所有的主服务器信息保存在一个列表中，通过master name拼接权重、递增序号hash为多个虚拟节点，最好不要用ip取hash，因为如果迁移redis换了ip就没办法保证迁移后缓存的key能路由到同步后的数据上，同样缓存的key也是用同一个hash算法hash为一个整数，通过SortedmMap的tailMap获取到子Map取第一个大于等于hash值的节点作为存储节点。

部署架构图类似这样，为了简单master和sentinel只画了三个节点。

redis迁移步骤
1、在老库bgsave 产生rdb文件，然后加载到新的master 实例，三个实例总耗时大概5-10分钟，由DBA操作；同步过程中新的数据库是不可以访问的，如果访问会报LOADING Redis is loading the dataset in memory。
2、业务指向新的业务库；修改vip-shared-redis-client项目的配置文件redis-deploy.xml，业务用到了一致性hash算法来路由
，redis主服务器是通过name来hash的，DBA那边不支持修改主库的name，为了确保迁移后缓存的key还能路由到相同的数据节点，所以业务配置文件加了一个map将新的主库name映射到旧的主库name上来保证。
3、观察老的业务库请求和连接情况；

风险：
1、中间有部分数据丢失；
2、如果迁移周期很长，意味着某些业务进程丢失数据的时间窗口越长；
3、如果中途发现业务进程迁移有问题，比如（业务进程耦合严重），导致某些业务进程没有迁移，这些业务进程又不能丢失数据，
此时需要回退，回退的过程新库和老库的数据差异问题，会再次丢失数据；
迁移那天我发了十来个项目，幸运的是一切顺利，感恩！接下来看一下原理部分。

1）先分析存储节点的hash过程。
配置文件：

<bean id="shardedJedisPool" class="redis.clients.jedis.ShardedJedisSentinelPool2">
  <constructor-arg index="0" ref="shardedJedisPoolConfig" />
  <constructor-arg index="1">
   <list>
    <value>vip_redis_002</value>
    <value>vip_redis_001</value>
    <value>vip_redis_003</value>
   </list>
  </constructor-arg>
  <constructor-arg index="2">
   <set>
    <value>groupbj-sentinel.yy.com:20188</value>
    <value>group1sz-sentinel.yy.com:20188</value>
    <value>groupwx-sentinel.yy.com:20188</value>
   </set>
  </constructor-arg>
  <constructor-arg index="3">
   <value></value> <!-- password -->
  </constructor-arg>
  <constructor-arg index="4">
   <value>false</value> <!-- useSlavesReader -->
  </constructor-arg>

  <constructor-arg index="5">
   <map> <!-- masterNameMap -->
    <entry key="vip_redis_002" value="vip_service_redis_01"></entry>
    <entry key="vip_redis_001" value="vip_service_redis_02"></entry>
    <entry key="vip_redis_003" value="vip_service_redis_03"></entry>
   </map>
  </constructor-arg>
 </bean>

通过debug的方式梳理了一下流程。

1、redis自定义连接池构造函数：

public ShardedJedisSentinelPool2(List<String> masters, Set<String> sentinels,
   final GenericObjectPoolConfig poolConfig, int timeout, final String password, final int database,
   final boolean useSlavesReader , final Map<String,String> masterNameMap) {
 this.poolConfig = poolConfig;
 this.timeout = timeout;
 if(password!=null && password.length()>0 && !password.equals("null")){
   this.password = password;
 }else{
   this.password=null;
 }
 this.database = database;
 this.sentinels = sentinels;
 this.currentMasterNames = masters;
 this.useSlavesReader = useSlavesReader;
 this.masterNameMap = masterNameMap;
 log.info("ShardedJedisSentinelPool2 masterNameMap:"+JsonUtil.object2JsonString(masterNameMap));//2
 List<SentinelHostAndPort> masterList = initSentinels(sentinels, masters);//3
 initPool(masterList);
  }

2.List<SentinelHostAndPort> masterList = initSentinels(sentinels, masters); 根据配置的sentinel和主库名称查询主库节点的ip和端口号信息并组装成主库对象

private List<SentinelHostAndPort> initSentinels(Set<String> sentinels, final List<String> masters) {

 Map<String, SentinelHostAndPort> masterMap = new HashMap<String, SentinelHostAndPort>();
 List<SentinelHostAndPort> shardMasters = new ArrayList<SentinelHostAndPort>();

 log.info("Trying to find all master from available Sentinels...");

 for (String masterName : masters) {//遍历配置的主库name列表
   SentinelHostAndPort master = null;
   boolean fetched = false;

   while (!fetched && sentinelRetry < MAX_RETRY_SENTINEL) {
  for (String sentinel : sentinels) {//遍历sentinel的ip和端口号列表
    final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));

    log.info("Connecting to Sentinel " + hap);

    try {
   Jedis jedis = new Jedis(hap.getHost(), hap.getPort());//建立一个sentinel连接
   master = masterMap.get(masterName);
   if (master == null) {
     List<String> hostAndPort = jedis.sentinelGetMasterAddrByName(masterName);//通过sentinel根据主库名称查询主库的ip和端口号

     if (hostAndPort != null && hostAndPort.size() > 0) {
    master = toSentinelHostAndPort(hostAndPort, masterName);//new一个主库对象
    log.info("Found Redis master at " + master);
    shardMasters.add(master);//分片的物理节点增加新创建的主库对象
    masterMap.put(masterName, master);//将主库对象保存在一个map中
    fetched = true;
    jedis.disconnect();//sentinel断开连接
    break;
     }
   }
    } catch (JedisConnectionException e) {
   log.info("Cannot connect to sentinel running @ " + hap + ". Trying next one.");
    }
  }

3、initPool(masterList);

 private void initPool(List<SentinelHostAndPort> masters) {
 if (!equals(currentHostMasters, masters)) {
   StringBuffer sb = new StringBuffer();
   for (SentinelHostAndPort master : masters) {
  sb.append(master.toString());
  sb.append(" ");
   }
   log.info("Created ShardedJedisPool to master at [" + sb.toString() + "]");
   List<JedisShardInfo> shardMasters = makeShardInfoList(masters);//4

   Map<String, SlaveJedisSentinelPool> slavePoolMap = initSlavesPools(this.currentMasterNames, this.sentinels,
    this.poolConfig, this.timeout, this.password, this.database);//5 此处不关注

   initPool(poolConfig, new ShardedJedisFactory(slavePoolMap, shardMasters, Hashing.MURMUR_HASH, null,
    password,this.useSlavesReader));//6
   currentHostMasters = masters;
 }
  }

4.List<JedisShardInfo> shardMasters = makeShardInfoList(masters); //组装分片的节点对象信息

private List<JedisShardInfo> makeShardInfoList(List<SentinelHostAndPort> masters) {
 List<JedisShardInfo> shardMasters = new ArrayList<JedisShardInfo>();
 for (SentinelHostAndPort master : masters) {
   // 这里加上master name 作为名称， 保证在一致性hash中不会改变hash规则
   SentinelJedisShardInfo jedisShardInfo = new SentinelJedisShardInfo(master.getHost(), master.getPort(),
    timeout,getMappedMasterName(master.getMasterName()));//此处设置的name会参与hash计算
   jedisShardInfo.setPassword(password);
   shardMasters.add(jedisShardInfo);
 }
 log.info("makeShardInfoList shardMasters:"+JsonUtil.object2JsonString(shardMasters));
 return shardMasters;
  }

  private String getMappedMasterName(String masterName){//将新的主库名称映射到旧的主库名称
  if (MapUtils.isNotEmpty(this.masterNameMap)){
   String mappedMasterName = this.masterNameMap.get(masterName);
   if (StringUtil.isNotEmpty(mappedMasterName)){
    return mappedMasterName;
   }
  }
  return masterName;
 }

6.initPool(poolConfig, new ShardedJedisFactory(slavePoolMap, shardMasters, Hashing.MURMUR_HASH, null,
password,this.useSlavesReader));

public void initPool(GenericObjectPoolConfig poolConfig, PooledObjectFactory<T> factory) {
        if(this.internalPool != null) {
            try {
                this.closeInternalPool();
            } catch (Exception var4) {
                ;
            }
        }

        this.internalPool = new GenericObjectPool(factory, poolConfig);
    }

7，我是通过打断点调试的，目前没有找到哪里调了这个方法

public PooledObject<RabbitSharedJedis> makeObject() throws Exception {
   RabbitSharedJedis jedis = new RabbitSharedJedis(shards, algo, keyTagPattern, this.useSlavesReader);//8

   jedis.setSlavePoolMap(slavePoolMap);
   return new DefaultPooledObject<RabbitSharedJedis>(jedis);
 }

8. RabbitSharedJedis jedis = new RabbitSharedJedis(shards, algo, keyTagPattern, this.useSlavesReader);初始化一个RabbitSharedJedis 对象

public RabbitSharedJedis(List<JedisShardInfo> shards, Hashing algo, Pattern keyTagPattern,
   final boolean useSlavesReader) {//RabbitSharedJedis.class
 super(shards, algo, keyTagPattern);//9
 this.useSlavesReader = useSlavesReader;
  }

9.调用父类ShardedJedis的构造方法

public ShardedJedis(List<JedisShardInfo> shards, Hashing algo, Pattern keyTagPattern) {//ShardedJedis.class
    super(shards, algo, keyTagPattern);//10
  }

10.super(shards, algo, keyTagPattern);调用父类BinaryShardedJedis的构造函数

public BinaryShardedJedis(List<JedisShardInfo> shards, Hashing algo, Pattern keyTagPattern) {//BinaryShardedJedis.class
    super(shards, algo, keyTagPattern);//11
  }

11.super(shards, algo, keyTagPattern);

public Sharded(List<S> shards, Hashing algo, Pattern tagPattern) {
    this.resources = new LinkedHashMap();
    this.tagPattern = null;
    this.algo = algo;
    this.tagPattern = tagPattern;
    this.initialize(shards);//12
  }

12.this.initialize(shards);重点，初始化环形结构

 private void initialize(List<S> shards) {
    this.nodes = new TreeMap();

    for(int i = 0; i != shards.size(); ++i) {
      ShardInfo shardInfo = (ShardInfo)shards.get(i);
      int n;
      if(shardInfo.getName() == null) {
        for(n = 0; n < 160 * shardInfo.getWeight(); ++n) {
          this.nodes.put(Long.valueOf(this.algo.hash("SHARD-" + i + "-NODE-" + n)), shardInfo);
        }
      } else {
        for(n = 0; n < 160 * shardInfo.getWeight(); ++n) {
          this.nodes.put(Long.valueOf(this.algo.hash(shardInfo.getName() + "*" + shardInfo.getWeight() + n)), shardInfo);
        }
      }

      this.resources.put(shardInfo, shardInfo.createResource());
    }

  }

13.this.algo.hash(shardInfo.getName() + "" + shardInfo.getWeight() + n)*

public long hash(String key) {
    return this.hash(SafeEncoder.encode(key));//14
  }

14.return this.hash(SafeEncoder.encode(key));

 public long hash(byte[] key) {
    return hash64A(key, 305441741); //15
  }

15.return hash64A(key, 305441741);

public static long hash64A(byte[] data, int seed) {
    return hash64A(ByteBuffer.wrap(data), seed);//16
  }

16.return hash64A(ByteBuffer.wrap(data), seed);

 public static long hash64A(ByteBuffer buf, int seed) {
    ByteOrder byteOrder = buf.order();
    buf.order(ByteOrder.LITTLE_ENDIAN);
    long m = -4132994306676758123L;
    byte r = 47;

    long h;
    for(h = (long)seed ^ (long)buf.remaining() * m; buf.remaining() >= 8; h *= m) {
      long k = buf.getLong();
      k *= m;
      k ^= k >>> r;
      k *= m;
      h ^= k;
    }

    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;
  }

2）客户端查询存储节点

1. 测试代码

public void testShardRedis() throws Exception {
ApplicationContext ac = new ClassPathXmlApplicationContext("classpath:/redis-deploy.xml");

 ShardedJedisSentinelPool2 pool = (ShardedJedisSentinelPool2) ac.getBean("shardedJedisPool");

 RabbitSharedJedis j = null;

 Long date1 = System.currentTimeMillis();
  String key = "key";
  try {
   j = pool.getResource();
   String value = j.get(key);//2
   System.out.println("result:"+value);
  }catch (JedisConnectionException e) {
   System.out.print("error，"+e.getMessage());
  }
 }

2.String value = j.get(key);

public String get(String key) {
 Jedis j = getShard(key);//3
 SlaveJedisSentinelPool jedisSlavePool = this.getSlavePoolByMasterJedis(j);
 Jedis slaveJedis = null;
 ListSlaveJedis listSlaveJedis = null;
 String result;
 if (jedisSlavePool != null) {
   try {
  listSlaveJedis = jedisSlavePool.getResource();
  slaveJedis = listSlaveJedis.getRandomCanUseSlaveJedis();
  if(slaveJedis!=null){
    result = slaveJedis.get(key);
  }else{
    result = j.get(key);
  }
   } catch (Exception e) {
  log.error("no slave can use", e);
  result = j.get(key);
  //throw new SentinalRedisException("sentinal get error");
   } finally {
  if (listSlaveJedis != null) {
    jedisSlavePool.returnResource(listSlaveJedis);
  }
   }
 } else {
   result = j.get(key);
 }
 return result;
  }

3.Jedis j = getShard(key);

public R getShard(String key) {
    return this.resources.get(this.getShardInfo(key));//4
  }

4.return this.resources.get(this.getShardInfo(key));

public S getShardInfo(String key) {
    return this.getShardInfo(SafeEncoder.encode(this.getKeyTag(key)));//5
  }

5.return this.getShardInfo(SafeEncoder.encode(this.getKeyTag(key)));

 public S getShardInfo(byte[] key) {
    SortedMap tail = this.nodes.tailMap(Long.valueOf(this.algo.hash(key)));
    return tail.isEmpty()?(ShardInfo)this.nodes.get(this.nodes.firstKey()):(ShardInfo)tail.get(tail.firstKey());
  }