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;@Beforepublic 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);}@Testpublic 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 DBprivate 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;}@Overridepublic PooledObject<ShardedJedis> makeObject() throws Exception {ShardedJedis jedis = new ShardedJedis(shards, algo, keyTagPattern);return new DefaultPooledObject<ShardedJedis>(jedis);}@Overridepublic 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) {}}}@Overridepublic 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