目录

• 前言
• 版本V1：基础版
• 版本V2：缓冲区、命令封装
• 版本V3：支持多种Redis命令
• 版本V4：异常处理、断线重连、连接池
• 总结

前言

Jedis是Redis官方推荐的Jedis客户端，在业界使用非常广泛。通过阅读Jedis源码，可以掌握如何设计和实现一个轻量级、带连接池的Redis客户端。我常常说，学习源码最好的方法就是以自己的思考重新写一遍，这样有助于把学到的知识固化成实际经验。本文会带着你从零开始设计一个极简版的Jedis，从支持最简单的GET/SET命令出发，逐步添加缓冲区、多种命令、断线重连、连接池等多种功能。本文的源代码放在Github上（mini-jedis），感兴趣的读者可以自行取阅。

版本V1：基础版

首先实现一个最简单的GET命令。我们知道Jedis的IO模式是BIO，它发送和接收数据依赖输入输出流的阻塞式读写。所以我们先为Jedis创建一个socket：

	// Jedis.javapublic Jedis(String host, int port) throws UnknownHostException, IOException {socket = new Socket();socket.setReuseAddress(true);socket.setKeepAlive(true); // Will monitor the TCP connection is validsocket.setTcpNoDelay(true); // Socket buffer Whetherclosed, to ensure timely delivery of datasocket.setSoLinger(true, 0); // Control calls close () method, the underlying socket is closed immediatelysocket.connect(new InetSocketAddress(host, port), 2000);socket.setSoTimeout(2000);}

注意socket有多项TCP参数要设置（参考Jedis源码），设置后的效果如下：

1. REUSE_ADDRESS：Jedis客户端断开连接处于TIMED_WAIT状态时，重连可以复用之前的端口。
2. SO_KEEPALIVE：当长时间没有消息收发时，也能保证连接不断开。
3. TCP_NODELAY：禁用nagle算法，保证发送数据的实时性。
4. SO_LINGER：当调用close()方法时，连接立刻关闭，而不是通常那样等待缓冲区中的剩余数据发送完。

建立好socket后，我们再来编写与Redis服务器通信的逻辑。在写之前，我们有必要了解一下Redis的通信协议：RESP协议（Redis Serialization Protocol）。

RESP协议是一种简单的应用层协议，它的特点是：实现精简、容易阅读。以GET命令为例，我们来看一个最简单的RESP协议例子：

// GET消息：发送
*2
$3
GET
$4
key1// GET消息：接收
$6
value1

以上消息中有一些特定符号，它们的意义总结如下：

根据这里的符号定义，可以解释上面的GET消息：

1. 发送：由包含2个元素的数组构成，第一个元素是长度为3的字符串”GET“，第二个元素是长度为4的字符串”key1“。这两个元素分别是命令名字和命令参数。
2. 接收：仅包含一个长度为6的字符串”value1“。

清楚了GET消息的协议构成，很容易写出它的实现代码：

	// Jedis.javapublic String get(String key) throws IOException {StringBuilder sb = new StringBuilder();sb.append("*2").append("\r\n").append("$3").append("\r\n").append("GET").append("\r\n").append("$").append(key.length()).append("\r\n").append(key).append("\r\n");socket.getOutputStream().write(sb.toString().getBytes());InputStream is = socket.getInputStream();byte[] b = new byte[1024];int len = is.read(b);String ret = new String(b, 0, len);return ret;}

同样地，我们编写SET消息的实现方法，并写好它们的测试例子：

// Jedis.java
public static void main(String args[]) throws UnknownHostException, IOException {Jedis jedis = new Jedis("localhost", 6379);System.out.print(jedis.set("key1", "value1"));System.out.print(jedis.get("key1"));
}

这样就得到了包含最简单两个命令GET/SET的基础版代码。

版本V2：缓冲区、命令封装

V1版本的实现虽然能跑得通，但是它有两个问题：

1. 性能不佳。问题出在socket.getOutputStream().write(sb.toString().getBytes());，短短一行代码包含三次字节数组的拷贝：toString()第一次，getBytes()第二次，write()第三次。对于追求性能的我们这种情况肯定不能忍，必须找到一种方法降低字节数组的拷贝次数。
2. 对复杂情况的处理不当。在前面的代码处理中，我们分配了一个1024字节的数组来从输入流读取数据。在通常的情况下，这个长度是够用的。但是在特殊情况下，返回的数据长度可能大于1024，这时我们需要多次读取数据并拼接起来。

解决上述问题的方法就是使用缓冲区来处理数据的读写。有了缓冲区，我们可以在恰当的时候直接将数据拷贝到socket的输出流中，仅需一次数据拷贝，而不会再有StringBuilder -> String -> byte array -> socket输出流的三次拷贝过程。同时，当从输入流读取较长数据时，我们可以使用缓冲区多次读取，再将每次读取结果加以拼接。设计缓冲区，也是许多网络通信底层应用提升性能的一种基本思路。

