Jedis客戶端分片的實現(xiàn)

對于單實例的Redis的使用，我們可以用Jedis，并發(fā)環(huán)境下我們可以用JedisPool。但是這兩種方法否是針對于單實例的Redis的情況下使用的，但是有時候我們的業(yè)務(wù)可能不是單實例Redis能支撐的，那么我們這時候需要引入多個實例進(jìn)行“數(shù)據(jù)分區(qū)”。其實好多人都說，用Redis集群不就搞定了嗎？但是Redis集群無論部署還是維護(hù)成本都比較高，對于一些業(yè)務(wù)來說，使用起來還是成本很高。所以，對我們來說更好的方案可能是在客戶端實現(xiàn)對數(shù)據(jù)的手動分區(qū).

對于分區(qū)的方案，我感覺大多數(shù)人都會想到Hash，的確Hash是最簡單最有效的方式。但是Hash的問題是：“單節(jié)點掛掉不可用，數(shù)據(jù)量大了不好擴(kuò)容”。對于如果業(yè)務(wù)的可靠性要求不高同時數(shù)據(jù)可控的情況下可以考慮數(shù)據(jù)分區(qū)的方式。

其實數(shù)據(jù)分區(qū)就是Shard，其實Redis已經(jīng)對Shard有很好的支持了，接下來簡單的搞一下數(shù)據(jù)分片：

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做的測試，我在本機(jī)開了兩個Redis實例：

Paste_Image.png

端口號分別是6381和6382。然后用ShardedJedisPool實現(xiàn)了一個Shard，主要是生成了50個Key，分別存到Redis中。運行結(jié)果如下：

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分別分發(fā)到了不同的Redis實例，這種Shard的方式需要我們提前計算好數(shù)據(jù)量的大小，便于決定實例的個數(shù)。同時這種shard的可靠性不是很好，如果單個Redis實例掛掉了，那么這個實例便不可用了。

其實Shard使用起來很簡單，接下來我們看看ShardedJedisPool的具體的實現(xiàn)：

首先在初始化ShardedJedisPool的時候我們需要創(chuàng)建一個JedisShardInfo實例，JedisShardInfo主要是對單個連接的相關(guān)配置：

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;

像連接超時時間、發(fā)送超時時間、Host和port等。這些都是之前我們實例化Jedis用到的。
同時還需要進(jìn)行JedisPoolConfig的設(shè)置，可以猜到ShardedJedisPool也是基于JedisPool來實現(xiàn)的。
看看ShardedJedisPool的構(gòu)造：

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

構(gòu)造方法很長，但是很清晰，關(guān)鍵點在ShardedJedisFactory的構(gòu)建，因為這是使用commons-pool的必要工廠類。同時我們可以看到，這里分分片策略使用的確實是Hash，而且還是沖突率很低的MURMUR_HASH。這里不了解commons-pool的可以看一下之前的Commons-pool源碼分析[http://m.itdecent.cn/p/b49452fb3a67]

那么我們直接看ShardedJedisFactory類就好了，因為commons-pool就是基于這個工廠類來管理相關(guān)的對象的，這里緩存的對象是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是創(chuàng)建一個ShardedJedis，同時ShardedJedis也是連接池里保存的對象。
可以看到destroyObject和validateObject都是將ShardedJedis里的redis實例當(dāng)做了一個整體去對待，一個失敗，全部失敗。
接下來看下ShardedJedis的實現(xiàn)，這個里面主要做了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的子類，所以構(gòu)造函數(shù)會一直向上調(diào)用，在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緩存池的初始化和分片的實現(xiàn)，可以看到首先獲取shardInfo就是之前的JedisShardInfo，根據(jù)shardInfo生成多個槽位，將這些槽位存到TreeMap中，同時將shardInfo和Jedis的映射存到resources中。當(dāng)我們做Client的獲取的時候：
首先調(diào)用ShardedJedisPool的getResource方法，從對象池中獲取一個ShardedJedis：

ShardedJedis jedisClient = sharedPool.getResource();

調(diào)用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中判斷，當(dāng)前的key落在哪個區(qū)間上，再通過這個區(qū)間上的ShardInfo從resources的Map中獲取對應(yīng)的Jedis實例。

這也就是說，每一個ShardedJedis都維護(hù)了所有的分片，將多個實例當(dāng)成一個整體去使用，這也就導(dǎo)致，只要集群中一個實例不可用，整個ShardedJedis就不可用了。同時對于hash的分片方式，是不可擴(kuò)容的，擴(kuò)容之后原本應(yīng)該存儲在一起的數(shù)據(jù)就分離了。

其實這種是Jedis默認(rèn)提供的分片方式，其實針對我們自己的場景我們也可以嘗試自己做一個路由機(jī)制，例如根據(jù)不同年份、月份的數(shù)據(jù)落到一個實例上。

上面就是所有的數(shù)據(jù)分片的jedis實現(xiàn)的分析，我們線上的業(yè)務(wù)也是基于ShardedJedis來實現(xiàn)的，由于線上業(yè)務(wù)的QPS不高，量也不是很大，所以運行還算平穩(wěn)。

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

  • 轉(zhuǎn)載:Redis 寶典 | 基礎(chǔ)、高級特性與性能調(diào)優(yōu) 本文由 DevOpsDays 本文由簡書作者kelgon供稿...
    meng_philip123閱讀 3,273評論 1贊 34
  • 1.1 資料 ，最好的入門小冊子，可以先于一切文檔之前看，免費。 作者Antirez的博客，Antirez維護(hù)的R...
    JefferyLcm閱讀 17,323評論 1贊 51
  • 對于日常開發(fā)，Redis由于單線程的并發(fā)模型、豐富的數(shù)據(jù)結(jié)構(gòu)和簡單的API，深受廣大程序員的喜愛。Redis提供了...
    一只小哈閱讀 9,140評論 10贊 26
  • 科技是第一生產(chǎn)力，世界歷史已經(jīng)證明了這樣一個原理。人類世界在我們現(xiàn)在所認(rèn)知的是經(jīng)歷了三次科技革命。 這三次科技革命...
    林沐日記閱讀 1,647評論 0贊 2
  • 雨天，繁忙的市中心，一個垂垂老矣，拄著拐杖的老人，正在攔車， 來來往往的出租車很多，但是司機(jī)或許是怕他在車上有意外...
    燚月仁心閱讀 283評論 0贊 0

友情鏈接更多精彩內(nèi)容