Jedis源码分析

Posted 2020-12-02 heapstark

对于日常开发，Redis由于单线程的并发模型、丰富的数据结构和简单的API，深受广大程序员的喜爱。Redis提供了多种语言的API，像java、c和python等。之前一直都是使用redis，但是没有多redis的API有一个系统的认识。忙里偷闲，撸一下Redis相关的API的实现，由于我是一个java猿，那么我主要学习了一下jedis的源码，来分析一下Redis的读写流程。

一、Jedis项目结构

：

 

 

代码是比较简单的，而且很多类也没有那么多的抽象和继承，其实是比较好懂的。commands包里面主要是封装的redis支持的各种命令。
exception包主要是封装了一些redis的exception。
在jedis包下的是一些redis的Client。
jedis的代码结构大致就是上述这些，这里我们就以最简单的jedis类来看一下读写的流程。

二、Jedis继承结构

这里是jedis的UML图：

 

 

public class Jedis extends BinaryJedis implements JedisCommands, MultiKeyCommands,
    AdvancedJedisCommands, ScriptingCommands, BasicCommands, ClusterCommands, SentinelCommands, ModuleCommands {

Jedis 继承了BinaryJedis 同时实现了一系列的Commands接口，BinaryJedis里主要和redis Server进行交互，一系列Commands接口主要是对redis支持的接口进行分类，像BasicCommands主要包含了info、flush等操作，BinaryJedisCommands 主要包含了get、set等操作，MultiKeyBinaryCommands主要包含了一些批量操作的接口例如mset等。

三、用set方法分析Redis的请求流程

由于Jedis实现了各种接口，导致它内部的方法十分的多，这里我们使用一个简单的Demo来学习一下Jedis：

    Jedis jed = new Jedis("locahost",6379);
    jed.set("hello","123");
    String out = jed.get("hello");

首先看Jedis的实例化过程:

public Jedis(final String host, final int port) {  super(host, port);}

public BinaryJedis(final String host, final int port) {  client = new Client(host, port);}

Jedis因为继承了BinaryJedis，大部分的操作都是在BinaryJedis中实现的，在BinaryJedis的构造方法中就实例化了Client。
Client的继承结构如下：

 

 

BinaryJedis中的方法主要是对Client做了代理，Client继承了BinaryClient，BinaryClient继承了Connection，实现了Commands接口。Client主要做了一些编解码的工作，BinaryClient做了Command的发送操作，而所有与redisServer交互的工作由Connection完成。

首先看Set方法：

  /**
   * Set the string value as value of the key. The string can‘t be longer than 1073741824 bytes (1
   * GB).
   * <p>
   * Time complexity: O(1)
   * @param key
   * @param value
   * @return Status code reply
   */
  @Override
  public String set(final String key, String value) {
    checkIsInMultiOrPipeline();
    client.set(key, value);
    return client.getStatusCodeReply();
  }

这里主要委托给Client进行处理。

  @Override
  public void set(final String key, final String value) {
    set(SafeEncoder.encode(key), SafeEncoder.encode(value));
  }
  

这里主要是调用了BinaryClient的set方法。

  public void set(final byte[] key, final byte[] value) {
    sendCommand(Command.SET, key, value);
  }

这里主要是委托了Connection的sendCommand方法。接下来到了关键部分：

  public Connection sendCommand(final ProtocolCommand cmd, final byte[]... args) {
    try {
      connect();
      Protocol.sendCommand(outputStream, cmd, args);
      return this;
    } catch (JedisConnectionException ex) {
      /*
       * When client send request which formed by invalid protocol, Redis send back error message
       * before close connection. We try to read it to provide reason of failure.
       */
      try {
        String errorMessage = Protocol.readErrorLineIfPossible(inputStream);
        if (errorMessage != null && errorMessage.length() > 0) {
          ex = new JedisConnectionException(errorMessage, ex.getCause());
        }
      } catch (Exception e) {
        /*
         * Catch any IOException or JedisConnectionException occurred from InputStream#read and just
         * ignore. This approach is safe because reading error message is optional and connection
         * will eventually be closed.
         */
      }
      // Any other exceptions related to connection?
      broken = true;
      throw ex;
    }
  }

1. 调用connect()方法进行连接：

  public void connect() {
    if (!isConnected()) {
      try {
        socket = new Socket();
        // ->@wjw_add
        socket.setReuseAddress(true);
        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
        // <[email protected]_add

        socket.connect(new InetSocketAddress(host, port), connectionTimeout);
        socket.setSoTimeout(soTimeout);

        if (ssl) {
          if (null == sslSocketFactory) {
            sslSocketFactory = (SSLSocketFactory)SSLSocketFactory.getDefault();
          }
          socket = (SSLSocket) sslSocketFactory.createSocket(socket, host, port, true);
          if (null != sslParameters) {
            ((SSLSocket) socket).setSSLParameters(sslParameters);
          }
          if ((null != hostnameVerifier) &&
              (!hostnameVerifier.verify(host, ((SSLSocket) socket).getSession()))) {
            String message = String.format(
                "The connection to ‘%s‘ failed ssl/tls hostname verification.", host);
            throw new JedisConnectionException(message);
          }
        }

        outputStream = new RedisOutputStream(socket.getOutputStream());
        inputStream = new RedisInputStream(socket.getInputStream());
      } catch (IOException ex) {
        broken = true;
        throw new JedisConnectionException("Failed connecting to host " 
            + host + ":" + port, ex);
      }
    }
  }

这里主要使用Socket通信来实现命令的发送，连接使用长连接来减小建立连接的开销。并实例化了RedisOutputStream和RedisInputStream。在每一次进行query的时候都会调用connect方法来保证之前连接失效之后能新建连接并操作成功。

2. 调用Protocol的sendCommand方法进行发送：

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

这里代码比较清晰，利用了Protocol提供的一些请求头来构造一个请求。这里具体的协议内容就不细解析了，发送完请求之后返回。

之后调用client.getStatusCodeReply();进行返回状态的获取：

  public String getStatusCodeReply() {
    flush();
    final byte[] resp = (byte[]) readProtocolWithCheckingBroken();
    if (null == resp) {
      return null;
    } else {
      return SafeEncoder.encode(resp);
    }
  }

首先调用了flush方法，保证之前的写入能发送出去，之后调用了readProtocolWithCheckingBroken来获取响应。

  protected Object readProtocolWithCheckingBroken() {
    try {
      return Protocol.read(inputStream);
    } catch (JedisConnectionException exc) {
      broken = true;
      throw exc;
    }
  }

调用Protocol.read进行对RedisInputStream进行读取，在这过程中可能会抛出连接异常。

public static Object read(final RedisInputStream is) {  return process(is);}
  private static Object process(final RedisInputStream is) {

    final byte b = is.readByte();
    if (b == PLUS_BYTE) {
      return processStatusCodeReply(is);
    } else if (b == DOLLAR_BYTE) {
      return processBulkReply(is);
    } else if (b == ASTERISK_BYTE) {
      return processMultiBulkReply(is);
    } else if (b == COLON_BYTE) {
      return processInteger(is);
    } else if (b == MINUS_BYTE) {
      processError(is);
      return null;
    } else {
      throw new JedisConnectionException("Unknown reply: " + (char) b);
    }
  }

最后在read的时候对返回的响应进行了判断，枚举出了几种响应方式，对不同的响应进行不同的处理。
这里可以看出，整个交互过程就是一个Socket通信过程。按照一定的协议发送请求，之后读取返回结果。但是这里也有一个问题就是线程安全问题，显然Jedis实例是线程不安全的，对于多线程共享jedis实例是会有问题的。同时直接使用jedis不能避免的需要反复的创建和销毁Socket，开销很大。所以就引出了后面的jedisPool的使用。

四、JedisPool的使用和实现

JedisPool是Jedis提供的一种对Redis的连接池，利用连接池可以很好的对Jedis的连接做一个很好的掌控，能避免创建和销毁的开销，同时可以进行定期的保活，能避免反复的创建连接。
下面是一个JedisPool例子：

    JedisPoolConfig config = new JedisPoolConfig();
    config.setTestOnBorrow(true);
    JedisPool pool = new JedisPool(config, hnp.getHost(), hnp.getPort(), 2000, "foobared");
    Jedis jedis = pool.getResource();
    jedis.set("foo", "bar");
    jedis.close();

可以看到新创建了一个JedisPoolConfig，用于对JedisPool的配置。这里没有使用之前JedisPool的returnResource。因为jedis.close()已经做了相关的returnResource方法。
我们先看一下JedisPoolConfig是什么：

public class JedisPoolConfig extends GenericObjectPoolConfig {
  public JedisPoolConfig() {
    // defaults to make your life with connection pool easier :)
    setTestWhileIdle(true);
    setMinEvictableIdleTimeMillis(60000);
    setTimeBetweenEvictionRunsMillis(30000);
    setNumTestsPerEvictionRun(-1);
  }
}

JedisPoolConfig继承了GenericObjectPoolConfig，GenericObjectPoolConfig是ApacheCommons pool提供的一个对象池的配置。JedisPool使用了ApacheCommons pool来进行连接池的实现。GenericObjectPoolConfig提供了很多的参数，我们可以使用JedisPoolConfig也可以使用GenericObjectPoolConfig。下面列出一些关键的参数：

maxActive：控制一个pool可分配多少个jedis实例，通过pool.getResource()来获取；如果赋值为-1，则表示不限制；如果pool已经分配了maxActive个jedis实例，则此时pool的状态为exhausted。
maxIdle：控制一个pool最多有多少个状态为idle(空闲)的jedis实例；
whenExhaustedAction：表示当pool中的jedis实例都被allocated完时，pool要采取的操作；默认有三种。
WHEN_EXHAUSTED_FAIL --> 表示无jedis实例时，直接抛出NoSuchElementException；
WHEN_EXHAUSTED_BLOCK --> 则表示阻塞住，或者达到maxWait时抛出JedisConnectionException；
WHEN_EXHAUSTED_GROW --> 则表示新建一个jedis实例，也就说设置的maxActive无用；
maxWait：表示当borrow一个jedis实例时，最大的等待时间，如果超过等待时间，则直接抛出JedisConnectionException；
testOnBorrow：在borrow一个jedis实例时，是否提前进行alidate操作；如果为true，则得到的jedis实例均是可用的；
testOnReturn：在return给pool时，是否提前进行validate操作；
testWhileIdle：如果为true，表示有一个idle object evitor线程对idle object进行扫描，如果validate失败，此object会被从pool中drop掉；这一项只有在timeBetweenEvictionRunsMillis大于0时才有意义；
timeBetweenEvictionRunsMillis：表示idle object evitor两次扫描之间要sleep的毫秒数；
numTestsPerEvictionRun：表示idle object evitor每次扫描的最多的对象数；
minEvictableIdleTimeMillis：表示一个对象至少停留在idle状态的最短时间，然后才能被idle object evitor扫描并驱逐；这一项只有在timeBetweenEvictionRunsMillis大于0时才有意义；
softMinEvictableIdleTimeMillis：在minEvictableIdleTimeMillis基础上，加入了至少minIdle个对象已经在pool里面了。如果为-1，evicted不会根据idle time驱逐任何对象。如果minEvictableIdleTimeMillis>0，则此项设置无意义，且只有在timeBetweenEvictionRunsMillis大于0时才有意义；

配置比较多，这里我不打算详细的写Commons Pool的实现机制，只是说说JedisPool是怎么实现的。

JedisPool的实例化过程如下：

  public JedisPool(final GenericObjectPoolConfig poolConfig, final String host, int port,
      int timeout, final String password) {
    this(poolConfig, host, port, timeout, password, Protocol.DEFAULT_DATABASE, null);
  }
    public JedisPool(final GenericObjectPoolConfig poolConfig, final String host, int port,
      int timeout, final String password, final int database, final String clientName) {
    this(poolConfig, host, port, timeout, timeout, password, database, clientName, false,
        null, null, null);
  }
    public JedisPool(final GenericObjectPoolConfig poolConfig, final String host, int port,
      final int connectionTimeout, final int soTimeout, final String password, final int database,
      final String clientName, final boolean ssl, final SSLSocketFactory sslSocketFactory,
      final SSLParameters sslParameters, final HostnameVerifier hostnameVerifier) {
    super(poolConfig, new JedisFactory(host, port, connectionTimeout, soTimeout, password,
        database, clientName, ssl, sslSocketFactory, sslParameters, hostnameVerifier));
  }
 

这里实例化了一个JedisFactory，这个工厂类十分关键，这个工厂类是Commons pool来进行多对象池对象进行管理的一个工厂，对于所有对象的创建、销毁、激活和有效性校验都是在JedisFactory中进行的：

class JedisFactory implements PooledObjectFactory<Jedis> {
  private final AtomicReference<HostAndPort> hostAndPort = new AtomicReference<HostAndPort>();
  private final int connectionTimeout;
  private final int soTimeout;
  private final String password;
  private final int database;
  private final String clientName;
  private final boolean ssl;
  private final SSLSocketFactory sslSocketFactory;
  private SSLParameters sslParameters;
  private HostnameVerifier hostnameVerifier;

  public JedisFactory(final String host, final int port, final int connectionTimeout,
      final int soTimeout, final String password, final int database, final String clientName,
      final boolean ssl, final SSLSocketFactory sslSocketFactory, final SSLParameters sslParameters,
      final HostnameVerifier hostnameVerifier) {
    this.hostAndPort.set(new HostAndPort(host, port));
    this.connectionTimeout = connectionTimeout;
    this.soTimeout = soTimeout;
    this.password = password;
    this.database = database;
    this.clientName = clientName;
    this.ssl = ssl;
    this.sslSocketFactory = sslSocketFactory;
    this.sslParameters = sslParameters;
    this.hostnameVerifier = hostnameVerifier;
  }

JedisFactory实现了PooledObjectFactory接口，PooledObjectFactory是Commons Pool提供的接口。PooledObjectFactory提供了很多的方法：

public interface PooledObjectFactory<T> {
    PooledObject<T> makeObject() throws Exception;

    void destroyObject(PooledObject<T> var1) throws Exception;

    boolean validateObject(PooledObject<T> var1);

    void activateObject(PooledObject<T> var1) throws Exception;

    void passivateObject(PooledObject<T> var1) throws Exception;
}

makeObject为创建对象的方法。
destroyObject为销毁对象的方法。
validateObject为校验对象有消息的方法。
activateObject为激活对象的方法。
passivateObject为钝化对象的方法。

对于对象池对对象的管理使用了PooledObjectFactory中的方法，也算做到了“解耦”，自己的东西自己管，Commons Pool 不侵入任何逻辑。

在创建好了JedisPool之后呢，在使用的时候利用getResource来获取jedis的客户端：

  public Jedis getResource() {
    Jedis jedis = super.getResource();
    jedis.setDataSource(this);
    return jedis;
  }
   public T getResource() {
    try {
      return internalPool.borrowObject();
    } catch (NoSuchElementException nse) {
      throw new JedisException("Could not get a resource from the pool", nse);
    } catch (Exception e) {
      throw new JedisConnectionException("Could not get a resource from the pool", e);
    }
  }

internalPool是一个Commons pool。我们在获取jedis的时候调用了Commons pool的borrowObject。表面的意思就是借一个链接。同时将JedisPool的引用交给jedis，便于在close的时候进行链接的返还：

  @Override
  public void close() {
    if (dataSource != null) {
      if (client.isBroken()) {
        this.dataSource.returnBrokenResource(this);
      } else {
        this.dataSource.returnResource(this);
      }
    } else {
      client.close();
    }
  }

在jedis调用close方法时候，调用dataSource.returnResource进行链接的返还。

这样jedis和JedisPool的实现就分析完了，但是对于Commons Pool对我们还是黑盒的，接下来会写一个对Commons pool的实现原理的笔记。同时呢对于jedis Pool只能进行单实例的链接操作，但是对于数据量大的时候，单实例不能满足需求。这个时候就需要对实例进行“分片”。Jedis也是提供了分片的支持，后面也会总结一个jedis分片的实现。