先来看输出缓冲区的设计。我们设计一个缓冲区的包装类RedisOutputStream，它同时还包含从socket中获取的输出流：

// RedisOutputStream.java
public class RedisOutputStream extends FilterOutputStream {protected final byte[] buf;// 下一个写入的位置protected int count;private final static int BUFFER_SIZE = 8192;

上面代码显示：我们分配了一个大小为8192字节的缓冲区，并用变量count监控下一个写入的位置。为了演示缓冲区的用法，我们以两个方法write(byte b)和flush()为例。前者用来向缓冲区写入一个字节，当缓冲区已满时，调用后者把当前缓冲区中的数据写入socket输出流：

// RedisOutputStream.javapublic void write(final byte b) throws IOException {if (count == buf.length) {flushBuffer();}buf[count++] = b;}public void flushBuffer() throws IOException {if (count > 0) {out.write(buf, 0, count);count = 0;}}

输入缓冲区的设计与输出缓冲区类似，不同点是有两个监控变量：count代表当前读到的位置，limit代表缓冲区中有用数据的边界。读者要是熟悉Java NIO的Buffer类的API，对比里面的类似概念，会更容易理解。

// RedisInputStream.java
public class RedisInputStream extends FilterInputStream {protected final byte[] buf;// count: 当前读到的位置// limit: 缓冲区有用数据的边界protected int count, limit;private static final int BUFFER_SIZE = 8192;

同样我们用两个方法来演示输入缓冲区的用法。readByte()用来从缓冲区中读取一个字节，读取前会先调用ensureFill()方法。这个方法用来确保缓冲区中有可读的数据，否则会从socket输入流中读取一次数据到缓冲区，并重置count和limit两个字段。

// RedisInputStream.javapublic byte readByte() throws IOException {ensureFill();return buf[count++];}private void ensureFill() throws IOException {if (count >= limit) {limit = in.read(buf);count = 0;if (limit == -1) {throw new RuntimeException("Unexpected end of stream.");}}}

完成了缓冲区代码的编写。接下来另一件重要的事情是优化命令处理的代码，加以封装，增加可读性。之前的代码几近于裸写，输出输入数据的逻辑都糅合在一起，当支持的命令增多时，代码可读性肯定会很低。

还是以GET命令为例。我们先创建CommandArguments类用于封装命令的名字及参数，再将发送命令的逻辑放到sendCommand(CommandArguments args)方法中，将处理接收数据的逻辑放到processReply()方法中。注意sendCommand方法调用后，会马上调用flushBuffer()将当前缓冲区中的数据立即写入到socket输出流中。

// Jedis.javapublic String get(String key) throws IOException {CommandArguments args = new CommandArguments();args.add(Command.GET.name());args.add(key);sendCommand(args);os.flushBuffer();String resp = processReply();return resp;}

从下面的实现细节可以看出，原来发送数据用到的StringBuilder不复存在，所有数据都是直接写入类型为RedisOutputStream的os。processReply() 方法中处理了输入数据以+开头和以$开头的两种情况。

// Jedis.javapublic void sendCommand(CommandArguments args) throws IOException {os.write(Protocol.ASTERISK);os.writeIntCrLf(args.size());for (String arg : args) {os.write(Protocol.DOLLAR);byte[] argBytes = arg.getBytes();os.writeIntCrLf(argBytes.length);os.write(argBytes);os.writeCrLf();}}public String processReply() throws IOException {final byte b = is.readByte();switch (b) {case Protocol.PLUS: // 以+开头return processStatusCodeReply();case Protocol.DOLLAR: // 以$开头return processBulkReply();default:throw new RuntimeException("can't parse reply");}}private String processStatusCodeReply() throws IOException {return is.readLine();}

其他实现细节在此不再赘述。至此，我们在版本V2中使用缓冲区优化了读写性能，并将命令处理的代码适当封装，进一步提升了可读性。

版本V3：支持多种Redis命令

在前面的版本我们支持了GET/SET两种命令。实际上Redis支持的命令成百上千，都支持显然是不现实的。但我们可以选择几个有代表性的命令，力求涵盖RESP命令中出现的所有数据类型。为此我们增加RPUSH和LRANGE两个命令，这里出现的新数据类型在于：RPUSH返回的是整数，而LRANGE返回的是多个字符串组成的数组。

扩展processReply()方法，使得它能处理整数和数组类型的返回：

// Jedis.javapublic Object processReply() throws IOException {final byte b = is.readByte();switch (b) {case Protocol.PLUS:return processStatusCodeReply();case Protocol.DOLLAR:return processBulkReply();case Protocol.COLON:return processInteger();case Protocol.ASTERISK:return processMultiBulkReply();default:throw new RuntimeException("can't parse reply");}}

在以下的实现细节中，可以看到：processMultiBulkReply()方法会根据读取到的数组元素个数，递归调用processReply()。注意这里的返回类型是List