Redis基于客户端分片的集群案例（待实践）

说明：

下面的示例基本都是基于Linux去实现，目的是为了环境的统一，以便于把性能调整到最优。且基于Java。建议生产环境不要使用Windows/Mac OS这些。

在Java领域，基于客户端进行分片最常用的库应该是Jedis，下面基本是基于Jedis进行实例实践。当然，除了这个还可以基于自己的业务去实现。

现在官方已经出到了4.0版本，也同样支持了集群功能，那么现在市面上基本不用客户端去实现分片做集群，主要集中在服务端来达到高可用的Redis集群，所以，是否有必要客户端去实现集群，需要在自己的业务上来深入考究。

同样的，除了官方集群外，还有很多成熟的方案去实现服务端集群，比如推特、豌豆荚这些官方开源的方案等。

在客户端进行分片来达到集群的效果，最简单的理解应该是这样：A和B两个Key，通过Hash得到A放在Redis1，B放在Redis2中。（先忽略Redis其中一台挂掉的问题，对于算法远没有在这里说的那么简单）。

分片的大致原理都是基于Hash算法来制定哪个Key放到哪个Redis中。

在这篇http://www.cnblogs.com/EasonJim/p/7625738.html文章中提到的几款客户端中都已经实现了分片的操作。

下面是基于Jedis去实现了客户端分片功能配置：

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

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

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

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。

那么我们直接看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默认提供的分片方式，其实针对我们自己的场景我们也可以尝试自己做一个路由机制，例如根据不同年份、月份的数据落到一个实例上。

而对于在Spring中集成Jedis分片来说，应该是做简单的：

1、在properties中定义其它Redis

redis.host2=192.168.142.34

2、注入Bean

<bean id = "shardedJedisPool" class = "redis.clients.jedis.ShardedJedisPool">

    <constructor-arg index="0" ref="jedisPoolConfig"/>

    <constructor-arg index="1">

        <list>

            <bean class="redis.clients.jedis.JedisShardInfo">

                <constructor-arg index="0" value="${redis.host}"/>

                <constructor-arg index="1" value="${redis.port}" type="int"/>

                <constructor-arg index="2" value="${redis.timeout}" type="int"/>

                <property name="password" value="${redis.password}"/>

            </bean>

            <bean class="redis.clients.jedis.JedisShardInfo">

                <constructor-arg index="0" value="${redis.host2}"/>

                <constructor-arg index="1" value="${redis.port}" type="int"/>

                <constructor-arg index="2" value="${redis.timeout}" type="int"/>

                <property name="password" value="${redis.password}"/>

            </bean>

        </list>

    </constructor-arg>

</bean>

3、代码使用

//获取Bean

ShardedJedisPool shardedPool = (ShardedJedisPool)context.getBean("shardedJedisPool");

ShardedJedis shardedJedis = shardedPool.getResource();

 ...

shardedPool.returnResource(shardedJedis);

//操作

shardedJedis.set("test", "123");

String president = shardedJedis.get("test");

shardedJedis.del("test");

参考：

http://www.jianshu.com/p/af0ea8d61dda（以上内容转自此篇文章）

http://www.jianshu.com/p/37b5b6cdb277

http://www.jianshu.com/p/a1038eed6d44

http://blog.csdn.net/yfkiss/article/details/38944179

http://hello-nick-xu.iteye.com/blog/2078153（以上内容部分转自此篇文章）

http://blog.csdn.net/benxiaohai529/article/details/52935216

http://blog.csdn.net/mid120/article/details/52799241

http://blog.csdn.net/lang_man_xing/article/details/38405269

http://www.cnblogs.com/hk315523748/p/6122263.html

http://ihenu.iteye.com/blog/2267881

http://blog.csdn.net/koushr/article/details/50956870

巴特西

Redis基于客户端分片的集群案例（待实践）

最新文章

热门文章