标签：list factory ret round encode 12px optional actor line
在使用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
// <-@wjw_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的访问流程解析完成
Redis java客户端 jedis 源码分析系列二：单实例 jedis
标签：list factory ret round encode 12px optional actor line
原文地址：http://blog.51cto.com/redleafstudios/2068307
