根据Redis官方文档的介绍，学习了一下Redis网络通信协议。然后偶然在GitHub上发现了个用Netty实现的Redis服务器，很有趣，于是就动手实现了一下！
Redis的客户端与服务端采用一种叫做 RESP(REdis Serialization Protocol)的网络通信协议交换数据。RESP的设计权衡了实现简单、解析快速、人类可读这三个因素。Redis客户端通过RESP序列化整数、字符串、数据等数据类型，发送字符串数组表示参数的命令到服务端。服务端根据不同的请求命令响应不同的数据类型。除了管道和订阅外，Redis客户端和服务端都是以这种简单的请求-响应模型通信的。
具体来看，RESP支持五种数据类型。以”*”消息头标识总长度，消息内部还可能有”$”标识字符串长度，每行以\r\n结束：
例如，观察下面命令对应的RESP，这一组set/get也正是我们要在Netty里实现的：
set name helloworld
->
*3\r\n
$3\r\n
set\r\n
$4\r\n
name\r\n
$10\r\n
helloworld\r\n
<-
:1\r\n
get name
->
*2\r\n
$3\r\n
get\r\n
$4\r\n
name\r\n
<-
$10\r\n
helloworld\r\n
set name abc111
->
*3\r\n
$3\r\n
set\r\n
$4\r\n
name\r\n
$6\r\n
abc111\r\n
<-
:0\r\n
get age
->
*2\r\n
$3\r\n
get\r\n
$3\r\n
age\r\n
<-
:-1\r\n下面就用高性能的网络通信框架Netty实现一个简单的Redis服务器后端，解析set和get命令，并保存键值对。
Netty版本，5.0还处于alpha，使用Final版里最新的。但即便是4.0.25.Final竟然也跟4.0的前几个版本有些不同，网上一些例子中用的API根本就找不到了。Netty的API改得有点太“任性”了吧？:)
<dependency>
<groupId>io.nettygroupId>
<artifactId>netty-allartifactId>
<version>4.0.25.Finalversion>
dependency>Netty服务器启动代码，这套代码应该是Netty 4里的标准模板了，具体细节就不在本文赘述了。主要关注我们注册的几个Handler。Netty中Handler分为Inbound和Outbound，RedisCommandDecoder和RedisCommandHandler是Inbound，RedisCommandDecoder是Outbound：
public class Main {
public static void main(String[] args) throws Exception {
new Main().start(6379);
}
public void start(int port) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap()
.group(group)
.channel(NioServerSocketChannel.class)
.localAddress(port)
.childHandler(new ChannelInitializer() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()
.addLast(new RedisCommandDecoder())
.addLast(new RedisReplyEncoder())
.addLast(new RedisCommandHandler());
}
});
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(port).sync();
// Wait until the server socket is closed.
f.channel().closeFuture().sync();
} finally {
// Shutdown the EventLoopGroup, which releases all resources.
group.shutdownGracefully();
}
}
} RedisCommandDecoder开始时cmds是null，进入doDecodeNumOfArgs先解析出命令和参数的个数，并初始化cmds。之后就会进入doDecodeArgs逐一解析命令名和参数了。当最后完成时，会根据解析结果创建出RedisCommand对象，并加入到out列表里。这样下一个handler就能继续处理了。
public class RedisCommandDecoder extends ReplayingDecoder<Void> {
/** Decoded command and arguments */
private byte[][] cmds;
/** Current argument */
private int arg;
/** Decode in block-io style, rather than nio. */
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List因为我们只是简单实现set和get命令，所以只可能有一个参数或两个参数：
public class RedisCommand {
/** Command name */
private final String name;
/** Optional arguments */
private byte[] arg1;
private byte[] arg2;
public RedisCommand(String name, byte[] arg1) {
this.name = name;
this.arg1 = arg1;
}
public RedisCommand(String name, byte[] arg1, byte[] arg2) {
this.name = name;
this.arg1 = arg1;
this.arg2 = arg2;
}
public String getName() {
return name;
}
public byte[] getArg1() {
return arg1;
}
public byte[] getArg2() {
return arg2;
}
@Override
public String toString() {
return "Command{" +
"name='" + name + '\'' +
", arg1=" + Arrays.toString(arg1) +
", arg2=" + Arrays.toString(arg2) +
'}';
}
}RedisCommandHandler拿到RedisCommand后，根据命令名执行命令。这里用一个HashMap模拟数据库了，set就往Map里放，get就从里面取。除了执行具体操作，还要根据执行结果返回不同的Reply对象：
@ChannelHandler.Sharable
public class RedisCommandHandler extends SimpleChannelInboundHandler<RedisCommand> {
private HashMapbyte[]> database = new HashMap byte[]>();
@Override
protected void channelRead0(ChannelHandlerContext ctx, RedisCommand msg) throws Exception {
System.out.println("RedisCommandHandler: " + msg);
if (msg.getName().equalsIgnoreCase("set")) {
if (database.put(new String(msg.getArg1()), msg.getArg2()) == null) {
ctx.writeAndFlush(new IntegerReply(1));
} else {
ctx.writeAndFlush(new IntegerReply(0));
}
}
else if (msg.getName().equalsIgnoreCase("get")) {
byte[] value = database.get(new String(msg.getArg1()));
if (value != null && value.length > 0) {
ctx.writeAndFlush(new BulkReply(value));
} else {
ctx.writeAndFlush(BulkReply.NIL_REPLY);
}
}
}
} RedisReplyEncoder实现比较简单，拿到RedisReply消息后，直接写入到ByteBuf中就可以了。具体的写入方法都在各个RedisReply的具体实现中。
public class RedisReplyEncoder extends MessageToByteEncoder<RedisReply> {
@Override
protected void encode(ChannelHandlerContext ctx, RedisReply msg, ByteBuf out) throws Exception {
System.out.println("RedisReplyEncoder: " + msg);
msg.write(out);
}
}public interface RedisReply<T> {
byte[] CRLF = new byte[] { '\r', '\n' };
T data();
void write(ByteBuf out) throws IOException;
}
public class IntegerReply implements RedisReply<Integer> {
private static final char MARKER = ':';
private final int data;
public IntegerReply(int data) {
this.data = data;
}
@Override
public Integer data() {
return this.data;
}
@Override
public void write(ByteBuf out) throws IOException {
out.writeByte(MARKER);
out.writeBytes(String.valueOf(data).getBytes());
out.writeBytes(CRLF);
}
@Override
public String toString() {
return "IntegerReply{" +
"data=" + data +
'}';
}
}
public class BulkReply implements RedisReply<byte[]> {
public static final BulkReply NIL_REPLY = new BulkReply();
private static final char MARKER = '$';
private final byte[] data;
private final int len;
public BulkReply() {
this.data = null;
this.len = -1;
}
public BulkReply(byte[] data) {
this.data = data;
this.len = data.length;
}
@Override
public byte[] data() {
return this.data;
}
@Override
public void write(ByteBuf out) throws IOException {
// 1.Write header
out.writeByte(MARKER);
out.writeBytes(String.valueOf(len).getBytes());
out.writeBytes(CRLF);
// 2.Write data
if (len > 0) {
out.writeBytes(data);
out.writeBytes(CRLF);
}
}
@Override
public String toString() {
return "BulkReply{" +
"bytes=" + Arrays.toString(data) +
'}';
}
}服务端跑起来后，用官方的redis-cli就能连上我们的服务，执行一些命令测试一下。看到自己实现的Redis“伪服务端”能够“骗过”redis-cli，还是很有成就感的！
127.0.0.1:6379> set name helloworld
(integer) 1
127.0.0.1:6379> get name
"helloworld"
127.0.0.1:6379> set name abc123
(integer) 0
127.0.0.1:6379> get name
"abc123"
127.0.0.1:6379> get age
(nil)因为是初次使用Netty 4，好多网上的资料都是Netty 3或者Netty 4早期版本的，API都不一样了，所以碰到了不少问题，官方文档里也没找到答案，一点点调试、猜测、看源码才摸出点儿“门道”：