Jedis客户端分片的实现

对于单实例的Redis的使用，我们可以用Jedis，并发环境下我们可以用JedisPool。但是这两种方法否是针对于单实例的Redis的情况下使用的，但是有时候我们的业务可能不是单实例Redis能支撑的，那么我们这时候需要引入多个实例进行“数据分区”。其实好多人都说，用Redis集群不就搞定了吗？但是Redis集群无论部署还是维护成本都比较高，对于一些业务来说，使用起来还是成本很高。所以，对我们来说更好的方案可能是在客户端实现对数据的手动分区.

对于分区的方案，我感觉大多数人都会想到Hash，的确Hash是最简单最有效的方式。但是Hash的问题是：“单节点挂掉不可用，数据量大了不好扩容”。对于如果业务的可靠性要求不高同时数据可控的情况下可以考虑数据分区的方式。

其实数据分区就是Shard，其实Redis已经对Shard有很好的支持了，接下来简单的搞一下数据分片：

package redis.clients.jedis.tests;
import org.junit.Before;
import org.junit.Test;
import redis.clients.jedis.*;
import java.util.ArrayList;
import java.util.List;
/**
 * ShardJedis的测试类
 */
public class ShardJedisTest {
 private ShardedJedisPool sharedPool;
 @Before
 public void initJedis(){
 JedisPoolConfig config =new JedisPoolConfig();//Jedis池配置
 config.setTestOnBorrow(true);
 String hostA = "127.0.0.1";
 int portA = 6381;
 String hostB = "127.0.0.1";
 int portB = 6382;
 List<JedisShardInfo> jdsInfoList =new ArrayList<JedisShardInfo>(2);
 JedisShardInfo infoA = new JedisShardInfo(hostA, portA);
 JedisShardInfo infoB = new JedisShardInfo(hostB, portB);
 jdsInfoList.add(infoA);
 jdsInfoList.add(infoB);
 sharedPool =new ShardedJedisPool(config, jdsInfoList);
 }
 @Test
 public void testSetKV() throws InterruptedException {
 try {
 for (int i=0;i<50;i++){
 String key = "test"+i;
 ShardedJedis jedisClient = sharedPool.getResource();
 System.out.println(key+":"+jedisClient.getShard(key).getClient().getHost()+":"+jedisClient.getShard(key).getClient().getPort());
 System.out.println(jedisClient.set(key,Math.random()+""));
 jedisClient.close();
 }
 }catch (Exception e){
 e.printStackTrace();
 }
 }
}

这里我是用JUnit做的测试，我在本机开了两个Redis实例：

端口号分别是6381和6382。然后用ShardedJedisPool实现了一个Shard，主要是生成了50个Key，分别存到Redis中。运行结果如下：

test0:127.0.0.1:6382
OK
test1:127.0.0.1:6382
OK
test2:127.0.0.1:6381
OK
test3:127.0.0.1:6382
OK
test4:127.0.0.1:6382
OK
test5:127.0.0.1:6382
OK
test6:127.0.0.1:6382
OK
test7:127.0.0.1:6382
OK
test8:127.0.0.1:6381
OK
test9:127.0.0.1:6381

可以看到，KV分别分发到了不同的Redis实例，这种Shard的方式需要我们提前计算好数据量的大小，便于决定实例的个数。同时这种shard的可靠性不是很好，如果单个Redis实例挂掉了，那么这个实例便不可用了。

其实Shard使用起来很简单，接下来我们看看ShardedJedisPool的具体的实现：

首先在初始化ShardedJedisPool的时候我们需要创建一个JedisShardInfo实例，JedisShardInfo主要是对单个连接的相关配置：

