Redis java客户端 jedis 源码分析系列二：单实例 jedis

Posted 2020-10-23

在使用Jedis的过程中最简单的用法就是单实例单连接的jedis，如下代码所示：

public void testJedis(){

   Jedis jedis = new Jedis("127.0.0.1");

   jedis.set("key", "value");

   jedis.get("key");

   jedis.close();

   

}

让我们深入到内部去看一看其结构，如下图所示：

此处请先忽略 JedisPool 类和 Pool<Jedis> 两个类。

Jedis 通过继承 BinaryJedis 来持有Client对象，Client类继承BinaryClient，BinaryClient继承Connection，下面分别介绍各个类主要做什么事情。

Connection：主要用于与redis服务器建立Socket连接，并管理连接

Protocol：主要用于redis客户端通信协议的处理

BinaryClient：是对Connection的包装，将字节数据发送Connection中进行通信协议的编码，同时实现了Redis字节命令的相关接口

Client：是对BinaryClient的包装，主要用于处理字符数据到字节数据的转换工作，然后通过BinaryClient的接口来实现后续与Redis服务器的交互，同时实现了Redis字符命令的相关接口

BinaryJedis：持有Client对象，Transaction事务对象，Pipeline管道对象，其中与Client是组成关系，强约束，其它两个是分别在使用事务和管道时才会使用到。通过持有的Client来执行相关命令的，此类处理字节命令

Jedis：继承BinaryJedis，与Client类似，Jedis处理字符命令，BinaryJedis处理字节命令。

下面我跟随前面的示例代码来进行源码的跟踪：

jedis.set("key", "value");

/**

* 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

*/

public String set(final String key, String value) {

 checkIsInMultiOrPipeline();

 client.set(key, value);

 return client.getStatusCodeReply();

}

很明显，Jedis的set方法直接转换成了client.set()方法。checkIsInMultiOrPipeline();这个方法是检查当前是否有事务或管道在执行。我们再看client的set方法，源码如下

public void set(final String key, final String value) {

 set(SafeEncoder.encode(key), SafeEncoder.encode(value));

}

在Client类中将字符转换成了字节，然后调用BinaryClient的set方法，再看BinaryClient的set方法，源码如下

public void set(final byte[] key, final byte[] value) {

 sendCommand(Command.SET, key, value);

}

在BinaryClient类中直接调用了Connection中的sendCommand方法来完成命令的发送，接着看Connection中的sendCommand方法，源码如下

protected Connection sendCommand(final Command cmd, final byte[]... args) {

 try {

   connect();

   Protocol.sendCommand(outputStream, cmd, args);

   pipelinedCommands++;

   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;

 }

}

到此，干实事的地方到了，前面都只是进行简单的包装。该方法首先建立连接，然后通过Protocol工具类来完成Redis通信协议报文的组装，并写入到 参数outputStream 指定的输出流。catch子句是在出错时获取redis返回的错误并包装到异常信息中，以便外面可以知道出现了什么错误。

接下来看一下Connection的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(ex);

   }

 }

}

该方法首先检查是否已有连接，若有不做任何事情直接返回，若没有则创建socket连接，如果是ssl连接，则对封装成ssl socket连接，最后将socket的输入输出流包装成redis的输入输出流并保存在成员变量中，其中outputStream成员变量在接下来的Protocol.sendCommand方法中作为参数传入，源代码如下：

public static void sendCommand(final RedisOutputStream os, final Command command,

   final byte[]... args) {

 sendCommand(os, command.raw, 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的公共方法sendCommand直接调用了私有的sendCommand方法，私有sendCommand方法就是按照通信协议组装命令，Redis的通信协议格式如下：

*<参数数量>CRLF

$<参数1字节数量>CRLF

<参数1>CRLF

...

$<参数N字节数量>CRLF

<参数N>CRLF

例如：set hello world 命令转换成通信协议如下：

*3

$5

hello

$5

world

至此，完成了整个发送过程，除了flush之外，到目前为止尚未真正发送到redis服务端，我们重新回到Jedis的set方法看看后面的语句：

public String set(final String key, String value) {

 checkIsInMultiOrPipeline();

 client.set(key, value);

 return client.getStatusCodeReply();

}

接下来调用 了client的getStatusCodeReply方法获取返回结果，我们跟进去看看

public String getStatusCodeReply() {

 flush();

 pipelinedCommands--;

 final byte[] resp = (byte[]) readProtocolWithCheckingBroken();

 if (null == resp) {

   return null;

 } else {

   return SafeEncoder.encode(resp);

 }

}

getStatusCodeReply其实是Connection中的方法，方法的作用就是将outputStream中的命令真正发送到redis服务端，然后从inputStream中读取返回结果，先看flush方法：

protected void flush() {

 try {

   outputStream.flush();

 } catch (IOException ex) {

   broken = true;

   throw new JedisConnectionException(ex);

 }

}

很简单，直接调用输出流的flush方法，将输出流中缓存的命令发送到服务端再看读取返回值的readProtocolWithCheckingBroken方法：

protected Object readProtocolWithCheckingBroken() {

 try {

   return Protocol.read(inputStream);

 } catch (JedisConnectionException exc) {

   broken = true;

   throw exc;

 }

}

通过协议处理工具类Protocol从输入流中读取结果，继续跟进到Protocol类的read方法

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

 }

}

Redis的返回结果分为如下5种类型：

• 状态回复：输入流第一个字节为“+”

• 错误回复：第一个字节为“-”

• 整数回复：第一个字节为“:”

• 字符串回复：第一个字节为“$”

• 多条字符串回复：第一个字节为“*”

这些方法就是解析相应的回复，在此不再分析，都只是按照通信协议解析出结果。

至此，单实例jedis的访问流程解析完成