Redis: Jedis 源代码剖析1-链接建立和收发命令

Jedis作为Redis  Java语言推荐的客户端被广泛使用。让我们一探Jedis源代码究竟。

我们以如下代码来DEBUG观察Jedis源代码：

//创建Redis客户端
Jedis jedis = new Jedis();
//调用set 命令,返回状态标记
String code=jedis.set("s", "s");
System.out.println("code="+code);
//调用get命令
String s =jedis.get("s");
System.out.println("s="+s);第一部分：Jedis对象的创建

Jedis jedis=new Jedis();主要是创建链接Redis服务器的客户端。
在Jedis(BinaryJedis)基类中主要有Connection对象。在创建Jedis对象的时候,其实尚未链接到Redis服务器。

在Connection类中,主要设置了链接Redis所使用socket的参数以及操作socket所使用的工具。

//Jedis客户端链接,使用原始socket进行链接
public class Connection implements Closeable
{

private static final byte[][] EMPTY_ARGS = new byte[0][];
//默认主机
private String host = Protocol.DEFAULT_HOST;
//默认端口
private int port = Protocol.DEFAULT_PORT;
//原始socket
private Socket socket;
//输入输出流
private RedisOutputStream outputStream;
private RedisInputStream inputStream;
private int connectionTimeout = Protocol.DEFAULT_TIMEOUT;
private int soTimeout = Protocol.DEFAULT_TIMEOUT;
private boolean broken = false;

public Connection() {
}
}

由Connection的成员变量中可以看出来。Jedis使用了元素的JDK IO  Socket来处理网络通信的。
到此,Jedis 对象就创建处理啦。

2-Jedis链接Redis服务器过程。

在调用  String code=jedis.set("s", "s"); 命令的时候,才是真正创建链接的过程。

Client(BinaryClient).set(byte[], byte[])   方法参数就是把由String 字符串转换成字节数值。

并调用Client(Connection).sendCommand(ProtocolCommand, byte[]...) 方法来发送Redis命令。

//每次发送命令前都判断是否链接,如果链接端口并且链接不上，则抛出异常
protected Connection sendCommand(final ProtocolCommand cmd, final byte[]... args) {
try {
connect();//每次发送Redis命令都会调用Connect()方法来链接Redis远程服务器
Protocol.sendCommand(outputStream, cmd, args); //操作socket 的输出流来发送命令
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;
}
}每次调用sendCommand发送命令时候，都会调用Connnect()方法尝试链接远程端口。
//在发送命令之前连接redis服务器
public void connect() {
if (!isConnected()) {
try {
//创建新socket
socket = new Socket();
//设置socket参数
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
// <-@wjw_add
//设置链接超时时间
socket.connect(new InetSocketAddress(host, port), connectionTimeout);
//设置读取超时时间
socket.setSoTimeout(soTimeout);
//获取socket原始输入输出流
outputStream = new RedisOutputStream(socket.getOutputStream());
inputStream = new RedisInputStream(socket.getInputStream());
} catch (IOException ex) {
broken = true;
throw new JedisConnectionException(ex);
}
}
}

每次链接到远程Redis服务器后，第一个命令就是发送密钥命令。
@Override
public void connect() {
if (!isConnected()) {
super.connect();
if (password != null) {
auth(password);
getStatusCodeReply();
}
if (db > 0) {
select(Long.valueOf(db).intValue());
getStatusCodeReply();
}
}
}

在每次发送一个命令后,都会去获取返回码。
/** 对于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
*/
@Override
public String set(final String key, String value) {
checkIsInMultiOrPipeline();
client.set(key, value);
return client.getStatusCodeReply();
}


在获取状态码时，每次都去刷新通道。
public String getStatusCodeReply() {
flush();
final byte[] resp = (byte[]) readProtocolWithCheckingBroken();
if (null == resp) {
return null;
} else {
return SafeEncoder.encode(resp);
}
}

读取数据流最终通过SocketInputStream 类来读取。
以上就涉及到Jedis发送Redis和接受Redis命令的过程。