public class JedisShardInfo extends ShardInfo<Jedis> {
 private static final String REDISS = "rediss";
 private int connectionTimeout;
 private int soTimeout;
 private String host;
 private int port;
 private String password = null;
 private String name = null;
 // Default Redis DB
 private int db = 0;
 private boolean ssl;
 private SSLSocketFactory sslSocketFactory;
 private SSLParameters sslParameters;
 private HostnameVerifier hostnameVerifier;

像连接超时时间、发送超时时间、Host和port等。这些都是之前我们实例化Jedis用到的。

同时还需要进行JedisPoolConfig的设置，可以猜到ShardedJedisPool也是基于JedisPool来实现的。

看看ShardedJedisPool的构造：

public ShardedJedisPool(final GenericObjectPoolConfig poolConfig, List<JedisShardInfo> shards) {
 this(poolConfig, shards, Hashing.MURMUR_HASH);
 }
 public ShardedJedisPool(final GenericObjectPoolConfig poolConfig, List<JedisShardInfo> shards,
 Hashing algo) {
 this(poolConfig, shards, algo, null);
 }
 public ShardedJedisPool(final GenericObjectPoolConfig poolConfig, List<JedisShardInfo> shards,
 Hashing algo, Pattern keyTagPattern) {
 super(poolConfig, new ShardedJedisFactory(shards, algo, keyTagPattern));
 }
 public Pool(final GenericObjectPoolConfig poolConfig, PooledObjectFactory<T> factory) {
 initPool(poolConfig, factory);
 }
 public void initPool(final GenericObjectPoolConfig poolConfig, PooledObjectFactory<T> factory) {
 if (this.internalPool != null) {
 try {
 closeInternalPool();
 } catch (Exception e) {
 }
 }
 this.internalPool = new GenericObjectPool<T>(factory, poolConfig);
 }

构造方法很长，但是很清晰，关键点在ShardedJedisFactory的构建，因为这是使用commons-pool的必要工厂类。同时我们可以看到，这里分分片策略使用的确实是Hash，而且还是冲突率很低的MURMUR_HASH。这里不了解commons-pool的可以看一下之前的Commons-pool源码分析[http://www.jianshu.com/p/b49452fb3a67]

那么我们直接看ShardedJedisFactory类就好了，因为commons-pool就是基于这个工厂类来管理相关的对象的，这里缓存的对象是ShardedJedis

我们先看一下ShardedJedisFactory：

public ShardedJedisFactory(List<JedisShardInfo> shards, Hashing algo, Pattern keyTagPattern) {
 this.shards = shards;
 this.algo = algo;
 this.keyTagPattern = keyTagPattern;
 }
 @Override
 public PooledObject<ShardedJedis> makeObject() throws Exception {
 ShardedJedis jedis = new ShardedJedis(shards, algo, keyTagPattern);
 return new DefaultPooledObject<ShardedJedis>(jedis);
 }
 @Override
 public void destroyObject(PooledObject<ShardedJedis> pooledShardedJedis) throws Exception {
 final ShardedJedis shardedJedis = pooledShardedJedis.getObject();
 for (Jedis jedis : shardedJedis.getAllShards()) {
 try {
 try {
 jedis.quit();
 } catch (Exception e) {
 }
 jedis.disconnect();
 } catch (Exception e) {
 }
 }
 }
 
 @Override
 public boolean validateObject(PooledObject<ShardedJedis> pooledShardedJedis) {
 try {
 ShardedJedis jedis = pooledShardedJedis.getObject();
 for (Jedis shard : jedis.getAllShards()) {
 if (!shard.ping().equals("PONG")) {
 return false;
 }
 }
 return true;
 } catch (Exception ex) {
 return false;
 }
 }

其实这里makeObject是创建一个ShardedJedis，同时ShardedJedis也是连接池里保存的对象。

可以看到destroyObject和validateObject都是将ShardedJedis里的redis实例当做了一个整体去对待，一个失败，全部失败。

接下来看下ShardedJedis的实现，这个里面主要做了Hash的处理和各个Shard的Client的缓存。

public class ShardedJedis extends BinaryShardedJedis implements JedisCommands, Closeable {
 protected ShardedJedisPool dataSource = null;
 public ShardedJedis(List<JedisShardInfo> shards) {
 super(shards);
 }
 public ShardedJedis(List<JedisShardInfo> shards, Hashing algo) {
 super(shards, algo);
 }
 public ShardedJedis(List<JedisShardInfo> shards, Pattern keyTagPattern) {
 super(shards, keyTagPattern);
 }
 public ShardedJedis(List<JedisShardInfo> shards, Hashing algo, Pattern keyTagPattern) {
 super(shards, algo, keyTagPattern);
 }

这里的dataSource是对连接池的引用，用于在Close的时候资源返还。和JedisPool的思想差不多。

由于ShardedJedis是BinaryShardedJedis的子类，所以构造函数会一直向上调用，在Shard中：

public Sharded(List<S> shards, Hashing algo, Pattern tagPattern) {
 this.algo = algo;
 this.tagPattern = tagPattern;
 initialize(shards);
 }
 
 private void 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());
 }
 }

这里主要做整个ShardedJedis中Jedis缓存池的初始化和分片的实现，可以看到首先获取shardInfo就是之前的JedisShardInfo，根据shardInfo生成多个槽位，将这些槽位存到TreeMap中，同时将shardInfo和Jedis的映射存到resources中。当我们做Client的获取的时候：

首先调用ShardedJedisPool的getResource方法，从对象池中获取一个ShardedJedis：

ShardedJedis jedisClient = sharedPool.getResource();

调用ShardedJedis的getShard方法获取一个Jedis实例——一个shard。

public R getShard(String key) {
 return resources.get(getShardInfo(key));
 }
 public S getShardInfo(String key) {
 return getShardInfo(SafeEncoder.encode(getKeyTag(key)));
 }
 public S getShardInfo(byte[] key) {
 SortedMap<Long, S> tail = nodes.tailMap(algo.hash(key));
 if (tail.isEmpty()) {
 return nodes.get(nodes.firstKey());
 }
 return tail.get(tail.firstKey());
 }

这里主要是对key做hash，然后去TreeMap中判断，当前的key落在哪个区间上，再通过这个区间上的ShardInfo从resources的Map中获取对应的Jedis实例。

这也就是说，每一个ShardedJedis都维护了所有的分片，将多个实例当成一个整体去使用，这也就导致，只要集群中一个实例不可用，整个ShardedJedis就不可用了。同时对于hash的分片方式，是不可扩容的，扩容之后原本应该存储在一起的数据就分离了。

其实这种是Jedis默认提供的分片方式，其实针对我们自己的场景我们也可以尝试自己做一个路由机制，例如根据不同年份、月份的数据落到一个实例上。

上面就是所有的数据分片的jedis实现的分析，我们线上的业务也是基于ShardedJedis来实现的，由于线上业务的QPS不高，量也不是很大，所以运行还算平稳。