Jedis 客户端及ShardedJedis 源码学习
Jedis 客户端及ShardedJedis 源码学习
陆大侠 发表于2年前
Jedis 客户端及ShardedJedis 源码学习
  • 发表于 2年前
  • 阅读 39
  • 收藏 0
  • 点赞 0
  • 评论 0

腾讯云 技术升级10大核心产品年终让利>>>   

摘要: Jedis源码解析

Jedis 源码比较简洁,只依赖了apache的commons-pool2 这个库。(当然还有JUnit)。

分析源码的最好方式,就是去github下载一份,然后用idea或则eclipse打开,调试单元测试代码。(其它开源项目均可以这样做)。 

redis.clients.jedis.Jedis 这个类实现了redis.clients.jedis.commands.Commands以及BinaryScriptingCommands鞥接口(包含几乎所有Jedis命令以及lua脚本操作),并且它有用一个redis.clients.jedis.Client这个类的实例。

Client及其基类也实现Commands等接口。Jedis是个代理。Client基类的基类Connection中,持有Socket对象,Socket关联的RedisOutputStream以及RedisInputStream。这两个流包装了Socket的output和input流。   redis.clients.jedis.Protocol协助Connection实现redis协议的解析。

整个通讯过程是同步的,Redis对象是非线程安全的。   协议很简单,可以参考这篇博客。

http://www.cnblogs.com/smark/p/3247620.html

看下最内部的Connection对象的主要成员吧:

public class Connection implements Closeable {

  private static final byte[][] EMPTY_ARGS = new byte[0][];

  private String host = Protocol.DEFAULT_HOST;//默认主机都有,值是localhost
  private int port = Protocol.DEFAULT_PORT; //6379
  private Socket socket;                    //这就是用于通讯的Socket了
  private RedisOutputStream outputStream;   //发送命令的output流
  private RedisInputStream inputStream;     //接受结果的input流
  private int connectionTimeout = Protocol.DEFAULT_TIMEOUT;   //tcp连接建立时候的超时设置
  private int soTimeout = Protocol.DEFAULT_TIMEOUT;         //tcp操作超时时间
  private boolean broken = false;          //发生异常时,这个设置为true。外部就知道如何处理,比如连接池就用到这个。

再看看关键方法:

  public void connect() { //方法很简单,没什么好说的。
    if (!isConnected()) {
      try {
        socket = new Socket();
        // ->@wjw_add
        socket.setReuseAddress(true); //允许tcp重复绑定
        socket.setKeepAlive(true); // Will monitor the TCP connection is
        // valid
        socket.setTcpNoDelay(true); // Socket buffer Whetherclosed, to
        // ensure timely delivery of data
        socket.setSoLinger(true, 0); // Control calls close () method,
        // the underlying socket is closed
        // immediately
        // <-@wjw_add

        socket.connect(new InetSocketAddress(host, port), connectionTimeout);
        socket.setSoTimeout(soTimeout);
        outputStream = new RedisOutputStream(socket.getOutputStream());
        inputStream = new RedisInputStream(socket.getInputStream());
      } catch (IOException ex) {
        broken = true;
        throw new JedisConnectionException(ex);
      }
    }
  }
  
    protected Connection sendCommand(final ProtocolCommand cmd, final byte[]... args) {
    try {
      connect();
      Protocol.sendCommand(outputStream, cmd, args);
      return this;
    } catch (JedisConnectionException ex) {
      。。。省略
      }
      broken = true;
      throw ex;
    }
  }

最好的操作到了Protocal方法:

  public static void sendCommand(final RedisOutputStream os, final ProtocolCommand command,
      final byte[]... args) {
    sendCommand(os, command.getRaw(), args);
  }

  private static void sendCommand(final RedisOutputStream os, final byte[] command,
      final byte[]... args) {
    try {
      os.write(ASTERISK_BYTE);
      os.writeIntCrLf(args.length + 1);
      os.write(DOLLAR_BYTE);
      os.writeIntCrLf(command.length);
      os.write(command);
      os.writeCrLf();

      for (final byte[] arg : args) {
        os.write(DOLLAR_BYTE);
        os.writeIntCrLf(arg.length);
        os.write(arg);
        os.writeCrLf();
      }
    } catch (IOException e) {
      throw new JedisConnectionException(e);
    }
  }

然后Redis这边同步的调用了Client的getBinaryBulkReply等等方法,从input流里面,获取结果,和以上类似。

先看类图的:redis.clients.util.Pool持有一个GenericObjectPool<T>的对象,实现了连接池。

再来看看ShardedJedis, 它层次如下:

public class ShardedJedis extends BinaryShardedJedis implements JedisCommands, Closeable {

public class BinaryShardedJedis extends Sharded<Jedis, JedisShardInfo> implements BinaryJedisCommands

重要的方法都封装了在基类Sharded类中,BinaryShardedJedis封装的是基于byte[]的接口,ShardedJedis则封装了String接口。

public class Sharded<R, S extends ShardInfo<R>> {
  public static final int DEFAULT_WEIGHT = 1;//默认每个redis的权重
  private TreeMap<Long, S> nodes;//红黑树纪录hash值到redisInfo的Map
  private final Hashing algo;//hash算法,默认MurMurHash
  //纪录了nodes的item的值ShardInfo<Redis>,到Redis连接对象的Map。
  private final Map<ShardInfo<R>, R> resources = new LinkedHashMap<ShardInfo<R>, R>();

ShardedJedis以及BinaryShardedJedis回调用Sharded的getShard方法,返回一个Redis对象,然后操作这个对象。

在看Sharded的主要方法:1 构造函数。2 getShard方法。实现了一致性hash算法。

  public Sharded(List<S> shards, Hashing algo, Pattern tagPattern) {
    this.algo = algo; //hash算法
    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);
      //为每个shardInfo在TreeMap中创建160个hashCode到shardInfo的映射。
      //如果有name就用else里面的key做hash,没有就是上面的"SHARD-" + i + "-NODE-" + n作为键。
      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);
      }
      //最后根据shardInfo创建Redis对象,并使用shardInfo作为Map的键。
      resources.put(shardInfo, shardInfo.createResource());
    }
  }
 子类就是根据这个方法得到正确的Jedis实例。
 public R getShard(byte[] key) {
    return resources.get(getShardInfo(key));
 }
 如果key是String 类型,有另外一个重载方法
 public S getShardInfo(String key) {
    return getShardInfo(SafeEncoder.encode(getKeyTag(key)));
 }
 Sharded最核心的代码,哈哈,一共5行。 利用了红黑树TreeMap。
 public S getShardInfo(byte[] key) {
    SortedMap<Long, S> tail = nodes.tailMap(algo.hash(key));//这个根据key的hash值,找到所有比它大的SortedMap,这个map的S为Redis对象。
    if (tail.isEmpty()) { //如果没有找到,就取nodes的第一个ShardInfo。
      return nodes.get(nodes.firstKey());
    }
    return tail.get(tail.firstKey()); //如果找到了tail,取tail第一个ShardInfo。
  }





共有 人打赏支持
粉丝 3
博文 52
码字总数 18787
×
陆大侠
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: